Added sync invalidation on server shutdown.

Author: lamai93

pkg/deployment/context_impl.go

@@ -385,3 +385,8 @@ func (d *Deployment) GetExpectedPodArguments(apiObject metav1.Object, deplSpec a
 func (d *Deployment) GetShardSyncStatus() bool {
 	return d.resources.GetShardSyncStatus()
 }
+
+// InvalidateSyncStatus resets the sync state to false and triggers an inspection
+func (d *Deployment) InvalidateSyncStatus() {
+	d.resources.InvalidateSyncStatus()
+}

pkg/deployment/reconcile/context.go

@@ -95,4 +95,6 @@ type Context interface {
 		agents api.MemberStatusList, id string, version driver.Version) []string
 	// GetShardSyncStatus returns true if all shards are in sync
 	GetShardSyncStatus() bool
+	// InvalidateSyncStatus resets the sync state to false and triggers an inspection
+	InvalidateSyncStatus()
 }

pkg/deployment/reconcile/plan_builder_context.go

@@ -48,6 +48,8 @@ type PlanBuilderContext interface {
 		agents api.MemberStatusList, id string, version driver.Version) []string
 	// GetShardSyncStatus returns true if all shards are in sync
 	GetShardSyncStatus() bool
+	// InvalidateSyncStatus resets the sync state to false and triggers an inspection
+	InvalidateSyncStatus()
 	// GetStatus returns the current status of the deployment
 	GetStatus() (api.DeploymentStatus, int32)
 }

pkg/deployment/resources/deployment_health.go

@@ -54,6 +54,7 @@ func (r *Resources) RunDeploymentHealthLoop(stopCh <-chan struct{}) {
 			deploymentHealthFetchesCounters.WithLabelValues(deploymentName, metrics.Success).Inc()
 		}
 		select {
+		case <-r.shardSync.triggerSyncInspection.Done():
 		case <-time.After(time.Second * 5):
 			// Continue
 		case <-stopCh:
@@ -118,6 +119,15 @@ func (r *Resources) RunDeploymentShardSyncLoop(stopCh <-chan struct{}) {
 	}
 }
 
+// InvalidateSyncStatus resets the sync state to false and triggers an inspection
+func (r *Resources) InvalidateSyncStatus() {
+	r.log.Debug().Msg("Invalidating sync status due to previous events")
+	r.shardSync.mutex.Lock()
+	defer r.shardSync.mutex.Unlock()
+	r.shardSync.allInSync = false
+	r.shardSync.triggerSyncInspection.Trigger()
+}
+
 // fetchClusterShardSyncState performs a single fetch of the cluster inventory and
 // checks if all shards are in sync
 func (r *Resources) fetchClusterShardSyncState() error {
@@ -154,9 +164,15 @@ dbloop:
 	}
 
 	r.shardSync.mutex.Lock()
-	defer r.shardSync.mutex.Unlock()
+	oldSyncState := r.shardSync.allInSync
 	r.shardSync.allInSync = allInSync
 	r.shardSync.timestamp = time.Now()
+	r.shardSync.mutex.Unlock()
+
+	if !oldSyncState && allInSync {
+		r.log.Debug().Msg("Everything is in sync by now")
+	}
+
 	return nil
 }
 

pkg/deployment/resources/pod_inspector.go

@@ -109,6 +109,7 @@ func (r *Resources) InspectPods(ctx context.Context) (util.Interval, error) {
 					// Record termination time
 					now := metav1.Now()
 					memberStatus.RecentTerminations = append(memberStatus.RecentTerminations, now)
+					r.InvalidateSyncStatus()
 				}
 			}
 		} else if k8sutil.IsPodFailed(&p) {
@@ -122,6 +123,7 @@ func (r *Resources) InspectPods(ctx context.Context) (util.Interval, error) {
 					// Record termination time
 					now := metav1.Now()
 					memberStatus.RecentTerminations = append(memberStatus.RecentTerminations, now)
+					r.InvalidateSyncStatus()
 				}
 			}
 		}

pkg/deployment/resources/resources.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	driver "github.com/arangodb/go-driver"
+	"github.com/arangodb/kube-arangodb/pkg/util/trigger"
 	"github.com/rs/zerolog"
 )
 
@@ -41,9 +42,10 @@ type Resources struct {
 		mutex         sync.Mutex           // Mutex guarding fields in this struct
 	}
 	shardSync struct {
-		allInSync bool
-		timestamp time.Time
-		mutex     sync.Mutex
+		allInSync             bool
+		timestamp             time.Time
+		mutex                 sync.Mutex
+		triggerSyncInspection trigger.Trigger
 	}
 }