[Bugfix] Fix Agency satellite collections (#951)

Author: ajanikow

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
 - (Feature) Allow configuration for securityContext.runAsUser value
+- (Bugfix) Fix Satellite collections in Agency
 
 ## [1.2.9](https://github.com/arangodb/kube-arangodb/tree/1.2.9) (2022-03-30)
 - (Feature) Improve Kubernetes clientsets management

pkg/deployment/agency/generator_collection_test.go

@@ -32,7 +32,7 @@ type collectionGenerator struct {
 	col string
 
 	wc     *int
-	rf     *int
+	rf     *ReplicationFactor
 	shards map[int]shardGenerator
 }
 
@@ -59,6 +59,6 @@ func (c collectionGenerator) WithWriteConcern(wc int) CollectionGeneratorInterfa
 }
 
 func (c collectionGenerator) WithReplicationFactor(rf int) CollectionGeneratorInterface {
-	c.rf = &rf
+	c.rf = (*ReplicationFactor)(&rf)
 	return c
 }

pkg/deployment/agency/plan_collections.go

@@ -57,17 +57,17 @@ type StatePlanCollection struct {
 	Shards Shards  `json:"shards"`
 	// deprecated
 	// MinReplicationFactor is deprecated, but we have to support it for backward compatibility
-	MinReplicationFactor *int `json:"minReplicationFactor,omitempty"`
-	WriteConcern         *int `json:"writeConcern,omitempty"`
-	ReplicationFactor    *int `json:"replicationFactor,omitempty"`
+	MinReplicationFactor *int               `json:"minReplicationFactor,omitempty"`
+	WriteConcern         *int               `json:"writeConcern,omitempty"`
+	ReplicationFactor    *ReplicationFactor `json:"replicationFactor,omitempty"`
 }
 
-func (a *StatePlanCollection) GetReplicationFactor(shard string) int {
+func (a *StatePlanCollection) GetReplicationFactor(shard string) ReplicationFactor {
 	if a == nil {
 		return 0
 	}
 
-	l := len(a.Shards[shard])
+	l := ReplicationFactor(len(a.Shards[shard]))
 
 	if z := a.ReplicationFactor; z == nil {
 		return l

pkg/deployment/agency/rf.go

@@ -0,0 +1,76 @@
+//
+// 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 agency
+
+import (
+	"encoding/json"
+
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+const (
+	UnknownReplicationFactor   ReplicationFactor = -1000
+	SatelliteReplicationFactor ReplicationFactor = -100
+)
+
+type ReplicationFactor int
+
+func (r *ReplicationFactor) IsUnknown() bool {
+	if r == nil {
+		return false
+	}
+
+	return *r == UnknownReplicationFactor
+}
+
+func (r *ReplicationFactor) IsSatellite() bool {
+	if r == nil {
+		return false
+	}
+
+	return *r == SatelliteReplicationFactor
+}
+
+func (r *ReplicationFactor) UnmarshalJSON(bytes []byte) error {
+	var i intstr.IntOrString
+
+	if err := json.Unmarshal(bytes, &i); err != nil {
+		return err
+	}
+
+	switch i.Type {
+	case intstr.Int:
+		*r = ReplicationFactor(i.IntVal)
+		return nil
+	case intstr.String:
+		switch i.StrVal {
+		case "satellite":
+			*r = SatelliteReplicationFactor
+			return nil
+		default:
+			*r = UnknownReplicationFactor
+			return nil
+		}
+	}
+
+	return errors.Errorf("Unable to parse value")
+}

pkg/deployment/agency/state.go

@@ -219,9 +219,21 @@ func FilterDBServerShardRestart(serverID string) StateShardFilter {
 			return true
 		}
 
-		// If WriteConcern equals replicationFactor then downtime is always there
 		wc := plan.GetWriteConcern(1)
-		if rf := plan.GetReplicationFactor(shard); wc >= rf {
+		var rf int
+
+		if plan.ReplicationFactor.IsUnknown() {
+			// We are on unknown
+			rf = len(currentShard)
+		} else if plan.ReplicationFactor.IsSatellite() {
+			// We are on satellite
+			rf = len(s.PlanServers())
+		} else {
+			rf = int(plan.GetReplicationFactor(shard))
+		}
+
+		// If WriteConcern equals replicationFactor then downtime is always there
+		if wc >= rf {
 			wc = rf - 1
 		}
 

pkg/deployment/agency/state_test.go

@@ -40,12 +40,16 @@ var agencyDump38 []byte
 //go:embed testdata/agency_dump.3.9.json
 var agencyDump39 []byte
 
+//go:embed testdata/agency_dump.3.9.satellite.json
+var agencyDump39Satellite []byte
+
 var (
 	data = map[string][]byte{
-		"3.6": agencyDump36,
-		"3.7": agencyDump37,
-		"3.8": agencyDump38,
-		"3.9": agencyDump39,
+		"3.6":           agencyDump36,
+		"3.7":           agencyDump37,
+		"3.8":           agencyDump38,
+		"3.9":           agencyDump39,
+		"3.9-satellite": agencyDump39Satellite,
 	}
 )