[Feature] Agency Cache

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
 - Add ArangoBackup backoff functionality
 - Allow to abort ArangoBackup uploads by removing spec.upload
+- Add Agency Cache internally
 
 ## [1.2.5](https://github.com/arangodb/kube-arangodb/tree/1.2.5) (2021-10-25)
 - Split & Unify Lifecycle management functionality

pkg/deployment/agency.go

@@ -0,0 +1,29 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2021 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 deployment
+
+import "github.com/arangodb/kube-arangodb/pkg/metrics"
+
+var (
+	inspectDeploymentAgencyIndex   = metrics.MustRegisterGaugeVec(metricsComponent, "inspect_deployment_agency_index", "Index of the agency cache", metrics.DeploymentName)
+	inspectDeploymentAgencyFetches = metrics.MustRegisterCounterVec(metricsComponent, "inspect_deployment_agency_fetches", "Number of agency fetches", metrics.DeploymentName)
+	inspectDeploymentAgencyErrors  = metrics.MustRegisterCounterVec(metricsComponent, "inspect_deployment_agency_errors", "Number of agency errors", metrics.DeploymentName)
+)

pkg/deployment/agency/cache.go

@@ -0,0 +1,113 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2021 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 (
+	"context"
+	"sync"
+
+	"github.com/arangodb/go-driver/agency"
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+)
+
+type Cache interface {
+	Reload(ctx context.Context, client agency.Agency) (uint64, error)
+	Data() (State, bool)
+	CommitIndex() uint64
+}
+
+func NewCache(mode *api.DeploymentMode) Cache {
+	if mode.Get() == api.DeploymentModeSingle {
+		return NewSingleCache()
+	}
+
+	return NewAgencyCache()
+}
+
+func NewAgencyCache() Cache {
+	return &cache{}
+}
+
+func NewSingleCache() Cache {
+	return &cacheSingle{}
+}
+
+type cacheSingle struct {
+}
+
+func (c cacheSingle) CommitIndex() uint64 {
+	return 0
+}
+
+func (c cacheSingle) Reload(ctx context.Context, client agency.Agency) (uint64, error) {
+	return 0, nil
+}
+
+func (c cacheSingle) Data() (State, bool) {
+	return State{}, true
+}
+
+type cache struct {
+	lock sync.Mutex
+
+	valid bool
+
+	commitIndex uint64
+
+	data State
+}
+
+func (c *cache) CommitIndex() uint64 {
+	return c.commitIndex
+}
+
+func (c *cache) Data() (State, bool) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	return c.data, c.valid
+}
+
+func (c *cache) Reload(ctx context.Context, client agency.Agency) (uint64, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.valid = false
+
+	cfg, err := getAgencyConfig(ctx, client)
+	if err != nil {
+		return cfg.CommitIndex, err
+	}
+
+	if cfg.CommitIndex == c.commitIndex {
+		// We are on same index, nothing to do
+		return cfg.CommitIndex, err
+	}
+
+	if data, err := loadState(ctx, client); err != nil {
+		return cfg.CommitIndex, err
+	} else {
+		c.data = data
+		c.valid = true
+		c.commitIndex = cfg.CommitIndex
+		return cfg.CommitIndex, err
+	}
+}

pkg/deployment/agency/cache_test.go

@@ -0,0 +1,21 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2021 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

pkg/deployment/agency/maintenance.go → pkg/deployment/agency/config.go

@@ -17,60 +17,52 @@
 //
 // Copyright holder is ArangoDB GmbH, Cologne, Germany
 //
-// Author Adam Janikowski
-//
 
 package agency
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 
-	"github.com/arangodb/go-driver/agency"
-
 	"github.com/arangodb/go-driver"
+	"github.com/arangodb/go-driver/agency"
 )
 
