Add Database and ObjectStorage Health Checks

- Delays deploying DSPA Manifests until DB and ObjStore
  Connections are healthy
- Introduces status condition of "ComponentDeploymentNotFound" for
  Required Components with missing Deployments

Author: gmfrasca

controllers/config/defaults.go

@@ -69,6 +69,8 @@ const (
 
 // DSPA Status Condition Types
 const (
+	DatabaseAvailable      = "DatabaseAvailable"
+	ObjectStoreAvailable   = "ObjectStoreAvailable"
 	APIServerReady         = "APIServerReady"
 	PersistenceAgentReady  = "PersistenceAgentReady"
 	ScheduledWorkflowReady = "ScheduledWorkflowReady"
@@ -81,9 +83,10 @@ const (
 // kubectl get output, and in summarizing
 // occurrences of causes
 const (
-	MinimumReplicasAvailable = "MinimumReplicasAvailable"
-	FailingToDeploy          = "FailingToDeploy"
-	Deploying                = "Deploying"
+	MinimumReplicasAvailable    = "MinimumReplicasAvailable"
+	FailingToDeploy             = "FailingToDeploy"
+	Deploying                   = "Deploying"
+	ComponentDeploymentNotFound = "ComponentDeploymentNotFound"
 )
 
 // Any required Configmap paths can be added here,

controllers/database.go

@@ -17,6 +17,11 @@ package controllers
 
 import (
 	"context"
+	"database/sql"
+	b64 "encoding/base64"
+	"fmt"
+
+	_ "github.com/go-sql-driver/mysql"
 
 	dspav1alpha1 "github.com/opendatahub-io/data-science-pipelines-operator/api/v1alpha1"
 )
@@ -31,6 +36,48 @@ var dbTemplates = []string{
 	dbSecret,
 }
 
+func (r *DSPAReconciler) VerifyMySQLDBConnection(host, port, username, password, dbname string) bool {
+	connectionString := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", username, password, host, port, dbname)
+	db, err := sql.Open("mysql", connectionString)
+	if err != nil {
+		return false
+	}
+	defer db.Close()
+
+	testStatement := "SELECT 1;"
+	_, err = db.Exec(testStatement)
+	return err == nil
+}
+
+func (r *DSPAReconciler) isDatabaseAccessible(ctx context.Context, dsp *dspav1alpha1.DataSciencePipelinesApplication,
+	params *DSPAParams) bool {
+	log := r.Log.WithValues("namespace", dsp.Namespace).WithValues("dspa_name", dsp.Name)
+
+	log.Info("Performing Database Health Check")
+	databaseSpecified := dsp.Spec.Database != nil
+	usingExternalDB := params.UsingExternalDB(dsp)
+	usingMariaDB := !databaseSpecified || dsp.Spec.Database.MariaDB != nil
+	if usingMariaDB || usingExternalDB {
+		decodePass, _ := b64.StdEncoding.DecodeString(params.DBConnection.Password)
+		db_connect := r.VerifyMySQLDBConnection(params.DBConnection.Host,
+			params.DBConnection.Port,
+			params.DBConnection.Username,
+			string(decodePass),
+			params.DBConnection.DBName)
+		if db_connect {
+			log.Info("Database Health Check Successful")
+		} else {
+			log.Info("Unable to connect to Database")
+		}
+		return db_connect
+
+	}
+
+	log.Info(fmt.Sprintf("Could not connect to Database: Unsupported Type"))
+	// Only MariaDB and Mysql-Compliant Database supported.
+	return false
+}
+
 func (r *DSPAReconciler) ReconcileDatabase(ctx context.Context, dsp *dspav1alpha1.DataSciencePipelinesApplication,
 	params *DSPAParams) error {
 

controllers/dspipeline_controller.go

@@ -19,6 +19,7 @@ package controllers
 import (
 	"context"
 	"fmt"
+
 	"github.com/go-logr/logr"
 	mf "github.com/manifestival/manifestival"
 	dspav1alpha1 "github.com/opendatahub-io/data-science-pipelines-operator/api/v1alpha1"
@@ -217,34 +218,42 @@ func (r *DSPAReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		return ctrl.Result{}, err
 	}
 
-	err = r.ReconcileCommon(dspa, params)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
+	// Get Prereq Status (DB and ObjStore Ready)
+	dbAvailable := r.isDatabaseAccessible(ctx, dspa, params)
+	objStoreAvailable := r.isObjectStorageAccessible(ctx, dspa, params)
+	dspa_prereqs_ready := (dbAvailable && objStoreAvailable)
 
-	err = r.ReconcileAPIServer(ctx, dspa, params)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
+	if dspa_prereqs_ready {
+		// Manage Common Manifests
+		err = r.ReconcileCommon(dspa, params)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
 
-	err = r.ReconcilePersistenceAgent(dspa, params)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
+		err = r.ReconcileAPIServer(ctx, dspa, params)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
 
-	err = r.ReconcileScheduledWorkflow(dspa, params)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
+		err = r.ReconcilePersistenceAgent(dspa, params)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
 
-	err = r.ReconcileUI(dspa, params)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
+		err = r.ReconcileScheduledWorkflow(dspa, params)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
 
-	err = r.ReconcileMLMD(dspa, params)
-	if err != nil {
-		return ctrl.Result{}, err
+		err = r.ReconcileUI(dspa, params)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+
+		err = r.ReconcileMLMD(dspa, params)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
 	}
 
 	log.Info("Updating CR status")
@@ -255,7 +264,7 @@ func (r *DSPAReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		return ctrl.Result{}, err
 	}
 
-	conditions, err := r.GenerateStatus(ctx, dspa)
+	conditions, err := r.GenerateStatus(ctx, dspa, params)
 	if err != nil {
 		log.Info(err.Error())
 		return ctrl.Result{}, err
@@ -271,24 +280,20 @@ func (r *DSPAReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 
 	r.PublishMetrics(
 		dspa,
+		util.GetConditionByType(config.DatabaseAvailable, conditions),
+		util.GetConditionByType(config.ObjectStoreAvailable, conditions),
 		util.GetConditionByType(config.APIServerReady, conditions),
 		util.GetConditionByType(config.PersistenceAgentReady, conditions),
 		util.GetConditionByType(config.ScheduledWorkflowReady, conditions),
 		util.GetConditionByType(config.CrReady, conditions),
 	)
-
 	return ctrl.Result{}, nil
 }
 
 // handleReadyCondition evaluates if condition with "name" is in condition of type "conditionType".
 // this procedure is valid only for conditions with bool status type, for conditions of non bool type
 // results are undefined.
-func (r *DSPAReconciler) handleReadyCondition(
-	ctx context.Context,
-	dspa *dspav1alpha1.DataSciencePipelinesApplication,
-	name string,
-	condition string,
-) (metav1.Condition, error) {
+func (r *DSPAReconciler) handleReadyCondition(ctx context.Context, dspa *dspav1alpha1.DataSciencePipelinesApplication, name string, condition string) (metav1.Condition, error) {
 	readyCondition := r.buildCondition(condition, dspa, config.MinimumReplicasAvailable)
 	deployment := &appsv1.Deployment{}
 
@@ -297,7 +302,14 @@ func (r *DSPAReconciler) handleReadyCondition(
 
 	err := r.Get(ctx, types.NamespacedName{Name: component, Namespace: dspa.Namespace}, deployment)
 	if err != nil {
-		return metav1.Condition{}, err
+		if apierrs.IsNotFound(err) {
+			readyCondition.Reason = config.ComponentDeploymentNotFound
+			readyCondition.Status = metav1.ConditionFalse
+			readyCondition.Message = fmt.Sprintf("Deployment for component \"%s\" is missing", component)
+			return readyCondition, nil
+		} else {
+			return metav1.Condition{}, err
+		}
 	}
 
 	// First check if deployment is scaled down, if it is, component is deemed not ready
@@ -397,7 +409,16 @@ func (r *DSPAReconciler) handleReadyCondition(
 
 }
 
-func (r *DSPAReconciler) GenerateStatus(ctx context.Context, dspa *dspav1alpha1.DataSciencePipelinesApplication) ([]metav1.Condition, error) {
+func (r *DSPAReconciler) GenerateStatus(ctx context.Context, dspa *dspav1alpha1.DataSciencePipelinesApplication, params *DSPAParams) ([]metav1.Condition, error) {
+	databaseAvailable := r.buildCondition(config.DatabaseAvailable, dspa, config.DatabaseAvailable)
+	if r.isDatabaseAccessible(ctx, dspa, params) {
+		databaseAvailable.Status = metav1.ConditionTrue
+	}
+
+	objStoreAvailable := r.buildCondition(config.ObjectStoreAvailable, dspa, config.ObjectStoreAvailable)
+	if r.isObjectStorageAccessible(ctx, dspa, params) {
+		objStoreAvailable.Status = metav1.ConditionTrue
+	}
 
 	apiServerReady, err := r.handleReadyCondition(ctx, dspa, "ds-pipeline", config.APIServerReady)
 	if err != nil {
@@ -411,7 +432,10 @@ func (r *DSPAReconciler) GenerateStatus(ctx context.Context, dspa *dspav1alpha1.
 	if err != nil {
 		return []metav1.Condition{}, err
 	}
+
 	var conditions []metav1.Condition
+	conditions = append(conditions, databaseAvailable)
+	conditions = append(conditions, objStoreAvailable)
 	conditions = append(conditions, apiServerReady)
 	conditions = append(conditions, persistenceAgentReady)
 	conditions = append(conditions, scheduledWorkflowReady)
@@ -420,7 +444,7 @@ func (r *DSPAReconciler) GenerateStatus(ctx context.Context, dspa *dspav1alpha1.
 	crReady := r.buildCondition(config.CrReady, dspa, config.MinimumReplicasAvailable)
 	crReady.Type = config.CrReady
 
-	componentConditions := []metav1.Condition{apiServerReady, persistenceAgentReady, scheduledWorkflowReady}
+	componentConditions := []metav1.Condition{databaseAvailable, objStoreAvailable, apiServerReady, persistenceAgentReady, scheduledWorkflowReady}
 	allReady := true
 	failureMessages := ""
 	for _, c := range componentConditions {
@@ -449,16 +473,24 @@ func (r *DSPAReconciler) GenerateStatus(ctx context.Context, dspa *dspav1alpha1.
 }
 
 func (r *DSPAReconciler) PublishMetrics(dspa *dspav1alpha1.DataSciencePipelinesApplication,
-	apiServerReady, persistenceAgentReady, scheduledWorkflowReady,
+	dbAvailable, objStoreAvailable, apiServerReady, persistenceAgentReady, scheduledWorkflowReady,
 	crReady metav1.Condition) {
 	log := r.Log.WithValues("namespace", dspa.Namespace).WithValues("dspa_name", dspa.Name)
 	log.Info("Publishing Ready Metrics")
-	if apiServerReady.Status == metav1.ConditionTrue {
-		log.Info("APIServer Ready")
-		APIServerReadyMetric.WithLabelValues(dspa.Name, dspa.Namespace).Set(1)
+	if dbAvailable.Status == metav1.ConditionTrue {
+		log.Info("Database Accessible")
+		DBAvailableMetric.WithLabelValues(dspa.Name, dspa.Namespace).Set(1)
+	} else {
+		log.Info("Database Not Yet Accessible")
+		DBAvailableMetric.WithLabelValues(dspa.Name, dspa.Namespace).Set(0)
+	}
+
+	if objStoreAvailable.Status == metav1.ConditionTrue {
+		log.Info("Object Store Accessible")
+		ObjectStoreAvailableMetric.WithLabelValues(dspa.Name, dspa.Namespace).Set(1)
 	} else {
-		log.Info("APIServer Not Ready")
-		APIServerReadyMetric.WithLabelValues(dspa.Name, dspa.Namespace).Set(0)
+		log.Info("Object Store Not Yet Accessible")
+		ObjectStoreAvailableMetric.WithLabelValues(dspa.Name, dspa.Namespace).Set(0)
 	}
 
 	if persistenceAgentReady.Status == metav1.ConditionTrue {

controllers/metrics.go

@@ -23,6 +23,26 @@ import (
 
 // Prometheus metrics gauges
 var (
+	DBAvailableMetric = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "data_science_pipelines_application_database_available",
+			Help: "Data Science Pipelines Application - Database Availability Status",
+		},
+		[]string{
+			"dspa_name",
+			"dspa_namespace",
+		},
+	)
+	ObjectStoreAvailableMetric = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "data_science_pipelines_application_object_store_available",
+			Help: "Data Science Pipelines Application - Object Store Availability Status",
+		},
+		[]string{
+			"dspa_name",
+			"dspa_namespace",
+		},
+	)
 	APIServerReadyMetric = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "data_science_pipelines_application_apiserver_ready",

controllers/storage.go

@@ -18,7 +18,12 @@ package controllers
 
 import (
 	"context"
+	"encoding/base64"
+	"fmt"
+	"net/http"
 
+	minio "github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
 	dspav1alpha1 "github.com/opendatahub-io/data-science-pipelines-operator/api/v1alpha1"
 )
 
@@ -31,6 +36,69 @@ var storageTemplates = []string{
 	storageSecret,
 }
 
+func joinHostPort(host, port string) string {
+	if port == "" {
+		return host
+	}
+	return fmt.Sprintf("%s:%s", host, port)
+}
+
+func createCredentialProvidersChain(endpoint, accessKey, secretKey string) *credentials.Credentials {
+	// first try with static api key
+	if accessKey != "" && secretKey != "" {
+		return credentials.NewStaticV4(accessKey, secretKey, "")
+	}
+	// otherwise use a chained provider: minioEnv -> awsEnv -> IAM
+	providers := []credentials.Provider{
+		&credentials.EnvMinio{},
+		&credentials.EnvAWS{},
+		&credentials.IAM{
+			Client: &http.Client{
+				Transport: http.DefaultTransport,
+			},
+		},
+	}
+	return credentials.New(&credentials.Chain{Providers: providers})
+}
+
+func (r *DSPAReconciler) isObjectStorageAccessible(ctx context.Context, dsp *dspav1alpha1.DataSciencePipelinesApplication,
+	params *DSPAParams) bool {
+	log := r.Log.WithValues("namespace", dsp.Namespace).WithValues("dspa_name", dsp.Name)
+	log.Info("Performing Object Storage Health Check")
+
+	endpoint := joinHostPort(params.ObjectStorageConnection.Host, params.ObjectStorageConnection.Port)
+	accesskey, err := base64.StdEncoding.DecodeString(params.ObjectStorageConnection.AccessKeyID)
+	if err != nil {
+		log.Error(err, "Could not decode Object Storage Access Key ID")
+		return false
+	}
+
+	secretkey, err := base64.StdEncoding.DecodeString(params.ObjectStorageConnection.SecretAccessKey)
+	if err != nil {
+		log.Error(err, "Could not decode Object Storage Secret Access Key")
+		return false
+	}
+
+	cred := createCredentialProvidersChain(endpoint, string(accesskey), string(secretkey))
+	minioClient, err := minio.New(endpoint, &minio.Options{
+		Creds:  cred,
+		Secure: params.ObjectStorageConnection.Secure,
+	})
+	if err != nil {
+		log.Info(fmt.Sprintf("Could not connect to object storage endpoint: %s", endpoint))
+		return false
+	}
+
+	_, err = minioClient.ListBuckets(ctx)
+	if err != nil {
+		log.Info(fmt.Sprintf("Could not perform ListBuckets health check on object storage endpoint: %s", endpoint))
+		return false
+	}
+
+	log.Info("Object Storage Health Check Successful")
+	return true
+}
+
 // ReconcileStorage will set up Storage Connection.
 func (r *DSPAReconciler) ReconcileStorage(ctx context.Context, dsp *dspav1alpha1.DataSciencePipelinesApplication,
 	params *DSPAParams) error {