[Bugfix] Do not stop Sync if Synchronization is in progress (#1203)

ajanikow · web-flow · commit 9f4d3975b944 · 2022-12-08T12:38:31.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -37,6 +37,7 @@
 - (Feature) Configurable ArangoD Port
 - (Feature) Allow to exclude metrics
 - (Feature) Do not restart member if all pods in group are not ready
+- (Bugfix) Do not stop Sync if Synchronization is in progress 
 
 ## [1.2.20](https://github.com/arangodb/kube-arangodb/tree/1.2.20) (2022-10-25)
 - (Feature) Add action progress
diff --git a/pkg/apis/deployment/v1/conditions.go b/pkg/apis/deployment/v1/conditions.go
@@ -121,6 +121,9 @@ const (
 
 	// ConditionTypeMaintenance indicates that maintenance is enabled on cluster
 	ConditionTypeMaintenance ConditionType = "Maintenance"
+
+	// ConditionTypeSyncEnabled Define if sync is enabled
+	ConditionTypeSyncEnabled ConditionType = "SyncEnabled"
 )
 
 // Condition represents one current condition of a deployment or deployment member.
diff --git a/pkg/apis/deployment/v2alpha1/conditions.go b/pkg/apis/deployment/v2alpha1/conditions.go
@@ -121,6 +121,9 @@ const (
 
 	// ConditionTypeMaintenance indicates that maintenance is enabled on cluster
 	ConditionTypeMaintenance ConditionType = "Maintenance"
+
+	// ConditionTypeSyncEnabled Define if sync is enabled
+	ConditionTypeSyncEnabled ConditionType = "SyncEnabled"
 )
 
 // Condition represents one current condition of a deployment or deployment member.
diff --git a/pkg/deployment/access_package.go b/pkg/deployment/access_package.go
@@ -50,7 +50,7 @@ func (d *Deployment) createAccessPackages(ctx context.Context) error {
 	log := d.sectionLogger("access-package")
 	spec := d.GetSpec()
 
-	if !spec.Sync.IsEnabled() {
+	if !d.IsSyncEnabled() {
 		// We're only relevant when sync is enabled
 		return nil
 	}
diff --git a/pkg/deployment/agency/cache.go b/pkg/deployment/agency/cache.go
@@ -153,6 +153,7 @@ type Health interface {
 type Cache interface {
 	Reload(ctx context.Context, size int, clients map[string]agency.Agency) (uint64, error)
 	Data() (State, bool)
+	DataDB() (StateDB, bool)
 	CommitIndex() uint64
 	// Health returns true when healthy object is available.
 	Health() (Health, bool)
@@ -184,6 +185,10 @@ func NewSingleCache() Cache {
 type cacheSingle struct {
 }
 
+func (c cacheSingle) DataDB() (StateDB, bool) {
+	return StateDB{}, false
+}
+
 func (c cacheSingle) CommitIndex() uint64 {
 	return 0
 }
@@ -212,7 +217,8 @@ type cache struct {
 
 	commitIndex uint64
 
-	data State
+	data   State
+	dataDB StateDB
 
 	health Health
 }
@@ -232,7 +238,22 @@ func (c *cache) Data() (State, bool) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
-	return c.data, c.valid
+	if !c.valid {
+		return State{}, false
+	}
+
+	return c.data, true
+}
+
+func (c *cache) DataDB() (StateDB, bool) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	if !c.valid {
+		return StateDB{}, false
+	}
+
+	return c.dataDB, c.valid
 }
 
 // Health returns always false for single cache.
@@ -275,7 +296,8 @@ func (c *cache) Reload(ctx context.Context, size int, clients map[string]agency.
 		c.valid = false
 		return leaderConfig.CommitIndex, err
 	} else {
-		c.data = data
+		c.data = data.Arango
+		c.dataDB = data.ArangoDB
 		c.valid = true
 		c.commitIndex = leaderConfig.CommitIndex
 		return leaderConfig.CommitIndex, nil
diff --git a/pkg/deployment/agency/definitions.go b/pkg/deployment/agency/definitions.go
@@ -26,7 +26,8 @@ import (
 )
 
 const (
-	ArangoKey = "arango"
+	ArangoKey   = "arango"
+	ArangoDBKey = "arangodb"
 
 	PlanKey    = "Plan"
 	CurrentKey = "Current"
@@ -48,6 +49,13 @@ const (
 	TargetJobFinishedKey = "Finished"
 
 	TargetCleanedServersKey = "CleanedServers"
+
+	ArangoSyncKey                     = "arangosync"
+	ArangoSyncStateKey                = "synchronizationState"
+	ArangoSyncStateIncomingKey        = "incoming"
+	ArangoSyncStateIncomingStateKey   = "state"
+	ArangoSyncStateOutgoingKey        = "outgoing"
+	ArangoSyncStateOutgoingTargetsKey = "targets"
 )
 
 func GetAgencyKey(parts ...string) string {
diff --git a/pkg/deployment/agency/state.go b/pkg/deployment/agency/state.go
@@ -31,12 +31,12 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 )
 
-func (c *cache) loadState(ctx context.Context, client agency.Agency) (State, error) {
+func (c *cache) loadState(ctx context.Context, client agency.Agency) (StateRoot, error) {
 	conn := client.Connection()
 
 	req, err := client.Connection().NewRequest(http.MethodPost, "/_api/agency/read")
 	if err != nil {
-		return State{}, err
+		return StateRoot{}, err
 	}
 
 	var data []byte
@@ -53,33 +53,35 @@ func (c *cache) loadState(ctx context.Context, client agency.Agency) (State, err
 		GetAgencyKey(ArangoKey, TargetKey, TargetJobFailedKey),
 		GetAgencyKey(ArangoKey, TargetKey, TargetJobFinishedKey),
 		GetAgencyKey(ArangoKey, TargetKey, TargetCleanedServersKey),
+		GetAgencyKey(ArangoDBKey, ArangoSyncKey, ArangoSyncStateKey, ArangoSyncStateIncomingKey, ArangoSyncStateIncomingStateKey),
+		GetAgencyKey(ArangoDBKey, ArangoSyncKey, ArangoSyncStateKey, ArangoSyncStateOutgoingKey, ArangoSyncStateOutgoingTargetsKey),
 	}
 
 	req, err = req.SetBody(GetAgencyReadRequest(GetAgencyReadKey(readKeys...)))
 	if err != nil {
-		return State{}, err
+		return StateRoot{}, err
 	}
 
 	resp, err := conn.Do(driver.WithRawResponse(ctx, &data), req)
 	if err != nil {
-		return State{}, err
+		return StateRoot{}, err
 	}
 
 	if err := resp.CheckStatus(http.StatusOK); err != nil {
-		return State{}, err
+		return StateRoot{}, err
 	}
 
 	var r StateRoots
 
 	if err := json.Unmarshal(data, &r); err != nil {
-		return State{}, err
+		return StateRoot{}, err
 	}
 
 	if len(r) != 1 {
-		return State{}, errors.Newf("Invalid response size")
+		return StateRoot{}, errors.Newf("Invalid response size")
 	}
 
-	state := r[0].Arango
+	state := r[0]
 
 	return state, nil
 }
diff --git a/pkg/deployment/deployment.go b/pkg/deployment/deployment.go
@@ -137,6 +137,21 @@ type Deployment struct {
 	metrics Metrics
 }
 
+func (d *Deployment) IsSyncEnabled() bool {
+	d.currentObjectLock.RLock()
+	defer d.currentObjectLock.RUnlock()
+
+	if d.currentObject.Spec.Sync.IsEnabled() {
+		return true
+	}
+
+	if d.currentObject.Status.Conditions.IsTrue(api.ConditionTypeSyncEnabled) {
+		return true
+	}
+
+	return false
+}
+
 func (d *Deployment) GetMembersState() memberState.StateInspector {
 	return d.memberState
 }
@@ -149,6 +164,10 @@ func (d *Deployment) GetAgencyHealth() (agency.Health, bool) {
 	return d.agencyCache.Health()
 }
 
+func (d *Deployment) GetAgencyArangoDBCache() (agency.StateDB, bool) {
+	return d.agencyCache.DataDB()
+}
+
 func (d *Deployment) RefreshAgencyCache(ctx context.Context) (uint64, error) {
 	if d.GetSpec().Mode.Get() == api.DeploymentModeSingle {
 		return 0, nil
diff --git a/pkg/deployment/reconcile/action_context.go b/pkg/deployment/reconcile/action_context.go
@@ -140,6 +140,10 @@ type actionContext struct {
 	metrics      *Metrics
 }
 
+func (ac *actionContext) IsSyncEnabled() bool {
+	return ac.context.IsSyncEnabled()
+}
+
 func (ac *actionContext) WithMemberStatusUpdateErr(ctx context.Context, id string, group api.ServerGroup, action reconciler.DeploymentMemberStatusUpdateErrFunc) error {
 	return ac.context.WithMemberStatusUpdateErr(ctx, id, group, action)
 }
@@ -251,6 +255,10 @@ func (ac *actionContext) GetAgencyCache() (agencyCache.State, bool) {
 	return ac.context.GetAgencyCache()
 }
 
+func (ac *actionContext) GetAgencyArangoDBCache() (agencyCache.StateDB, bool) {
+	return ac.context.GetAgencyArangoDBCache()
+}
+
 func (ac *actionContext) SetAgencyMaintenanceMode(ctx context.Context, enabled bool) error {
 	return ac.context.SetAgencyMaintenanceMode(ctx, enabled)
 }
diff --git a/pkg/deployment/reconcile/plan_builder_high.go b/pkg/deployment/reconcile/plan_builder_high.go
@@ -48,6 +48,7 @@ func (r *Reconciler) createHighPlan(ctx context.Context, apiObject k8sutil.APIOb
 		ApplyIfEmpty(r.updateMemberPodTemplateSpec).
 		ApplyIfEmpty(r.updateMemberPhasePlan).
 		ApplyIfEmpty(r.createCleanOutPlan).
+		ApplyIfEmpty(r.createSyncPlan).
 		ApplyIfEmpty(r.updateMemberUpdateConditionsPlan).
 		ApplyIfEmpty(r.updateMemberRotationConditionsPlan).
 		ApplyIfEmpty(r.createMemberRecreationConditionsPlan).
diff --git a/pkg/deployment/reconcile/plan_builder_scale.go b/pkg/deployment/reconcile/plan_builder_scale.go
@@ -63,7 +63,7 @@ func (r *Reconciler) createScaleMemberPlan(ctx context.Context, apiObject k8suti
 	}
 	if spec.GetMode().SupportsSync() {
 		// Scale syncmasters & syncworkers
-		if spec.Sync.IsEnabled() {
+		if context.IsSyncEnabled() {
 			plan = append(plan, r.createScalePlan(status, status.Members.SyncMasters, api.ServerGroupSyncMasters, spec.SyncMasters.GetCount(), context)...)
 			plan = append(plan, r.createScalePlan(status, status.Members.SyncWorkers, api.ServerGroupSyncWorkers, spec.SyncWorkers.GetCount(), context)...)
 		} else {
diff --git a/pkg/deployment/reconcile/plan_builder_sync.go b/pkg/deployment/reconcile/plan_builder_sync.go
@@ -0,0 +1,63 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package reconcile
+
+import (
+	"context"
+
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+)
+
+func (r *Reconciler) createSyncPlan(ctx context.Context, apiObject k8sutil.APIObject,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	context PlanBuilderContext) api.Plan {
+
+	if spec.Sync.IsEnabled() {
+		if status.Conditions.IsTrue(api.ConditionTypeSyncEnabled) {
+			// We do not have anything to do here
+			return nil
+		}
+
+		// We need to enable condition
+		return api.Plan{updateConditionActionV2("Sync Enabled", api.ConditionTypeSyncEnabled, true, "Sync enabled", "", "")}
+	}
+
+	if !status.Conditions.IsTrue(api.ConditionTypeSyncEnabled) {
+		// Sync is disabled
+		return nil
+	}
+
+	cache, ok := context.GetAgencyArangoDBCache()
+	if !ok {
+		// Unable to get agency cache
+		return nil
+	}
+
+	if cache.ArangoSync.Error == nil {
+		if !cache.ArangoSync.IsSyncInProgress() {
+			// Remove condition
+			return api.Plan{removeConditionActionV2("Sync Disabled", api.ConditionTypeSyncEnabled)}
+		}
+	}
+
+	return nil
+}
diff --git a/pkg/deployment/reconcile/plan_builder_test.go b/pkg/deployment/reconcile/plan_builder_test.go
@@ -85,6 +85,15 @@ type testContext struct {
 	state     member.StateInspector
 }
 
+func (c *testContext) IsSyncEnabled() bool {
+	return false
+}
+
+func (c *testContext) GetAgencyArangoDBCache() (agencyCache.StateDB, bool) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (c *testContext) WithMemberStatusUpdateErr(ctx context.Context, id string, group api.ServerGroup, action reconciler.DeploymentMemberStatusUpdateErrFunc) error {
 	//TODO implement me
 	panic("implement me")
diff --git a/pkg/deployment/reconcile/plan_builder_tls.go b/pkg/deployment/reconcile/plan_builder_tls.go
@@ -282,7 +282,7 @@ func (r *Reconciler) createKeyfileRenewalPlanSynced(ctx context.Context, apiObje
 	spec api.DeploymentSpec, status api.DeploymentStatus,
 	planCtx PlanBuilderContext) api.Plan {
 
-	if !spec.Sync.IsEnabled() || !spec.Sync.TLS.IsSecure() {
+	if !planCtx.IsSyncEnabled() || !spec.Sync.TLS.IsSecure() {
 		return nil
 	}
 
diff --git a/pkg/deployment/reconciler/context.go b/pkg/deployment/reconciler/context.go
@@ -96,6 +96,7 @@ type DeploymentImageManager interface {
 
 type ArangoAgencyGet interface {
 	GetAgencyCache() (agencyCache.State, bool)
+	GetAgencyArangoDBCache() (agencyCache.StateDB, bool)
 	GetAgencyHealth() (agencyCache.Health, bool)
 }
 
@@ -118,6 +119,8 @@ type DeploymentInfoGetter interface {
 	GetName() string
 	// GetNamespace returns the namespace that contains the deployment
 	GetNamespace() string
+	// IsSyncEnabled returns information if sync is enabled
+	IsSyncEnabled() bool
 }
 
 type ArangoApplier interface {
diff --git a/pkg/deployment/resources/pdbs.go b/pkg/deployment/resources/pdbs.go
@@ -61,7 +61,7 @@ func (r *Resources) EnsurePDBs(ctx context.Context) error {
 		// Setting those to zero triggers a remove of the PDB
 		minSyncMaster := 0
 		minSyncWorker := 0
-		if spec.Sync.IsEnabled() {
+		if r.context.IsSyncEnabled() {
 			minSyncMaster = spec.GetServerGroupSpec(api.ServerGroupSyncMasters).GetCount() - 1
 			minSyncWorker = spec.GetServerGroupSpec(api.ServerGroupSyncWorkers).GetCount() - 1
 		}
diff --git a/pkg/deployment/resources/pod_inspector.go b/pkg/deployment/resources/pod_inspector.go
@@ -503,7 +503,7 @@ func (r *Resources) InspectPods(ctx context.Context, cachedStatus inspectorInter
 	}
 
 	spec := r.context.GetSpec()
-	allMembersReady := status.Members.AllMembersReady(spec.GetMode(), spec.Sync.IsEnabled())
+	allMembersReady := status.Members.AllMembersReady(spec.GetMode(), r.context.IsSyncEnabled())
 	status.Conditions.Update(api.ConditionTypeReady, allMembersReady, "", "")
 
 	// Update conditions
diff --git a/pkg/deployment/resources/secret_hashes.go b/pkg/deployment/resources/secret_hashes.go
@@ -185,7 +185,7 @@ func (r *Resources) ValidateSecretHashes(ctx context.Context, cachedStatus inspe
 			}
 		}
 	}
-	if spec.Sync.IsEnabled() {
+	if r.context.IsSyncEnabled() {
 		secretName := spec.Sync.TLS.GetCASecretName()
 		getExpectedHash := func() string { return getHashes().SyncTLSCA }
 		setExpectedHash := func(h string) error {
diff --git a/pkg/deployment/resources/secrets.go b/pkg/deployment/resources/secrets.go
@@ -170,7 +170,7 @@ func (r *Resources) EnsureSecrets(ctx context.Context, cachedStatus inspectorInt
 			}
 		}
 	}
-	if spec.Sync.IsEnabled() {
+	if r.context.IsSyncEnabled() {
 		counterMetric.Inc()
 		if err := reconcileRequired.WithError(r.ensureTokenSecret(ctx, cachedStatus, secrets, spec.Sync.Authentication.GetJWTSecretName())); err != nil {
 			return errors.Section(err, "Sync Auth")
diff --git a/pkg/deployment/resources/services.go b/pkg/deployment/resources/services.go

Original file line number	Diff line number	Diff line change
`@@ -121,6 +121,9 @@ const (`
`121`	`121`
`122`	`122`	`// ConditionTypeMaintenance indicates that maintenance is enabled on cluster`
`123`	`123`	`ConditionTypeMaintenance ConditionType = "Maintenance"`
	`124`	`+`
	`125`	`+ // ConditionTypeSyncEnabled Define if sync is enabled`
	`126`	`+ ConditionTypeSyncEnabled ConditionType = "SyncEnabled"`
`124`	`127`	`)`
`125`	`128`
`126`	`129`	`// Condition represents one current condition of a deployment or deployment member.`
Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ func (d *Deployment) createAccessPackages(ctx context.Context) error {`
`50`	`50`	`log := d.sectionLogger("access-package")`
`51`	`51`	`spec := d.GetSpec()`
`52`	`52`
`53`		`- if !spec.Sync.IsEnabled() {`
	`53`	`+ if !d.IsSyncEnabled() {`
`54`	`54`	`// We're only relevant when sync is enabled`
`55`	`55`	`return nil`
`56`	`56`	`}`
Original file line number	Diff line number	Diff line change
`@@ -31,12 +31,12 @@ import (`
`31`	`31`	`"github.com/arangodb/kube-arangodb/pkg/util/errors"`
`32`	`32`	`)`
`33`	`33`
`34`		`-func (c *cache) loadState(ctx context.Context, client agency.Agency) (State, error) {`
	`34`	`+func (c *cache) loadState(ctx context.Context, client agency.Agency) (StateRoot, error) {`
`35`	`35`	`conn := client.Connection()`
`36`	`36`
`37`	`37`	`req, err := client.Connection().NewRequest(http.MethodPost, "/_api/agency/read")`
`38`	`38`	`if err != nil {`
`39`		`- return State{}, err`
	`39`	`+ return StateRoot{}, err`
`40`	`40`	`}`
`41`	`41`
`42`	`42`	`var data []byte`
`@@ -53,33 +53,35 @@ func (c *cache) loadState(ctx context.Context, client agency.Agency) (State, err`
`53`	`53`	`GetAgencyKey(ArangoKey, TargetKey, TargetJobFailedKey),`
`54`	`54`	`GetAgencyKey(ArangoKey, TargetKey, TargetJobFinishedKey),`
`55`	`55`	`GetAgencyKey(ArangoKey, TargetKey, TargetCleanedServersKey),`
	`56`	`+ GetAgencyKey(ArangoDBKey, ArangoSyncKey, ArangoSyncStateKey, ArangoSyncStateIncomingKey, ArangoSyncStateIncomingStateKey),`
	`57`	`+ GetAgencyKey(ArangoDBKey, ArangoSyncKey, ArangoSyncStateKey, ArangoSyncStateOutgoingKey, ArangoSyncStateOutgoingTargetsKey),`
`56`	`58`	`}`
`57`	`59`
`58`	`60`	`req, err = req.SetBody(GetAgencyReadRequest(GetAgencyReadKey(readKeys...)))`
`59`	`61`	`if err != nil {`
`60`		`- return State{}, err`
	`62`	`+ return StateRoot{}, err`
`61`	`63`	`}`
`62`	`64`
`63`	`65`	`resp, err := conn.Do(driver.WithRawResponse(ctx, &data), req)`
`64`	`66`	`if err != nil {`
`65`		`- return State{}, err`
	`67`	`+ return StateRoot{}, err`
`66`	`68`	`}`
`67`	`69`
`68`	`70`	`if err := resp.CheckStatus(http.StatusOK); err != nil {`
`69`		`- return State{}, err`
	`71`	`+ return StateRoot{}, err`
`70`	`72`	`}`
`71`	`73`
`72`	`74`	`var r StateRoots`
`73`	`75`
`74`	`76`	`if err := json.Unmarshal(data, &r); err != nil {`
`75`		`- return State{}, err`
	`77`	`+ return StateRoot{}, err`
`76`	`78`	`}`
`77`	`79`
`78`	`80`	`if len(r) != 1 {`
`79`		`- return State{}, errors.Newf("Invalid response size")`
	`81`	`+ return StateRoot{}, errors.Newf("Invalid response size")`
`80`	`82`	`}`
`81`	`83`
`82`		`- state := r[0].Arango`
	`84`	`+ state := r[0]`
`83`	`85`
`84`	`86`	`return state, nil`
`85`	`87`	`}`
Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,10 @@ type actionContext struct {`
`140`	`140`	`metrics *Metrics`
`141`	`141`	`}`
`142`	`142`
	`143`	`+func (ac *actionContext) IsSyncEnabled() bool {`
	`144`	`+ return ac.context.IsSyncEnabled()`
	`145`	`+}`
	`146`	`+`
`143`	`147`	`func (ac *actionContext) WithMemberStatusUpdateErr(ctx context.Context, id string, group api.ServerGroup, action reconciler.DeploymentMemberStatusUpdateErrFunc) error {`
`144`	`148`	`return ac.context.WithMemberStatusUpdateErr(ctx, id, group, action)`
`145`	`149`	`}`
`@@ -251,6 +255,10 @@ func (ac *actionContext) GetAgencyCache() (agencyCache.State, bool) {`
`251`	`255`	`return ac.context.GetAgencyCache()`
`252`	`256`	`}`
`253`	`257`
	`258`	`+func (ac *actionContext) GetAgencyArangoDBCache() (agencyCache.StateDB, bool) {`
	`259`	`+ return ac.context.GetAgencyArangoDBCache()`
	`260`	`+}`
	`261`	`+`
`254`	`262`	`func (ac *actionContext) SetAgencyMaintenanceMode(ctx context.Context, enabled bool) error {`
`255`	`263`	`return ac.context.SetAgencyMaintenanceMode(ctx, enabled)`
`256`	`264`	`}`