[BUGFIX] Do not remove used DBServer (#515)

Author: ajanikow

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Change Log
 
 ## Master
+- Prevent DBServer deletion if there are any shards active on it
 - Add Maintenance mode annotation for ArangoDeployment
 
 ## [0.4.2](https://github.com/arangodb/kube-arangodb/tree/0.4.2) (2019-11-12)

pkg/apis/deployment/v1/plan.go

@@ -36,6 +36,8 @@ const (
 	ActionTypeAddMember ActionType = "AddMember"
 	// ActionTypeRemoveMember causes a member to be removed.
 	ActionTypeRemoveMember ActionType = "RemoveMember"
+	// ActionTypeRecreateMember recreates member. Used when member is still owner of some shards.
+	ActionTypeRecreateMember ActionType = "RecreateMember"
 	// ActionTypeCleanOutMember causes a member to be cleaned out (dbserver only).
 	ActionTypeCleanOutMember ActionType = "CleanOutMember"
 	// ActionTypeShutdownMember causes a member to be shutdown and removed from the cluster.

pkg/deployment/agency/agency.go

@@ -0,0 +1,62 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+// Author Adam Janikowski
+//
+
+package agency
+
+type ArangoPlanDatabases map[string]ArangoPlanCollections
+
+func (a ArangoPlanDatabases) IsDBServerInDatabases(name string) bool {
+	for _, collections := range a {
+		if collections.IsDBServerInCollections(name) {
+			return true
+		}
+	}
+	return false
+}
+
+type ArangoPlanCollections map[string]ArangoPlanCollection
+
+func (a ArangoPlanCollections) IsDBServerInCollections(name string) bool {
+	for _, collection := range a {
+		if collection.IsDBServerInShards(name) {
+			return true
+		}
+	}
+	return false
+}
+
+type ArangoPlanCollection struct {
+	Shards ArangoPlanShard `json:"shards"`
+}
+
+func (a ArangoPlanCollection) IsDBServerInShards(name string) bool {
+	for _, dbservers := range a.Shards {
+		for _, dbserver := range dbservers {
+			if dbserver == name {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+type ArangoPlanShard map[string][]string

pkg/deployment/agency/definitions.go

@@ -0,0 +1,29 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+// Author Adam Janikowski
+//
+
+package agency
+
+const (
+	ArangoKey = "arango"
+	PlanKey = "Plan"
+	PlanCollectionsKey = "Collections"
+)

pkg/deployment/context_impl.go

@@ -426,3 +426,17 @@ func (d *Deployment) DisableScalingCluster() error {
 func (d *Deployment) EnableScalingCluster() error {
 	return d.clusterScalingIntegration.EnableScalingCluster()
 }
+
+// GetAgencyPlan returns agency plan
+func (d *Deployment) GetAgencyData(ctx context.Context, i interface{}, keyParts ... string) error {
+	a, err := d.GetAgency(ctx)
+	if err != nil {
+		return err
+	}
+
+	if err = a.ReadKey(ctx, keyParts, i); err != nil {
+		return err
+	}
+
+	return err
+}

pkg/deployment/reconcile/action_context.go

@@ -25,6 +25,7 @@ package reconcile
 import (
 	"context"
 	"fmt"
+	v1 "k8s.io/api/core/v1"
 
 	"github.com/arangodb/go-driver/agency"
 
@@ -70,6 +71,9 @@ type ActionContext interface {
 	// DeletePvc deletes a persistent volume claim with given name in the namespace
 	// of the deployment. If the pvc does not exist, the error is ignored.
 	DeletePvc(pvcName string) error
+	// GetPvc returns PVC info about PVC with given name in the namespace
+	// of the deployment.
+	GetPvc(pvcName string) (*v1.PersistentVolumeClaim, error)
 	// RemovePodFinalizers removes all the finalizers from the Pod with given name in the namespace
 	// of the deployment. If the pod does not exist, the error is ignored.
 	RemovePodFinalizers(podName string) error
@@ -110,6 +114,10 @@ type actionContext struct {
 	context Context
 }
 
+func (ac *actionContext) GetPvc(pvcName string) (*v1.PersistentVolumeClaim, error) {
+	return ac.context.GetPvc(pvcName)
+}
+
 // Gets the specified mode of deployment
 func (ac *actionContext) GetMode() api.DeploymentMode {
 	return ac.context.GetSpec().GetMode()

pkg/deployment/reconcile/action_recreate_member.go

@@ -0,0 +1,98 @@
+//
+// DISCLAIMER
+//
+// Copyright 2020 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
+//
+// Author Adam Janikowski
+//
+
+package reconcile
+
+import (
+	"context"
+	"fmt"
+	kubeErrors "k8s.io/apimachinery/pkg/api/errors"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+)
+
+// NewRecreateMemberAction creates a new Action that implements the given
+// planned RecreateMember action.
+func NewRecreateMemberAction(log zerolog.Logger, action api.Action, actionCtx ActionContext) Action {
+	return &actionRecreateMember{
+		log:       log,
+		action:    action,
+		actionCtx: actionCtx,
+	}
+}
+
+// actionRecreateMember implements an RecreateMemberAction.
+type actionRecreateMember struct {
+	log       zerolog.Logger
+	action    api.Action
+	actionCtx ActionContext
+}
+
+// Start performs the start of the action.
+// Returns true if the action is completely finished, false in case
+// the start time needs to be recorded and a ready condition needs to be checked.
+func (a *actionRecreateMember) Start(ctx context.Context) (bool, error) {
+	m, ok := a.actionCtx.GetMemberStatusByID(a.action.MemberID)
+	if !ok {
+		return false, fmt.Errorf("expecting member to be present in list, but it is not")
+	}
+
+	_, err := a.actionCtx.GetPvc(m.PersistentVolumeClaimName)
+	if err != nil {
+		if kubeErrors.IsNotFound(err) {
+			return false, fmt.Errorf("PVC is missing %s. DBServer wont be recreated without old PV", m.PersistentVolumeClaimName)
+		}
+
+		return false, maskAny(err)
+	}
+
+	if m.Phase == api.MemberPhaseFailed {
+		// Change cluster phase to ensure it wont be removed
+		m.Phase = api.MemberPhaseNone
+	}
+
+	if err = a.actionCtx.UpdateMember(m); err != nil {
+		return false, maskAny(err)
+	}
+
+	return true, nil
+}
+
+// CheckProgress checks the progress of the action.
+// Returns true if the action is completely finished, false otherwise.
+func (a *actionRecreateMember) CheckProgress(ctx context.Context) (bool, bool, error) {
+	// Nothing todo
+	return true, false, nil
+}
+
+// Timeout returns the amount of time after which this action will timeout.
+func (a *actionRecreateMember) Timeout() time.Duration {
+	return recreateMemberTimeout
+}
+
+// Return the MemberID used / created in this action
+func (a *actionRecreateMember) MemberID() string {
+	return a.action.MemberID
+}

pkg/deployment/reconcile/context.go

@@ -24,7 +24,6 @@ package reconcile
 
 import (
 	"context"
-
 	"github.com/arangodb/arangosync-client/client"
 	driver "github.com/arangodb/go-driver"
 	"github.com/arangodb/go-driver/agency"
@@ -105,4 +104,8 @@ type Context interface {
 	DisableScalingCluster() error
 	// EnableScalingCluster enables scaling DBservers and coordinators
 	EnableScalingCluster() error
+	// GetAgencyData object for key path
+	GetAgencyData(ctx context.Context, i interface{}, keyParts ... string) error
 }
+
+

pkg/deployment/reconcile/plan_builder.go

@@ -23,6 +23,11 @@
 package reconcile
 
 import (
+	goContext "context"
+	"fmt"
+	"github.com/arangodb/kube-arangodb/pkg/deployment/agency"
+	"time"
+
 	driver "github.com/arangodb/go-driver"
 	upgraderules "github.com/arangodb/go-upgrade-rules"
 	"github.com/rs/zerolog"
@@ -78,6 +83,27 @@ func (d *Reconciler) CreatePlan() error {
 	return nil
 }
 
+func fetchAgency(log zerolog.Logger,
+	spec api.DeploymentSpec, status api.DeploymentStatus,
+	context PlanBuilderContext) (*agency.ArangoPlanDatabases,  error) {
+	if spec.GetMode() != api.DeploymentModeCluster && spec.GetMode() != api.DeploymentModeActiveFailover {
+		return nil, nil
+	} else if status.Members.Agents.MembersReady() > 0 {
+		agencyCtx, agencyCancel := goContext.WithTimeout(goContext.Background(), time.Minute)
+		defer agencyCancel()
+
+		ret := &agency.ArangoPlanDatabases{}
+
+		if err := context.GetAgencyData(agencyCtx, ret, agency.ArangoKey, agency.PlanKey, agency.PlanCollectionsKey); err != nil {
+			return nil, err
+		}
+
+		return ret, nil
+	} else {
+		return nil, fmt.Errorf("not able to read from agency when agency is down")
+	}
+}
+
 // createPlan considers the given specification & status and creates a plan to get the status in line with the specification.
 // If a plan already exists, the given plan is returned with false.
 // Otherwise the new plan is returned with a boolean true.
@@ -90,30 +116,63 @@ func createPlan(log zerolog.Logger, apiObject k8sutil.APIObject,
 		return currentPlan, false
 	}
 
+	// Fetch agency plan
+	agencyPlan, agencyErr := fetchAgency(log, spec, status, context)
+
 	// Check for various scenario's
 	var plan api.Plan
 
 	// Check for members in failed state
 	status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
 		for _, m := range members {
-			if m.Phase == api.MemberPhaseFailed && plan.IsEmpty() {
-				log.Debug().
-					Str("id", m.ID).
-					Str("role", group.AsRole()).
-					Msg("Creating member replacement plan because member has failed")
-				newID := ""
-				if group == api.ServerGroupAgents {
-					newID = m.ID // Agents cannot (yet) be replaced with new IDs
+			if m.Phase != api.MemberPhaseFailed || len(plan) > 0 {
+				continue
+			}
+
+			memberLog := log.Info().Str("id", m.ID).Str("role", group.AsRole())
+
+			if group == api.ServerGroupDBServers && spec.GetMode() == api.DeploymentModeCluster {
+				// Do pre check for DBServers. If agency is down DBServers should not be touch
+				if agencyErr != nil {
+					memberLog.Msg("Error in agency")
+					continue
 				}
-				plan = append(plan,
-					api.NewAction(api.ActionTypeRemoveMember, group, m.ID),
-					api.NewAction(api.ActionTypeAddMember, group, newID),
-				)
+
+				if agencyPlan == nil {
+					memberLog.Msg("AgencyPlan is nil")
+					continue
+				}
+
+				if agencyPlan.IsDBServerInDatabases(m.ID) {
+					// DBServer still exists in agency plan! Will not be removed, but needs to be recreated
+					memberLog.Msg("Recreating DBServer - it cannot be removed gracefully")
+					plan = append(plan,
+						api.NewAction(api.ActionTypeRecreateMember, group, m.ID))
+					continue
+				}
+
+				// Everything is fine, proceed
 			}
+
+			memberLog.Msg("Creating member replacement plan because member has failed")
+			newID := ""
+			if group == api.ServerGroupAgents {
+				newID = m.ID // Agents cannot (yet) be replaced with new IDs
+			}
+			plan = append(plan,
+				api.NewAction(api.ActionTypeRemoveMember, group, m.ID),
+				api.NewAction(api.ActionTypeAddMember, group, newID),
+			)
 		}
 		return nil
 	})
 
+	// Ensure that we were able to get agency info
+	if len(plan) == 0 && agencyErr != nil {
+		log.Err(agencyErr).Msg("unable to build further plan without access to agency")
+		return plan, false
+	}
+
 	// Check for cleaned out dbserver in created state
 	for _, m := range status.Members.DBServers {
 		if plan.IsEmpty() && m.Phase.IsCreatedOrDrain() && m.Conditions.IsTrue(api.ConditionTypeCleanedOut) {