Update LogDB, Logsearch API deployments on updates to spec (#409)

pkg/apis/minio.min.io/v1/types.go

@@ -229,6 +229,14 @@ func (c ConsoleConfiguration) EqualImage(currentImage string) bool {
 	return c.Image == currentImage
 }
 
+// EqualImage returns true if image specified in `LogConfig` is equal to `image`
+func (lc *LogConfig) EqualImage(image string) bool {
+	if lc == nil {
+		return false
+	}
+	return lc.Image == image
+}
+
 // LogConfig defines configuration parameters for Log feature
 type LogConfig struct {
 	// Image defines the tenant's LogSearchAPI container image.

pkg/controller/cluster/log-search.go

@@ -18,11 +18,17 @@
 package cluster
 
 import (
+	"errors"
 	"fmt"
 
 	miniov1 "github.com/minio/operator/pkg/apis/minio.min.io/v1"
+	"github.com/minio/operator/pkg/resources/deployments"
 	"github.com/minio/operator/pkg/resources/services"
+	"github.com/minio/operator/pkg/resources/statefulsets"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/klog/v2"
 )
 
 type auditWebhookConfig struct {
@@ -41,3 +47,49 @@ func newAuditWebhookConfig(tenant *miniov1.Tenant, secret *corev1.Secret) auditW
 		args:   whArgs,
 	}
 }
+
+// logDBStatefulsetMatchesSpec checks if the log DB statefulset `actualSS`
+// matches the desired spec provided by `tenant`
+func logDBStatefulsetMatchesSpec(tenant *miniov1.Tenant, actualSS *appsv1.StatefulSet) (bool, error) {
+	if actualSS == nil {
+		return false, errors.New("cannot process an empty Log DB statefulset")
+	}
+	if tenant == nil {
+		return false, errors.New("cannot process an empty tenant")
+	}
+
+	// Note: Since tenant's Headless service name for postgres server in log
+	// search feature is internal to operator and is unaffectd by tenant
+	// spec changes we use `actualSS`'s service name to create `expectedSS`.
+	expectedSS := statefulsets.NewForLogDb(tenant, actualSS.Spec.ServiceName)
+	if !equality.Semantic.DeepDerivative(expectedSS.Spec, actualSS.Spec) {
+		// some fields set by the operator have changed
+		return false, nil
+	}
+	return true, nil
+}
+
+// logSearchAPIDeploymentMatchesSpec checks if the log DB statefulset `actualSS`
+// matches the desired spec provided by `tenant`
+func logSearchAPIDeploymentMatchesSpec(tenant *miniov1.Tenant, actualDeployment *appsv1.Deployment) (bool, error) {
+	if actualDeployment == nil {
+		return false, errors.New("cannot process an empty Logsearch API deployment")
+	}
+	if tenant == nil {
+		return false, errors.New("cannot process an empty tenant")
+	}
+
+	// compare container image version directly
+	if !tenant.Spec.Log.EqualImage(actualDeployment.Spec.Template.Spec.Containers[0].Image) {
+		klog.V(2).Infof("Tenant %s's Logsearch API server version %s doesn't match: %s", tenant.Name,
+			tenant.Spec.Log.Image, actualDeployment.Spec.Template.Spec.Containers[0].Image)
+		return false, nil
+	}
+
+	expectedDeployment := deployments.NewForLogSearchAPI(tenant)
+	if !equality.Semantic.DeepDerivative(expectedDeployment.Spec, actualDeployment.Spec) {
+		// some fields set by the operator have changed
+		return false, nil
+	}
+	return true, nil
+}

pkg/controller/cluster/main-controller.go

@@ -122,6 +122,8 @@ const (
 	StatusWaitingConsoleCert                 = "Waiting for Console TLS Certificate"
 	StatusUpdatingMinIOVersion               = "Updating MinIO Version"
 	StatusUpdatingConsole                    = "Updating Console"
+	StatusUpdatingLogPGStatefulSet           = "Updating Postgres server for Log Search feature"
+	StatusUpdatingLogSearchAPIServer         = "Updating Log Search API server"
 	StatusUpdatingResourceRequirements       = "Updating Resource Requirements"
 	StatusUpdatingAffinity                   = "Updating Pod Affinity"
 	StatusNotOwned                           = "Statefulset not controlled by operator"
@@ -1565,19 +1567,39 @@ func (c *Controller) checkAndCreateLogHeadless(ctx context.Context, tenant *mini
 }
 
 func (c *Controller) checkAndCreateLogStatefulSet(ctx context.Context, tenant *miniov1.Tenant, svcName string) error {
-	_, err := c.statefulSetLister.StatefulSets(tenant.Namespace).Get(tenant.LogStatefulsetName())
-	if err == nil || !k8serrors.IsNotFound(err) {
+	logPgSS, err := c.statefulSetLister.StatefulSets(tenant.Namespace).Get(tenant.LogStatefulsetName())
+	if err != nil {
+		if !k8serrors.IsNotFound(err) {
+			return err
+		}
+		if tenant, err = c.updateTenantStatus(ctx, tenant, StatusProvisioningLogPGStatefulSet, 0); err != nil {
+			return err
+		}
+
+		klog.V(2).Infof("Creating a new Log StatefulSet for %s", tenant.Namespace)
+		searchSS := statefulsets.NewForLogDb(tenant, svcName)
+		_, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Create(ctx, searchSS, metav1.CreateOptions{})
 		return err
+
 	}
 
-	if tenant, err = c.updateTenantStatus(ctx, tenant, StatusProvisioningLogPGStatefulSet, 0); err != nil {
+	// check if expected and actual values of Log DB spec match
+	dbSpecMatches, err := logDBStatefulsetMatchesSpec(tenant, logPgSS)
+	if err != nil {
 		return err
 	}
+	if !dbSpecMatches {
+		// Note: using current spec replica count works as long as we don't expose replicas via tenant spec.
+		if tenant, err = c.updateTenantStatus(ctx, tenant, StatusUpdatingLogPGStatefulSet, *logPgSS.Spec.Replicas); err != nil {
+			return err
+		}
+		logPgSS = statefulsets.NewForLogDb(tenant, svcName)
+		if _, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Update(ctx, logPgSS, metav1.UpdateOptions{}); err != nil {
+			return err
+		}
+	}
 
-	klog.V(2).Infof("Creating a new Log StatefulSet for %s", tenant.Namespace)
-	searchSS := statefulsets.NewForLogDb(tenant, svcName)
-	_, err = c.kubeClientSet.AppsV1().StatefulSets(tenant.Namespace).Create(ctx, searchSS, metav1.CreateOptions{})
-	return err
+	return nil
 }
 
 func (c *Controller) checkAndCreateLogSearchAPIService(ctx context.Context, tenant *miniov1.Tenant) error {
@@ -1593,18 +1615,36 @@ func (c *Controller) checkAndCreateLogSearchAPIService(ctx context.Context, tena
 }
 
 func (c *Controller) checkAndCreateLogSearchAPIDeployment(ctx context.Context, tenant *miniov1.Tenant) error {
-	_, err := c.deploymentLister.Deployments(tenant.Namespace).Get(tenant.LogSearchAPIDeploymentName())
-	if err == nil || !k8serrors.IsNotFound(err) {
+	logSearchDeployment, err := c.deploymentLister.Deployments(tenant.Namespace).Get(tenant.LogSearchAPIDeploymentName())
+	if err != nil {
+		if !k8serrors.IsNotFound(err) {
+			return err
+		}
+		if tenant, err = c.updateTenantStatus(ctx, tenant, StatusProvisioningLogSearchAPIDeployment, 0); err != nil {
+			return err
+		}
+
+		klog.V(2).Infof("Creating a new Log Search API deployment for %s", tenant.Name)
+		_, err = c.kubeClientSet.AppsV1().Deployments(tenant.Namespace).Create(ctx, deployments.NewForLogSearchAPI(tenant), metav1.CreateOptions{})
 		return err
 	}
 
-	if tenant, err = c.updateTenantStatus(ctx, tenant, StatusProvisioningLogSearchAPIDeployment, 0); err != nil {
+	// check if expected and actual values of Log search API deployment match
+	apiDeploymentMatches, err := logSearchAPIDeploymentMatchesSpec(tenant, logSearchDeployment)
+	if err != nil {
 		return err
 	}
-
-	klog.V(2).Infof("Creating a new Log Search API deployment for %s", tenant.Name)
-	_, err = c.kubeClientSet.AppsV1().Deployments(tenant.Namespace).Create(ctx, deployments.NewForLogSearchAPI(tenant), metav1.CreateOptions{})
-	return err
+	if !apiDeploymentMatches {
+		// Note: using current spec replica count works as long as we don't expose replicas via tenant spec.
+		if tenant, err = c.updateTenantStatus(ctx, tenant, StatusUpdatingLogSearchAPIServer, *logSearchDeployment.Spec.Replicas); err != nil {
+			return err
+		}
+		logSearchDeployment = deployments.NewForLogSearchAPI(tenant)
+		if _, err := c.kubeClientSet.AppsV1().Deployments(tenant.Namespace).Update(ctx, logSearchDeployment, metav1.UpdateOptions{}); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func (c *Controller) checkAndCreateLogSecret(ctx context.Context, tenant *miniov1.Tenant) (*corev1.Secret, error) {