-func GetMaintenanceMode(ctx context.Context, client agency.Agency) (bool, error) {
-	var data interface{}
-	err := client.ReadKey(ctx, []string{"arango", "Supervision", "Maintenance"}, &data)
-
-	if err == nil {
-		// We got 200
-		return true, nil
-	}
+func getAgencyConfig(ctx context.Context, client agency.Agency) (*agencyConfig, error) {
+	conn := client.Connection()
 
-	if agency.IsKeyNotFound(err) {
-		return false, nil
+	req, err := client.Connection().NewRequest(http.MethodGet, "/_api/agency/config")
+	if err != nil {
+		return nil, err
 	}
 
-	return false, err
-}
-
-func SetMaintenanceMode(ctx context.Context, client driver.Client, enabled bool) error {
-	data := "on"
-	if !enabled {
-		data = "off"
-	}
+	var data []byte
 
-	conn := client.Connection()
-	r, err := conn.NewRequest(http.MethodPut, "/_admin/cluster/maintenance")
+	resp, err := conn.Do(driver.WithRawResponse(ctx, &data), req)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	if _, err := r.SetBody(data); err != nil {
-		return err
+	if err := resp.CheckStatus(http.StatusOK); err != nil {
+		return nil, err
 	}
 
-	resp, err := conn.Do(ctx, r)
-	if err != nil {
-		return err
-	}
+	var c agencyConfig
 
-	if err := resp.CheckStatus(http.StatusOK); err != nil {
-		return err
+	if err := json.Unmarshal(data, &c); err != nil {
+		return nil, err
 	}
 
-	return nil
+	return &c, nil
+}
+
+type agencyConfig struct {
+	LeaderId string `json:"leaderId"`
+
+	CommitIndex uint64 `json:"commitIndex"`
+
+	Configuration struct {
+		ID string `json:"id"`
+	} `json:"configuration"`
 }

pkg/deployment/agency/config_test.go

@@ -0,0 +1,77 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2021 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"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_Config_Unmarshal(t *testing.T) {
+	data := `{
+  "term": 0,
+  "leaderId": "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82",
+  "commitIndex": 94,
+  "lastCompactionAt": 0,
+  "nextCompactionAfter": 500,
+  "lastAcked": {
+    "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82": {
+      "lastAckedTime": 0,
+      "lastAckedIndex": 94
+    }
+  },
+  "configuration": {
+    "pool": {
+      "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82": "tcp://[::1]:4001"
+    },
+    "active": [
+      "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82"
+    ],
+    "id": "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82",
+    "agency size": 1,
+    "pool size": 1,
+    "endpoint": "tcp://[::1]:4001",
+    "min ping": 1,
+    "max ping": 5,
+    "timeoutMult": 1,
+    "supervision": true,
+    "supervision frequency": 1,
+    "compaction step size": 500,
+    "compaction keep size": 50000,
+    "supervision grace period": 10,
+    "supervision ok threshold": 5,
+    "version": 2,
+    "startup": "origin"
+  },
+  "engine": "rocksdb",
+  "version": "3.10.0-devel"
+}`
+
+	var cfg agencyConfig
+
+	require.NoError(t, json.Unmarshal([]byte(data), &cfg))
+
+	require.Equal(t, "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82", cfg.LeaderId)
+	require.Equal(t, uint64(94), cfg.CommitIndex)
+	require.Equal(t, "AGNT-fd0f4fc7-b60b-44bb-9f5e-5fc91f708f82", cfg.Configuration.ID)
+}

pkg/deployment/agency/agency.go → pkg/deployment/agency/current_collections.go

@@ -17,31 +17,15 @@
 //
 // Copyright holder is ArangoDB GmbH, Cologne, Germany
 //
-// Author Adam Janikowski
-//
 
 package agency
 
-import (
-	"context"
-
-	"github.com/arangodb/kube-arangodb/pkg/util/errors"
-
-	"github.com/arangodb/go-driver/agency"
-)
+type StateCurrentCollections map[string]StateCurrentDBCollections
 
-type Fetcher func(ctx context.Context, i interface{}, keyParts ...string) error
+type StateCurrentDBCollections map[string]StateCurrentDBCollection
 
-func NewFetcher(a agency.Agency) Fetcher {
-	return func(ctx context.Context, i interface{}, keyParts ...string) error {
-		if err := a.ReadKey(ctx, []string{
-			ArangoKey,
-			PlanKey,
-			PlanCollectionsKey,
-		}, i); err != nil {
-			return errors.WithStack(err)
-		}
+type StateCurrentDBCollection map[string]StateCurrentDBShard
 
-		return nil
-	}
+type StateCurrentDBShard struct {
+	Servers []string `json:"servers,omitempty"`
 }