[Feature] [ACS] Add resource plan (#986)

CHANGELOG.md

@@ -4,6 +4,7 @@
 - (Bugfix) Fix arangosync members state inspection
 - (Feature) (ACS) Improve Reconciliation Loop
 - (Bugfix) Allow missing Monitoring CRD
+- (Feature) (ACS) Add Resource plan
 
 ## [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/deployment_status.go

@@ -64,6 +64,9 @@ type DeploymentStatus struct {
 	// HighPriorityPlan to update this deployment. Executed before plan
 	HighPriorityPlan Plan `json:"highPriorityPlan,omitempty"`
 
+	// ResourcesPlan to update this deployment. Executed before plan, after highPlan
+	ResourcesPlan Plan `json:"resourcesPlan,omitempty"`
+
 	// AcceptedSpec contains the last specification that was accepted by the operator.
 	AcceptedSpec *DeploymentSpec `json:"accepted-spec,omitempty"`
 
@@ -103,6 +106,8 @@ func (ds *DeploymentStatus) Equal(other DeploymentStatus) bool {
 		ds.Members.Equal(other.Members) &&
 		ds.Conditions.Equal(other.Conditions) &&
 		ds.Plan.Equal(other.Plan) &&
+		ds.HighPriorityPlan.Equal(other.HighPriorityPlan) &&
+		ds.ResourcesPlan.Equal(other.ResourcesPlan) &&
 		ds.AcceptedSpec.Equal(other.AcceptedSpec) &&
 		ds.SecretHashes.Equal(other.SecretHashes) &&
 		ds.Agency.Equal(other.Agency) &&

pkg/apis/deployment/v1/plan.go

@@ -192,6 +192,9 @@ const (
 	// Rebalancer
 	ActionTypeRebalancerGenerate ActionType = "RebalancerGenerate"
 	ActionTypeRebalancerCheck    ActionType = "RebalancerCheck"
+
+	// Resources
+	ActionTypeResourceSync ActionType = "ResourceSync"
 )
 
 const (

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

@@ -1133,6 +1133,13 @@ func (in *DeploymentStatus) DeepCopyInto(out *DeploymentStatus) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.ResourcesPlan != nil {
+		in, out := &in.ResourcesPlan, &out.ResourcesPlan
+		*out = make(Plan, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	if in.AcceptedSpec != nil {
 		in, out := &in.AcceptedSpec, &out.AcceptedSpec
 		*out = new(DeploymentSpec)

pkg/apis/deployment/v2alpha1/cluster_synchronization_spec.go

@@ -20,6 +20,11 @@
 
 package v2alpha1
 
+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/v2alpha1/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 v2alpha1
+
+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/v2alpha1/cluster_synchronization_status.go

@@ -23,8 +23,10 @@ package v2alpha1
 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/v2alpha1/deployment_status.go

@@ -64,6 +64,9 @@ type DeploymentStatus struct {
 	// HighPriorityPlan to update this deployment. Executed before plan
 	HighPriorityPlan Plan `json:"highPriorityPlan,omitempty"`
 
+	// ResourcesPlan to update this deployment. Executed before plan, after highPlan
+	ResourcesPlan Plan `json:"resourcesPlan,omitempty"`
+
 	// AcceptedSpec contains the last specification that was accepted by the operator.
 	AcceptedSpec *DeploymentSpec `json:"accepted-spec,omitempty"`
 
@@ -103,6 +106,8 @@ func (ds *DeploymentStatus) Equal(other DeploymentStatus) bool {
 		ds.Members.Equal(other.Members) &&
 		ds.Conditions.Equal(other.Conditions) &&
 		ds.Plan.Equal(other.Plan) &&
+		ds.HighPriorityPlan.Equal(other.HighPriorityPlan) &&
+		ds.ResourcesPlan.Equal(other.ResourcesPlan) &&
 		ds.AcceptedSpec.Equal(other.AcceptedSpec) &&
 		ds.SecretHashes.Equal(other.SecretHashes) &&
 		ds.Agency.Equal(other.Agency) &&

pkg/apis/deployment/v2alpha1/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))
@@ -1128,6 +1133,13 @@ func (in *DeploymentStatus) DeepCopyInto(out *DeploymentStatus) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.ResourcesPlan != nil {
+		in, out := &in.ResourcesPlan, &out.ResourcesPlan
+		*out = make(Plan, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 	if in.AcceptedSpec != nil {
 		in, out := &in.AcceptedSpec, &out.AcceptedSpec
 		*out = new(DeploymentSpec)

pkg/deployment/acs/acs.community.go

@@ -28,13 +28,39 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/deployment/acs/sutil"
 	inspectorInterface "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
+	"k8s.io/apimachinery/pkg/types"
 )
 
-func NewACS() sutil.ACS {
-	return acs{}
+func NewACS(main types.UID, cache inspectorInterface.Inspector) sutil.ACS {
+	return acs{
+		main:  main,
+		cache: cache,
+	}
 }
 
 type acs struct {
+	main  types.UID
+	cache inspectorInterface.Inspector
+}
+
+func (a acs) UID() types.UID {
+	return a.main
+}
+
+func (a acs) Cache() inspectorInterface.Inspector {
+	return a.cache
+}
+
+func (a acs) Cluster(uid types.UID) (sutil.ACSItem, bool) {
+	if a.main == uid || uid == "" {
+		return a, true
+	}
+
+	return nil, false
+}
+
+func (a acs) RemoteClusters() []types.UID {
+	return nil
 }
 
 func (a acs) Inspect(ctx context.Context, deployment *api.ArangoDeployment, client kclient.Client, cachedStatus inspectorInterface.Inspector) error {

pkg/deployment/acs/sutil/interfaces.go

@@ -26,8 +26,20 @@ import (
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
 	inspectorInterface "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 type ACS interface {
+	ACSItem
+
 	Inspect(ctx context.Context, deployment *api.ArangoDeployment, client kclient.Client, cachedStatus inspectorInterface.Inspector) error
+
+	Cluster(uid types.UID) (ACSItem, bool)
+
+	RemoteClusters() []types.UID
+}
+
+type ACSItem interface {
+	UID() types.UID
+	Cache() inspectorInterface.Inspector
 }

pkg/deployment/deployment.go

@@ -221,6 +221,8 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*De
 		return nil, errors.WithStack(err)
 	}
 
+	i := inspector.NewInspector(inspector.NewDefaultThrottle(), deps.Client, apiObject.GetNamespace(), apiObject.GetName())
+
 	d := &Deployment{
 		apiObject:    apiObject,
 		name:         apiObject.GetName(),
@@ -230,8 +232,8 @@ 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(inspector.NewDefaultThrottle(), deps.Client, apiObject.GetNamespace(), apiObject.GetName()),
-		acs:          acs.NewACS(),
+		currentState: i,
+		acs:          acs.NewACS(apiObject.GetUID(), i),
 	}
 
 	d.memberState = memberState.NewStateInspector(d)

pkg/deployment/reconcile/plan_builder.go

@@ -42,5 +42,5 @@ const (
 // get the status in line with the specification.
 // If a plan already exists, nothing is done.
 func (d *Reconciler) CreatePlan(ctx context.Context, cachedStatus inspectorInterface.Inspector) (error, bool) {
-	return d.generatePlan(ctx, cachedStatus, d.generatePlanFunc(createHighPlan, plannerHigh{}), d.generatePlanFunc(createNormalPlan, plannerNormal{}))
+	return d.generatePlan(ctx, cachedStatus, d.generatePlanFunc(createHighPlan, plannerHigh{}), d.generatePlanFunc(createResourcesPlan, plannerResources{}), d.generatePlanFunc(createNormalPlan, plannerNormal{}))
 }

pkg/deployment/reconcile/plan_builder_resources.go

@@ -0,0 +1,44 @@
+//
+// 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 reconcile
+
+import (
+	"context"
+
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+	inspectorInterface "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector"
+	"github.com/rs/zerolog"
+)
+
+func createResourcesPlan(ctx context.Context, log zerolog.Logger, apiObject k8sutil.APIObject,
+	currentPlan api.Plan, spec api.DeploymentSpec,
+	status api.DeploymentStatus, cachedStatus inspectorInterface.Inspector,
+	builderCtx PlanBuilderContext) (api.Plan, api.BackOff, bool) {
+	if !currentPlan.IsEmpty() {
+		// Plan already exists, complete that first
+		return currentPlan, nil, false
+	}
+
+	r := recoverPlanAppender(log, newPlanAppender(NewWithPlanBuilder(ctx, log, apiObject, spec, status, cachedStatus, builderCtx), status.BackOff, currentPlan))
+
+	return r.Plan(), r.BackOff(), true
+}

pkg/deployment/reconcile/plan_executor.go

@@ -94,6 +94,26 @@ func (p plannerHigh) Set(deployment *api.DeploymentStatus, plan api.Plan) bool {
 	return false
 }
 
+type plannerResources struct {
+}
+
+func (p plannerResources) Type() string {
+	return "resources"
+}
+
+func (p plannerResources) Get(deployment *api.DeploymentStatus) api.Plan {
+	return deployment.ResourcesPlan
+}
+
+func (p plannerResources) Set(deployment *api.DeploymentStatus, plan api.Plan) bool {
+	if !deployment.ResourcesPlan.Equal(plan) {
+		deployment.ResourcesPlan = plan
+		return true
+	}
+
+	return false
+}
+
 // ExecutePlan tries to execute the plan as far as possible.
 // Returns true when it has to be called again soon.
 // False otherwise.
@@ -106,6 +126,13 @@ func (d *Reconciler) ExecutePlan(ctx context.Context, cachedStatus inspectorInte
 		callAgain = true
 	}
 
+	if again, err := d.executePlanStatus(ctx, cachedStatus, d.log, plannerResources{}); err != nil {
+		d.log.Error().Err(err).Msg("Execution of plan failed")
+		return false, nil
+	} else if again {
+		callAgain = true
+	}
+
 	if again, err := d.executePlanStatus(ctx, cachedStatus, d.log, plannerNormal{}); err != nil {
 		return false, errors.WithStack(err)
 	} else if again {