[Bugfix] Fix CLusterScaling Integration (#1091)

CHANGELOG.md

@@ -19,6 +19,7 @@
 - (Bugfix) Always recreate DBServers if they have a leader on it.
 - (Feature) Immutable spec
 - (Bugfix) Proper agent cleanout
+- (Bugfix) Fix ClusterScaling integration
 
 ## [1.2.15](https://github.com/arangodb/kube-arangodb/tree/1.2.15) (2022-07-20)
 - (Bugfix) Ensure pod names not too long

pkg/deployment/cluster_scaling_integration.go

@@ -79,7 +79,7 @@ func newClusterScalingIntegration(depl *Deployment) *clusterScalingIntegration {
 func (ci *clusterScalingIntegration) SendUpdateToCluster(spec api.DeploymentSpec) {
 	ci.pendingUpdate.mutex.Lock()
 	defer ci.pendingUpdate.mutex.Unlock()
-	ci.pendingUpdate.spec = &spec
+	ci.pendingUpdate.spec = spec.DeepCopy()
 }
 
 // checkScalingCluster checks if inspection
@@ -112,10 +112,6 @@ func (ci *clusterScalingIntegration) checkScalingCluster(ctx context.Context, ex
 		return false
 	}
 
-	if !status.Conditions.IsTrue(api.ConditionTypeUpToDate) {
-		return false
-	}
-
 	// Update cluster with our state
 	safeToAskCluster, err := ci.updateClusterServerCount(ctx, expectSuccess)
 	if err != nil {
@@ -135,7 +131,7 @@ func (ci *clusterScalingIntegration) checkScalingCluster(ctx context.Context, ex
 	return false
 }
 
-// listenForClusterEvents keep listening for changes entered in the UI of the cluster.
+// ListenForClusterEvents keep listening for changes entered in the UI of the cluster.
 func (ci *clusterScalingIntegration) ListenForClusterEvents(stopCh <-chan struct{}) {
 	start := time.Now()
 	goodInspections := 0
@@ -176,6 +172,7 @@ func (ci *clusterScalingIntegration) cleanClusterServers(ctx context.Context) er
 	}
 
 	if req.Coordinators != nil || req.DBServers != nil {
+		log.Debug("Clean number of servers")
 		if err := arangod.CleanNumberOfServers(ctx, c.Connection()); err != nil {
 			log.Err(err).Debug("Failed to clean number of servers")
 			return errors.WithStack(err)
@@ -261,6 +258,7 @@ func (ci *clusterScalingIntegration) updateClusterServerCount(ctx context.Contex
 	ci.pendingUpdate.mutex.Lock()
 	spec := ci.pendingUpdate.spec
 	ci.pendingUpdate.mutex.Unlock()
+
 	if spec == nil {
 		// Nothing pending
 		return true, nil
@@ -288,8 +286,8 @@ func (ci *clusterScalingIntegration) updateClusterServerCount(ctx context.Contex
 
 	// This is to prevent unneseccary updates that may override some values written by the WebUI (in the case of a update loop)
 	if coordinatorCount != lastNumberOfServers.GetCoordinators() || dbserverCount != lastNumberOfServers.GetDBServers() {
+		log.Debug("Setting number of servers %d/%d", coordinatorCount, dbserverCount)
 		if err := ci.depl.SetNumberOfServers(ctx, coordinatorCountPtr, dbserverCountPtr); err != nil {
-			log.Debug("Setting number of servers %d/%d", coordinatorCount, dbserverCount)
 			if expectSuccess {
 				log.Err(err).Debug("Failed to set number of servers")
 			}

pkg/deployment/deployment.go

@@ -354,14 +354,7 @@ func (d *Deployment) run() {
 	inspectionInterval := d.inspectDeployment(minInspectionInterval)
 	log.Str("interval", inspectionInterval.String()).Debug("...deployment inspect started")
 
-	if ci := d.clusterScalingIntegration; ci != nil {
+	d.sendCIUpdate()
-		if c := d.currentObjectStatus; c != nil {
-			if a := c.AcceptedSpec; a != nil {
-				log.Debug("Send initial CI update")
-				ci.SendUpdateToCluster(*a)
-			}
-		}
-	}
 
 	for {
 		select {

pkg/deployment/deployment_inspector.go

@@ -106,16 +106,6 @@ func (d *Deployment) inspectDeployment(lastInterval util.Interval) util.Interval
 			}
 		} else if changed {
 			d.log.Info("Accepted new spec")
-			// Notify cluster of desired server count
-			if ci := d.clusterScalingIntegration; ci != nil {
-				if c := d.currentObjectStatus; c != nil {
-					if a := c.AcceptedSpec; a != nil {
-						if c.Conditions.IsTrue(api.ConditionTypeUpToDate) {
-							ci.SendUpdateToCluster(*a)
-						}
-					}
-				}
-			}
 			return minInspectionInterval // Retry ASAP
 		} else if !canProceed {
 			d.log.Err(err).Error("Cannot proceed with reconciliation")
@@ -375,6 +365,7 @@ func (d *Deployment) inspectDeploymentWithError(ctx context.Context, lastInterva
 		}
 
 		if isUpToDate && !status.Conditions.IsTrue(api.ConditionTypeUpToDate) {
+			d.sendCIUpdate()
 			if err = d.updateConditionWithHash(ctx, api.ConditionTypeUpToDate, true, "Spec is Up To Date", "Spec is Up To Date", *v); err != nil {
 				return minInspectionInterval, errors.Wrapf(err, "Unable to update UpToDate condition")
 			}
@@ -414,6 +405,16 @@ func (d *Deployment) inspectDeploymentWithError(ctx context.Context, lastInterva
 	return
 }
 
+func (d *Deployment) sendCIUpdate() {
+	if ci := d.clusterScalingIntegration; ci != nil {
+		if c := d.currentObjectStatus; c != nil {
+			if a := c.AcceptedSpec; a != nil {
+				ci.SendUpdateToCluster(*a)
+			}
+		}
+	}
+}
+
 func (d *Deployment) isUpToDateStatus(status api.DeploymentStatus) (upToDate bool, reason string) {
 	if !status.IsPlanEmpty() {
 		return false, "Plan is not empty"

pkg/deployment/features/security.go

@@ -32,6 +32,18 @@ var ephemeralVolumes = &feature{
 	enabledByDefault:   false,
 }
 
+var sensitiveInformationProtection = &feature{
+	name:               "sensitive-information-protection",
+	description:        "Hide sensitive information from metrics and logs",
+	version:            "3.7.0",
+	enterpriseRequired: false,
+	enabledByDefault:   false,
+}
+
 func EphemeralVolumes() Feature {
 	return ephemeralVolumes
 }
+
+func SensitiveInformationProtection() Feature {
+	return sensitiveInformationProtection
+}