[Bugfix] Do not check checksums for DeploymentReplicationStatus.IncomingSynchronization field values (#1179)

Author: nikita-vanyasin

CHANGELOG.md

@@ -25,6 +25,7 @@
 - (Bugfix) Fix License handling in case of broken license secret
 - (Bugfix) Check ArangoSync availability without checking healthiness
 - (Improvement) Add Anonymous Inspector mods
+- (Improvement) Do not check checksums for DeploymentReplicationStatus.IncomingSynchronization field values
 
 ## [1.2.20](https://github.com/arangodb/kube-arangodb/tree/1.2.20) (2022-10-25)
 - (Feature) Add action progress

pkg/replication/sync_inspector.go

@@ -23,14 +23,12 @@ package replication
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/arangodb/arangosync-client/client"
-	"github.com/arangodb/arangosync-client/client/synccheck"
-	"github.com/arangodb/go-driver"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/replication/v1"
-	"github.com/arangodb/kube-arangodb/pkg/deployment/features"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 )
 
@@ -68,7 +66,6 @@ func (dr *DeploymentReplication) inspectDeploymentReplication(lastInterval time.
 			}
 			timeout := CancellationTimeout + AbortTimeout
 			if isTimeExceeded(timestamp, timeout) {
-				// Cancellation and abort timeout exceeded, so it must go into failed state.
 				dr.failOnError(err, fmt.Sprintf("Failed to cancel synchronization in %s", timeout.String()))
 			}
 		}
