[Feature] Add InSync Cache (#1298)

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Change Log
 
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
+- (Feature) Add InSync Cache
 
 ## [1.2.26](https://github.com/arangodb/kube-arangodb/tree/1.2.26) (2023-04-18)
 - (Bugfix) Fix manual overwrite for ReplicasCount in helm

pkg/deployment/agency/cache.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 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.
@@ -23,6 +23,7 @@ package agency
 import (
 	"context"
 	"sync"
+	"time"
 
 	"github.com/rs/zerolog"
 
@@ -157,6 +158,8 @@ type Cache interface {
 	CommitIndex() uint64
 	// Health returns true when healthy object is available.
 	Health() (Health, bool)
+	// ShardsInSyncMap returns last in sync state of particular shard
+	ShardsInSyncMap() (ShardsSyncStatus, bool)
 }
 
 func NewCache(namespace, name string, mode *api.DeploymentMode) Cache {
@@ -169,8 +172,9 @@ func NewCache(namespace, name string, mode *api.DeploymentMode) Cache {
 
 func NewAgencyCache(namespace, name string) Cache {
 	c := &cache{
-		namespace: namespace,
-		name:      name,
+		namespace:        namespace,
+		name:             name,
+		shardsSyncStatus: ShardsSyncStatus{},
 	}
 
 	c.log = logger.WrapObj(c)
@@ -185,6 +189,10 @@ func NewSingleCache() Cache {
 type cacheSingle struct {
 }
 
+func (c cacheSingle) ShardsInSyncMap() (ShardsSyncStatus, bool) {
+	return nil, false
+}
+
 func (c cacheSingle) DataDB() (StateDB, bool) {
 	return StateDB{}, false
 }
@@ -221,6 +229,8 @@ type cache struct {
 	dataDB StateDB
 
 	health Health
+
+	shardsSyncStatus ShardsSyncStatus
 }
 
 func (c *cache) WrapLogger(in *zerolog.Event) *zerolog.Event {
@@ -272,6 +282,38 @@ func (c *cache) Reload(ctx context.Context, size int, clients map[string]agency.
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
+	index, err := c.reload(ctx, size, clients)
+	if err != nil {
+		return index, err
+	}
+
+	if !c.valid {
+		return index, nil
+	}
+
+	// Refresh map of the shards
+	shardNames := c.data.GetShardsStatus()
+
+	n := time.Now()
+
+	for k := range c.shardsSyncStatus {
+		if _, ok := shardNames[k]; !ok {
+			delete(c.shardsSyncStatus, k)
+		}
+	}
+
+	for k, v := range shardNames {
+		if _, ok := c.shardsSyncStatus[k]; !ok {
+			c.shardsSyncStatus[k] = n
+		} else if v {
+			c.shardsSyncStatus[k] = n
+		}
+	}
+
+	return index, nil
+}
+
+func (c *cache) reload(ctx context.Context, size int, clients map[string]agency.Agency) (uint64, error) {
 	leaderCli, leaderConfig, health, err := c.getLeader(ctx, size, clients)
 	if err != nil {
 		// Invalidate a leader ID and agency state.
@@ -304,6 +346,21 @@ func (c *cache) Reload(ctx context.Context, size int, clients map[string]agency.
 	}
 }
 
+func (c *cache) ShardsInSyncMap() (ShardsSyncStatus, bool) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	if !c.valid {
+		return nil, false
+	}
+
+	if c.shardsSyncStatus == nil {
+		return nil, false
+	}
+
+	return c.shardsSyncStatus, true
+}
+
 // getLeader returns config and client to a leader agency, and health to check if agencies are on the same page.
 // If there is no quorum for the leader then error is returned.
 func (c *cache) getLeader(ctx context.Context, size int, clients map[string]agency.Agency) (agency.Agency, *Config, health, error) {

pkg/deployment/agency/server.go

@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2023 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.
@@ -20,6 +20,8 @@
 
 package agency
 
+import "sort"
+
 type Server string
 
 type Servers []Server
@@ -45,3 +47,50 @@ func (s Servers) Join(ids Servers) Servers {
 
 	return r
 }
+
+func (s Servers) Equals(ids Servers) bool {
+	if len(ids) != len(s) {
+		return false
+	}
+
+	for id := range ids {
+		if ids[id] != s[id] {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (s Servers) Sort() {
+	sort.Slice(s, func(i, j int) bool {
+		return s[i] < s[j]
+	})
+}
+
+func (s Servers) InSync(ids Servers) bool {
+	if len(s) != len(ids) {
+		return false
+	}
+
+	if len(s) == 0 {
+		return false
+	}
+
+	if s[0] != ids[0] {
+		return false
+	}
+
+	if len(s) > 1 {
+		s[1:].Sort()
+		ids[1:].Sort()
+
+		if s.Equals(ids) {
+			return true
+		} else {
+			return false
+		}
+	}
+
+	return true
+}

pkg/deployment/agency/shards_in_sync.go

@@ -0,0 +1,41 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 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 "time"
+
+type ShardsSyncStatus map[string]time.Time
+
+func (s ShardsSyncStatus) NotInSyncSince(t time.Duration) []string {
+	r := make([]string, 0, len(s))
+
+	for k, v := range s {
+		if v.IsZero() {
+			continue
+		}
+
+		if time.Since(v) > t {
+			r = append(r, k)
+		}
+	}
+
+	return r
+}

pkg/deployment/agency/shards_in_sync_test.go

@@ -0,0 +1,75 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 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 (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_ShardsInSync(t *testing.T) {
+	s := State{
+		Current: StateCurrent{
+			Collections: map[string]StateCurrentDBCollections{
+				"a": map[string]StateCurrentDBCollection{
+					"a": map[string]StateCurrentDBShard{
+						"s0001": {
+							Servers: Servers{
+								"A",
+								"B",
+								"C",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	t.Run("All in sync", func(t *testing.T) {
+		require.True(t, s.IsShardInSync("a", "a", "s0001", Servers{"A", "B", "C"}))
+	})
+
+	t.Run("InSync with random order", func(t *testing.T) {
+		require.True(t, s.IsShardInSync("a", "a", "s0001", Servers{"A", "C", "B"}))
+	})
+
+	t.Run("Invalid leader", func(t *testing.T) {
+		require.False(t, s.IsShardInSync("a", "a", "s0001", Servers{"B", "A", "C"}))
+	})
+
+	t.Run("Missing server", func(t *testing.T) {
+		require.False(t, s.IsShardInSync("a", "a", "s0001", Servers{"A"}))
+	})
+
+	t.Run("Missing db", func(t *testing.T) {
+		require.False(t, s.IsShardInSync("a1", "a", "s0001", Servers{"A", "B", "C"}))
+	})
+
+	t.Run("Missing col", func(t *testing.T) {
+		require.False(t, s.IsShardInSync("a", "a1", "s0001", Servers{"A", "B", "C"}))
+	})
+
+	t.Run("Missing shard", func(t *testing.T) {
+		require.False(t, s.IsShardInSync("a", "a", "s00011", Servers{"A", "B", "C"}))
+	})
+}

pkg/deployment/agency/state.go

@@ -167,6 +167,39 @@ func (s State) GetDBServerWithLowestShards() Server {
 	return resultServer
 }
 
+func (s State) GetShardsStatus() map[string]bool {
+	q := map[string]bool{}
+
+	for dName, d := range s.Plan.Collections {
+		for cName, c := range d {
+			for sName, shard := range c.Shards {
+				q[sName] = s.IsShardInSync(dName, cName, sName, shard)
+			}
+		}
+	}
+
+	return q
+}
+
+func (s State) IsShardInSync(db, col, shard string, servers Servers) bool {
+	dCurrent, ok := s.Current.Collections[db]
+	if !ok {
+		return false
+	}
+
+	cCurrent, ok := dCurrent[col]
+	if !ok {
+		return false
+	}
+
+	sCurrent, ok := cCurrent[shard]
+	if !ok {
+		return false
+	}
+
+	return sCurrent.Servers.InSync(servers)
+}
+
 // PlanServers returns all servers which are part of the plan
 func (s State) PlanServers() Servers {
 	q := map[Server]bool{}