GT-526 Use agency cache lock in metrics exporter

Author: jwierzbo

CHANGELOG.md

@@ -14,6 +14,7 @@
 - (Documentation) Update ArangoDeploymentReplication and ArangoLocalStorage CR auto-generated docs
 - (Feature) Add ArangoMember Message and extend ArangoMember CRD
 - (Documentation) Use OpenAPI-compatible type names in docs
+- (Improvement) Use agency cache lock in metrics exporter
 
 ## [1.2.34](https://github.com/arangodb/kube-arangodb/tree/1.2.34) (2023-10-16)
 - (Bugfix) Fix make manifests-crd-file command

pkg/deployment/agency/cache.go

@@ -144,7 +144,9 @@ type Health interface {
 
 type Cache interface {
 	Reload(ctx context.Context, size int, clients Connections) (uint64, error)
-	Data() (state.State, bool)
+	// Data returns current state. If no state is available, false is returned.
+	// Use lock to access data which can be changed by reload.
+	Data() (state.State, *sync.RWMutex, bool)
 	DataDB() (state.DB, bool)
 	CommitIndex() uint64
 	// Health returns true when healthy object is available.
@@ -202,8 +204,8 @@ func (c cacheSingle) Reload(_ context.Context, _ int, _ Connections) (uint64, er
 	return 0, nil
 }
 
-func (c cacheSingle) Data() (state.State, bool) {
-	return state.State{}, true
+func (c cacheSingle) Data() (state.State, *sync.RWMutex, bool) {
+	return state.State{}, nil, true
 }
 
 type cache struct {
@@ -232,16 +234,16 @@ func (c *cache) CommitIndex() uint64 {
 	return index
 }
 
-func (c *cache) Data() (state.State, bool) {
+func (c *cache) Data() (state.State, *sync.RWMutex, bool) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
 	data, _, ok := c.loader.State()
 	if ok {
-		return data.Arango, true
+		return data.Arango, &c.lock, true
 	}
 
-	return state.State{}, false
+	return state.State{}, nil, false
 }
 
 func (c *cache) DataDB() (state.DB, bool) {

pkg/deployment/deployment.go

@@ -159,10 +159,16 @@ func (d *Deployment) GetMembersState() memberState.StateInspector {
 	return d.memberState
 }
 
-func (d *Deployment) GetAgencyCache() (state.State, bool) {
+func (d *Deployment) GetAgencyCacheWithLock() (state.State, *sync.RWMutex, bool) {
 	return d.agencyCache.Data()
 }
 
+// Deprecated: Use GetAgencyCacheWithLock instead
+func (d *Deployment) GetAgencyCache() (state.State, bool) {
+	data, _, err := d.agencyCache.Data()
+	return data, err
+}
+
 func (d *Deployment) GetAgencyHealth() (agency.Health, bool) {
 	return d.agencyCache.Health()
 }

pkg/deployment/old_metrics.go

@@ -92,7 +92,8 @@ func (i *inventory) CollectMetrics(in metrics.PushMetric) {
 			}
 
 			if spec.Mode.Get().HasAgents() {
-				agency, agencyOk := deployment.GetAgencyCache()
+				// We get here a copy of the object, not a pointer
+				agency, lock, agencyOk := deployment.GetAgencyCacheWithLock()
 				if !agencyOk {
 					in.Push(i.deploymentAgencyStateMetric.Gauge(0, deployment.GetNamespace(), deployment.GetName()))
 					continue
@@ -101,6 +102,7 @@ func (i *inventory) CollectMetrics(in metrics.PushMetric) {
 				in.Push(i.deploymentAgencyStateMetric.Gauge(1, deployment.GetNamespace(), deployment.GetName()))
 
 				if spec.Mode.Get() == api.DeploymentModeCluster {
+					lock.RLock()
 					for db, collections := range agency.Current.Collections {
 						dbName := db
 						if features.SensitiveInformationProtection().Enabled() {
@@ -142,6 +144,7 @@ func (i *inventory) CollectMetrics(in metrics.PushMetric) {
 							}
 						}
 					}
+					lock.RUnlock()
 				}
 			}
 		}

pkg/deployment/reconcile/action_context.go

@@ -22,6 +22,7 @@ package reconcile
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	core "k8s.io/api/core/v1"
@@ -260,6 +261,10 @@ func (ac *actionContext) ShardsInSyncMap() (state.ShardsSyncStatus, bool) {
 	return ac.context.ShardsInSyncMap()
 }
 
+func (ac *actionContext) GetAgencyCacheWithLock() (state.State, *sync.RWMutex, bool) {
+	return ac.context.GetAgencyCacheWithLock()
+}
+
 func (ac *actionContext) GetAgencyCache() (state.State, bool) {
 	return ac.context.GetAgencyCache()
 }

pkg/deployment/reconcile/plan_builder_test.go

@@ -24,6 +24,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"sync"
 	"testing"
 
 	monitoringClient "github.com/prometheus-operator/prometheus-operator/pkg/client/versioned/typed/monitoring/v1"
@@ -176,6 +177,10 @@ func (c *testContext) GenerateMemberEndpoint(group api.ServerGroup, member api.M
 	return pod2.GenerateMemberEndpoint(c.Inspector, c.ArangoDeployment, c.ArangoDeployment.Spec, group, member)
 }
 
+func (c *testContext) GetAgencyCacheWithLock() (state.State, *sync.RWMutex, bool) {
+	return state.State{}, nil, true
+}
+
 func (c *testContext) GetAgencyCache() (state.State, bool) {
 	return state.State{}, true
 }

pkg/deployment/reconciler/context.go

@@ -22,8 +22,8 @@ package reconciler
 
 import (
 	"context"
-
 	core "k8s.io/api/core/v1"
+	"sync"
 
 	"github.com/arangodb/arangosync-client/client"
 	"github.com/arangodb/go-driver"
@@ -96,6 +96,10 @@ type DeploymentImageManager interface {
 }
 
 type ArangoAgencyGet interface {
+	// GetAgencyCacheWithLock returns current Agency state, if no state is available, false is returned.
+	// Use lock to access data which can be changed by reload.
+	GetAgencyCacheWithLock() (state.State, *sync.RWMutex, bool)
+	// Deprecated: Use GetAgencyCacheWithLock instead
 	GetAgencyCache() (state.State, bool)
 	GetAgencyArangoDBCache() (state.DB, bool)
 	GetAgencyHealth() (agencyCache.Health, bool)

pkg/kuba/start.go

@@ -0,0 +1,30 @@
+package main
+
+type one struct {
+	Item two
+}
+
+type two struct {
+	Map map[string]string
+}
+
+var kuba = &one{
+	Item: two{
+		Map: map[string]string{
+			"foo": "bar",
+		},
+	},
+}
+
+func main() {
+	copy := getKuba()
+	copy.Item.Map["foo"] = "baz"
+
+	println(kuba.Item.Map["foo"])
+	println(copy.Item.Map["foo"])
+
+}
+
+func getKuba() one {
+	return *kuba
+}