[Feature] [ACS] Improve Reconciliation loop (#983)

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
 - (Bugfix) Fix arangosync members state inspection
+- (Feature) (ACS) Improve Reconciliation Loop
 
 ## [1.2.12](https://github.com/arangodb/kube-arangodb/tree/1.2.12) (2022-05-10)
 - (Feature) Add CoreV1 Endpoints Inspector

pkg/apis/deployment/v1/cluster_synchronization_spec.go

@@ -20,6 +20,11 @@
 
 package v1
 
+import (
+	"github.com/arangodb/kube-arangodb/pkg/apis/shared"
+	"github.com/pkg/errors"
+)
+
 type ArangoClusterSynchronizationSpec struct {
 	DeploymentName string                                      `json:"deploymentName,omitempty"`
 	KubeConfig     *ArangoClusterSynchronizationKubeConfigSpec `json:"kubeconfig,omitempty"`
@@ -30,3 +35,15 @@ type ArangoClusterSynchronizationKubeConfigSpec struct {
 	SecretKey  string `json:"secretKey"`
 	Namespace  string `json:"namespace"`
 }
+
+func (a *ArangoClusterSynchronizationKubeConfigSpec) Validate() error {
+	if a == nil {
+		return errors.Errorf("KubeConfig Spec cannot be nil")
+	}
+
+	return shared.WithErrors(
+		shared.PrefixResourceError("secretName", shared.ValidateResourceName(a.SecretName)),
+		shared.PrefixResourceError("secretKey", shared.ValidateResourceName(a.SecretKey)),
+		shared.PrefixResourceError("namespace", shared.ValidateResourceName(a.Namespace)),
+	)
+}

pkg/apis/deployment/v1/cluster_synchronization_spec_test.go

@@ -0,0 +1,93 @@
+//
+// 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 v1
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_ACS_KubeConfigSpec(t *testing.T) {
+	test := func(t *testing.T, spec *ArangoClusterSynchronizationKubeConfigSpec, error error) {
+		err := spec.Validate()
+
+		if error != nil {
+			require.EqualError(t, err, error.Error())
+		} else {
+			require.NoError(t, err)
+		}
+	}
+
+	type testCase struct {
+		spec  *ArangoClusterSynchronizationKubeConfigSpec
+		error string
+	}
+
+	testCases := map[string]testCase{
+		"Nil": {
+			error: "KubeConfig Spec cannot be nil",
+		},
+		"Empty": {
+			spec:  &ArangoClusterSynchronizationKubeConfigSpec{},
+			error: "Received 3 errors: secretName: Name '' is not a valid resource name, secretKey: Name '' is not a valid resource name, namespace: Name '' is not a valid resource name",
+		},
+		"Missing key & NS": {
+			spec: &ArangoClusterSynchronizationKubeConfigSpec{
+				SecretName: "secret",
+			},
+			error: "Received 2 errors: secretKey: Name '' is not a valid resource name, namespace: Name '' is not a valid resource name",
+		},
+		"Missing NS": {
+			spec: &ArangoClusterSynchronizationKubeConfigSpec{
+				SecretName: "secret",
+				SecretKey:  "key",
+			},
+			error: "Received 1 errors: namespace: Name '' is not a valid resource name",
+		},
+		"Valid": {
+			spec: &ArangoClusterSynchronizationKubeConfigSpec{
+				SecretName: "secret",
+				SecretKey:  "key",
+				Namespace:  "ns",
+			},
+		},
+		"Invalid": {
+			spec: &ArangoClusterSynchronizationKubeConfigSpec{
+				SecretName: "secret_n",
+				SecretKey:  "key",
+				Namespace:  "ns",
+			},
+			error: "Received 1 errors: secretName: Name 'secret_n' is not a valid resource name",
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			var err error
+			if tc.error != "" {
+				err = errors.Errorf(tc.error)
+			}
+			test(t, tc.spec, err)
+		})
+	}
+}

pkg/apis/deployment/v1/cluster_synchronization_status.go

@@ -23,8 +23,10 @@ package v1
 import "k8s.io/apimachinery/pkg/types"
 
 type ArangoClusterSynchronizationStatus struct {
-	Deployment *ArangoClusterSynchronizationDeploymentStatus `json:"deployment,omitempty"`
-	Conditions ConditionList                                 `json:"conditions,omitempty"`
+	Deployment       *ArangoClusterSynchronizationDeploymentStatus `json:"deployment,omitempty"`
+	RemoteDeployment *ArangoClusterSynchronizationDeploymentStatus `json:"remoteDeployment,omitempty"`
+
+	Conditions ConditionList `json:"conditions,omitempty"`
 }
 
 type ArangoClusterSynchronizationDeploymentStatus struct {

pkg/apis/deployment/v1/zz_generated.deepcopy.go

@@ -213,6 +213,11 @@ func (in *ArangoClusterSynchronizationStatus) DeepCopyInto(out *ArangoClusterSyn
 		*out = new(ArangoClusterSynchronizationDeploymentStatus)
 		**out = **in
 	}
+	if in.RemoteDeployment != nil {
+		in, out := &in.RemoteDeployment, &out.RemoteDeployment
+		*out = new(ArangoClusterSynchronizationDeploymentStatus)
+		**out = **in
+	}
 	if in.Conditions != nil {
 		in, out := &in.Conditions, &out.Conditions
 		*out = make(ConditionList, len(*in))

pkg/apis/shared/names_test.go

@@ -0,0 +1,33 @@
+//
+// 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 shared
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_Names(t *testing.T) {
+	t.Run("Empty", func(t *testing.T) {
+		require.EqualError(t, ValidateResourceName(""), "Name '' is not a valid resource name")
+	})
+}

pkg/deployment/acs/sutil/conditions.go

@@ -23,6 +23,9 @@ package sutil
 import api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
 
 const (
-	DeploymentReadyCondition     api.ConditionType = "DeploymentReady"
-	KubernetesConnectedCondition api.ConditionType = "KubernetesConnected"
+	DeploymentReadyCondition       api.ConditionType = "DeploymentReady"
+	KubernetesConnectedCondition   api.ConditionType = "KubernetesConnected"
+	RemoteDeploymentReadyCondition api.ConditionType = "RemoteDeploymentReadyCondition"
+	RemoteCacheReadyCondition      api.ConditionType = "RemoteCacheReady"
+	ConnectionReadyCondition       api.ConditionType = "ConnectionReady"
 )

pkg/deployment/deployment.go

@@ -63,7 +63,6 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/deployment/resources"
 	"github.com/arangodb/kube-arangodb/pkg/util"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
-	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/throttle"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
 	"github.com/arangodb/kube-arangodb/pkg/util/trigger"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -217,22 +216,6 @@ func (d *Deployment) SetAgencyMaintenanceMode(ctx context.Context, enabled bool)
 	return nil
 }
 
-func newDeploymentThrottle() throttle.Components {
-	return throttle.NewThrottleComponents(
-		30*time.Second, // ArangoDeploymentSynchronization
-		30*time.Second, // ArangoMember
-		30*time.Second, // ArangoTask
-		30*time.Second, // Node
-		15*time.Second, // PVC
-		time.Second,    // Pod
-		30*time.Second, // PDB
-		10*time.Second, // Secret
-		10*time.Second, // Service
-		30*time.Second, // SA
-		30*time.Second, // ServiceMonitor
-		15*time.Second) // Endpoints
-}
-
 // New creates a new Deployment from the given API object.
 func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*Deployment, error) {
 	if err := apiObject.Spec.Validate(); err != nil {
@@ -248,7 +231,7 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*De
 		eventCh:      make(chan *deploymentEvent, deploymentEventQueueSize),
 		stopCh:       make(chan struct{}),
 		agencyCache:  agency.NewCache(apiObject.Spec.Mode),
-		currentState: inspector.NewInspector(newDeploymentThrottle(), deps.Client, apiObject.GetNamespace(), apiObject.GetName()),
+		currentState: inspector.NewInspector(inspector.NewDefaultThrottle(), deps.Client, apiObject.GetNamespace(), apiObject.GetName()),
 		acs:          acs.NewACS(),
 	}
 

pkg/deployment/resources/inspector/inspector.go

@@ -245,6 +245,13 @@ func (i *inspectorState) Client() kclient.Client {
 	return i.client
 }
 
+func (i *inspectorState) SetClient(k kclient.Client) {
+	i.lock.Lock()
+	defer i.lock.Unlock()
+
+	i.client = k
+}
+
 func (i *inspectorState) Namespace() string {
 	return i.namespace
 }

pkg/deployment/resources/inspector/throttles.go

@@ -0,0 +1,43 @@
+//
+// 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 inspector
+
+import (
+	"time"
+
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/throttle"
+)
+
+func NewDefaultThrottle() throttle.Components {
+	return throttle.NewThrottleComponents(
+		30*time.Second, // ArangoDeploymentSynchronization
+		30*time.Second, // ArangoMember
+		30*time.Second, // ArangoTask
+		30*time.Second, // Node
+		15*time.Second, // PVC
+		time.Second,    // Pod
+		30*time.Second, // PDB
+		10*time.Second, // Secret
+		10*time.Second, // Service
+		30*time.Second, // SA
+		30*time.Second, // ServiceMonitor
+		15*time.Second) // Endpoints
+}

pkg/util/k8sutil/inspector/inspector.go

@@ -53,7 +53,9 @@ type Object interface {
 }
 
 type Inspector interface {
+	SetClient(k kclient.Client)
 	Client() kclient.Client
+
 	Namespace() string
 
 	Initialised() bool

pkg/util/kclient/client_factory.go

@@ -32,6 +32,10 @@ import (
 	"k8s.io/client-go/rest"
 )
 
+const (
+	unattachedFactoryID = "unattached-factory-id"
+)
+
 var (
 	factories     = map[string]*factory{}
 	factoriesLock sync.Mutex
@@ -66,6 +70,12 @@ func GetFactory(name string) Factory {
 	return factories[name]
 }
 
+func GetUnattachedFactory() Factory {
+	return &factory{
+		name: unattachedFactoryID,
+	}
+}
+
 type ConfigGetter func() (*rest.Config, string, error)
 
 func NewStaticConfigGetter(f func() (*rest.Config, error)) ConfigGetter {
@@ -135,7 +145,11 @@ func (f *factory) refresh() error {
 		return nil
 	}
 
-	cfg.RateLimiter = GetRateLimiter(f.name)
+	if f.name == unattachedFactoryID {
+		cfg.RateLimiter = GetUnattachedRateLimiter()
+	} else {
+		cfg.RateLimiter = GetRateLimiter(f.name)
+	}
 
 	client, err := newClient(cfg)
 	if err != nil {

pkg/util/kclient/ratelimiter.go

@@ -64,6 +64,14 @@ func GetRateLimiter(name string) flowcontrol.RateLimiter {
 	return l
 }
 
+func GetUnattachedRateLimiter() flowcontrol.RateLimiter {
+	return &rateLimiter{
+		limiter: rate.NewLimiter(rate.Limit(defaultQPS), defaultBurst),
+		clock:   clock{},
+		qps:     defaultQPS,
+	}
+}
+
 func SetDefaultBurst(q int) {
 	rateLimitersLock.Lock()
 	defer rateLimitersLock.Unlock()