GT-449 Block traffic on the services if there is more than 1 active leader in ActiveFailover mode (#1337)

Author: jwierzbo

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Change Log
 
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
+- (Improvement) Block traffic on the services if there is more than 1 active leader in ActiveFailover mode
 
 ## [1.2.30](https://github.com/arangodb/kube-arangodb/tree/1.2.30) (2023-06-16)
 - (Feature) AgencyCache Interface

pkg/deployment/resources/logger.go

@@ -30,6 +30,6 @@ var (
 	logger = logging.Global().RegisterAndGetLogger("deployment-resources", logging.Info)
 )
 
-func (d *Resources) WrapLogger(in *zerolog.Event) *zerolog.Event {
-	return in.Str("namespace", d.namespace).Str("name", d.name)
+func (r *Resources) WrapLogger(in *zerolog.Event) *zerolog.Event {
+	return in.Str("namespace", r.namespace).Str("name", r.name)
 }

pkg/deployment/resources/metrics.go

@@ -168,29 +168,29 @@ type MetricMemberRestarts map[int32]MetricMemberRestartReason
 
 type MetricMemberRestartReason map[string]uint64
 
-func (d *Resources) CollectMetrics(m metrics.PushMetric) {
-	for member, info := range d.metrics.Members {
+func (r *Resources) CollectMetrics(m metrics.PushMetric) {
+	for member, info := range r.metrics.Members {
 		// Containers
 		for container, restarts := range info.ContainerRestarts {
 			for code, reasons := range restarts {
 				for reason, count := range reasons {
-					m.Push(metric_descriptions.ArangodbOperatorMembersUnexpectedContainerExitCodesCounter(float64(count), d.namespace, d.name, member, container, "container", fmt.Sprintf("%d", code), reason))
+					m.Push(metric_descriptions.ArangodbOperatorMembersUnexpectedContainerExitCodesCounter(float64(count), r.namespace, r.name, member, container, "container", fmt.Sprintf("%d", code), reason))
 				}
 			}
 		}
 		// InitContainers
 		for container, restarts := range info.InitContainerRestarts {
 			for code, reasons := range restarts {
 				for reason, count := range reasons {
-					m.Push(metric_descriptions.ArangodbOperatorMembersUnexpectedContainerExitCodesCounter(float64(count), d.namespace, d.name, member, container, "initContainer", fmt.Sprintf("%d", code), reason))
+					m.Push(metric_descriptions.ArangodbOperatorMembersUnexpectedContainerExitCodesCounter(float64(count), r.namespace, r.name, member, container, "initContainer", fmt.Sprintf("%d", code), reason))
 				}
 			}
 		}
 		// EphemeralContainers
 		for container, restarts := range info.EphemeralContainerRestarts {
 			for code, reasons := range restarts {
 				for reason, count := range reasons {
-					m.Push(metric_descriptions.ArangodbOperatorMembersUnexpectedContainerExitCodesCounter(float64(count), d.namespace, d.name, member, container, "ephemeralContainer", fmt.Sprintf("%d", code), reason))
+					m.Push(metric_descriptions.ArangodbOperatorMembersUnexpectedContainerExitCodesCounter(float64(count), r.namespace, r.name, member, container, "ephemeralContainer", fmt.Sprintf("%d", code), reason))
 				}
 			}
 		}

pkg/deployment/resources/pod_leader.go

@@ -22,6 +22,7 @@ package resources
 
 import (
 	"context"
+	"strings"
 	"sync"
 
 	core "k8s.io/api/core/v1"
@@ -150,11 +151,11 @@ func (r *Resources) EnsureLeader(ctx context.Context, cachedStatus inspectorInte
 	return errors.Reconcile()
 }
 
-// getSingleServerLeaderID returns id of a single server leader.
-func (r *Resources) getSingleServerLeaderID(ctx context.Context) (string, error) {
+// getSingleServerLeaderID returns ids of a single server leaders.
+func (r *Resources) getSingleServerLeaderID(ctx context.Context) ([]string, error) {
 	status := r.context.GetStatus()
 	var mutex sync.Mutex
-	var leaderID string
+	var leaderIDs []string
 	var anyError error
 
 	ctxCancel, cancel := context.WithCancel(ctx)
@@ -177,10 +178,8 @@ func (r *Resources) getSingleServerLeaderID(ctx context.Context) (string, error)
 					return errors.New("not available")
 				}
 
-				// Other requests can be interrupted, because a leader is known already.
-				cancel()
 				mutex.Lock()
-				leaderID = id
+				leaderIDs = append(leaderIDs, id)
 				mutex.Unlock()
 				return nil
 			})
@@ -194,15 +193,15 @@ func (r *Resources) getSingleServerLeaderID(ctx context.Context) (string, error)
 	}
 	wg.Wait()
 
-	if len(leaderID) > 0 {
-		return leaderID, nil
+	if len(leaderIDs) > 0 {
+		return leaderIDs, nil
 	}
 
 	if anyError != nil {
-		return "", errors.WithMessagef(anyError, "unable to get a leader")
+		return nil, errors.WithMessagef(anyError, "unable to get a leader")
 	}
 
-	return "", errors.New("unable to get a leader")
+	return nil, errors.New("unable to get a leader")
 }
 
 // setSingleServerLeadership adds or removes leadership label on a single server pod.
@@ -212,10 +211,16 @@ func (r *Resources) ensureSingleServerLeader(ctx context.Context, cachedStatus i
 	enabled := features.FailoverLeadership().Enabled()
 	var leaderID string
 	if enabled {
-		var err error
-		if leaderID, err = r.getSingleServerLeaderID(ctx); err != nil {
+		leaderIDs, err := r.getSingleServerLeaderID(ctx)
+		if err != nil {
 			return err
 		}
+
+		if len(leaderIDs) == 1 {
+			leaderID = leaderIDs[0]
+		} else if len(leaderIDs) > 1 {
+			r.log.Error("multiple leaders found: %s. Blocking traffic to the deployment services", strings.Join(leaderIDs, ", "))
+		}
 	}
 
 	status := r.context.GetStatus()