@@ -80,12 +77,6 @@ func (dr *DeploymentReplication) inspectDeploymentReplication(lastInterval time.
 		if err != nil {
 			dr.log.Err(err).Warn("Failed to create destination syncmaster client")
 		} else {
-			destArangosyncVersion, err := destClient.Version(ctx)
-			if err != nil {
-				dr.log.Err(err).Warn("Failed to get destination arangosync version")
-				hasError = true
-			}
-
 			// Fetch status of destination
 			updateStatusNeeded := false
 			configureSyncNeeded := false
@@ -108,8 +99,7 @@ func (dr *DeploymentReplication) inspectDeploymentReplication(lastInterval time.
 							// Destination is correctly configured
 							dr.status.Conditions.Update(api.ConditionTypeConfigured, true, api.ConditionConfiguredReasonActive,
 								"Destination syncmaster is configured correctly and active")
-							dr.status.IncomingSynchronization = dr.inspectIncomingSynchronizationStatus(ctx, destClient,
-								driver.Version(destArangosyncVersion.Version), destStatus.Shards)
+							dr.status.IncomingSynchronization = dr.inspectIncomingSynchronizationStatus(destStatus)
 							updateStatusNeeded = true
 						} else {
 							// Sync is active, but from different source
@@ -249,91 +239,28 @@ func (dr *DeploymentReplication) hasOutgoingEndpoint(status client.SyncInfo, epS
 }
 
 // inspectIncomingSynchronizationStatus returns the synchronization status for the incoming sync
-func (dr *DeploymentReplication) inspectIncomingSynchronizationStatus(ctx context.Context, syncClient client.API, arangosyncVersion driver.Version, localShards []client.ShardSyncInfo) api.SynchronizationStatus {
-	dataCentersResp, err := syncClient.Master().GetDataCentersInfo(ctx)
-	if err != nil {
-		errMsg := "Failed to fetch data-centers info"
-		dr.log.Err(err).Warn(errMsg)
-		return api.SynchronizationStatus{
-			Error: fmt.Sprintf("%s: %s", errMsg, err.Error()),
-		}
-	}
+func (dr *DeploymentReplication) inspectIncomingSynchronizationStatus(destStatus client.SyncInfo) api.SynchronizationStatus {
+	const maxReportedIncomingSyncErrorsPerDatabase = 10
 
-	ch := synccheck.NewSynchronizationChecker(syncClient, time.Minute)
-	incomingSyncStatus, err := ch.CheckSync(ctx, &dataCentersResp, localShards)
-	if err != nil {
-		errMsg := "Failed to check synchronization status"
-		dr.log.Err(err).Warn(errMsg)
-		return api.SynchronizationStatus{
-			Error: fmt.Sprintf("%s: %s", errMsg, err.Error()),
+	dbs := make(map[string]api.DatabaseSynchronizationStatus, 0)
+	for _, s := range destStatus.Shards {
+		db := dbs[s.Database]
+		db.ShardsTotal++
+		if s.Status == client.SyncStatusRunning {
+			db.ShardsInSync++
+		} else if s.Status == client.SyncStatusFailed && len(db.Errors) < maxReportedIncomingSyncErrorsPerDatabase {
+			db.Errors = append(db.Errors, api.DatabaseSynchronizationError{
+				Collection: s.Collection,
+				Shard:      strconv.Itoa(s.ShardIndex),
+				Message:    fmt.Sprintf("shard sync failed: %s", s.StatusMessage),
+			})
 		}
+		dbs[s.Database] = db
 	}
-	return dr.createSynchronizationStatus(arangosyncVersion, incomingSyncStatus)
-}
 
-// createSynchronizationStatus returns aggregated info about DCSyncStatus
-func (dr *DeploymentReplication) createSynchronizationStatus(arangosyncVersion driver.Version, dcSyncStatus *synccheck.DCSyncStatus) api.SynchronizationStatus {
-	dbs := make(map[string]api.DatabaseSynchronizationStatus, len(dcSyncStatus.Databases))
-	i := 0
-	for dbName, dbSyncStatus := range dcSyncStatus.Databases {
-		i++
-		db := dbName
-		if features.SensitiveInformationProtection().Enabled() {
-			// internal IDs are not available in older versions
-			if arangosyncVersion.CompareTo("2.12.0") >= 0 {
-				db = dbSyncStatus.ID
-			} else {
-				db = fmt.Sprintf("<PROTECTED_INFO_%d>", i)
-			}
-		}
-		dbs[db] = dr.createDatabaseSynchronizationStatus(dbSyncStatus)
-	}
 	return api.SynchronizationStatus{
-		AllInSync: dcSyncStatus.AllInSync(),
+		AllInSync: destStatus.Status == client.SyncStatusRunning,
 		Databases: dbs,
 		Error:     "",
 	}
 }
-
-// createDatabaseSynchronizationStatus returns sync status for DB
-func (dr *DeploymentReplication) createDatabaseSynchronizationStatus(dbSyncStatus synccheck.DatabaseSyncStatus) api.DatabaseSynchronizationStatus {
-	// use limit for errors because the resulting status object should not be too big
-	const maxReportedIncomingSyncErrors = 20
-
-	var errs []api.DatabaseSynchronizationError
-	var shardsTotal, shardsInSync int
-	var errorsReportedToLog = 0
-	for colName, colSyncStatus := range dbSyncStatus.Collections {
-		if colSyncStatus.Error != "" && len(errs) < maxReportedIncomingSyncErrors {
-			col := colName
-			if features.SensitiveInformationProtection().Enabled() {
-				col = colSyncStatus.ID
-			}
-
-			errs = append(errs, api.DatabaseSynchronizationError{
-				Collection: col,
-				Shard:      "",
-				Message:    colSyncStatus.Error,
-			})
-		}
-
-		shardsTotal += len(colSyncStatus.Shards)
-		for shardIndex, shardSyncStatus := range colSyncStatus.Shards {
-			if shardSyncStatus.InSync {
-				shardsInSync++
-			} else if errorsReportedToLog < maxReportedIncomingSyncErrors {
-				dr.log.Str("db", dbSyncStatus.ID).
-					Str("col", colSyncStatus.ID).
-					Int("shard", shardIndex).
-					Debug("incoming synchronization shard status is not in-sync: %s", shardSyncStatus.Message)
-				errorsReportedToLog++
-			}
-		}
-	}
-
-	return api.DatabaseSynchronizationStatus{
-		ShardsTotal:  shardsTotal,
-		ShardsInSync: shardsInSync,
-		Errors:       errs,
-	}
-}