Preparing pods for termination when PVC is deleted

pkg/deployment/reconcile/plan_builder.go

@@ -101,6 +101,10 @@ func createPlan(log zerolog.Logger, apiObject k8sutil.APIObject,
 	status.Members.ForeachServerGroup(func(group api.ServerGroup, members api.MemberStatusList) error {
 		for _, m := range members {
 			if m.Phase == api.MemberPhaseFailed && len(plan) == 0 {
+				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
@@ -117,6 +121,10 @@ func createPlan(log zerolog.Logger, apiObject k8sutil.APIObject,
 	// Check for cleaned out dbserver in created state
 	for _, m := range status.Members.DBServers {
 		if len(plan) == 0 && m.Phase == api.MemberPhaseCreated && m.Conditions.IsTrue(api.ConditionTypeCleanedOut) {
+			log.Debug().
+				Str("id", m.ID).
+				Str("role", api.ServerGroupDBServers.AsRole()).
+				Msg("Creating dbserver replacement plan because server is cleanout in created phase")
 			plan = append(plan,
 				api.NewAction(api.ActionTypeRemoveMember, api.ServerGroupDBServers, m.ID),
 				api.NewAction(api.ActionTypeAddMember, api.ServerGroupDBServers, ""),

pkg/deployment/resources/pod_finalizers.go

@@ -31,7 +31,6 @@ import (
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"github.com/arangodb/go-driver/agency"
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
@@ -84,68 +83,13 @@ func (r *Resources) runPodFinalizers(ctx context.Context, p *v1.Pod, memberStatu
 // inspectFinalizerPodAgencyServing checks the finalizer condition for agency-serving.
 // It returns nil if the finalizer can be removed.
 func (r *Resources) inspectFinalizerPodAgencyServing(ctx context.Context, log zerolog.Logger, p *v1.Pod, memberStatus api.MemberStatus) error {
-	// Inspect member phase
-	if memberStatus.Phase.IsFailed() {
-		log.Debug().Msg("Pod is already failed, safe to remove agency serving finalizer")
-		return nil
-	}
-	// Inspect deployment deletion state
-	apiObject := r.context.GetAPIObject()
-	if apiObject.GetDeletionTimestamp() != nil {
-		log.Debug().Msg("Entire deployment is being deleted, safe to remove agency serving finalizer")
-		return nil
-	}
-
-	// Check node the pod is scheduled on
-	agentDataWillBeGone := false
-	if p.Spec.NodeName != "" {
-		node, err := r.context.GetKubeCli().CoreV1().Nodes().Get(p.Spec.NodeName, metav1.GetOptions{})
-		if err != nil {
-			log.Warn().Err(err).Msg("Failed to get node for member")
-			return maskAny(err)
-		}
-		if node.Spec.Unschedulable {
-			agentDataWillBeGone = true
-		}
-	}
-
-	// Check PVC
-	pvcs := r.context.GetKubeCli().CoreV1().PersistentVolumeClaims(apiObject.GetNamespace())
-	pvc, err := pvcs.Get(memberStatus.PersistentVolumeClaimName, metav1.GetOptions{})
-	if err != nil {
-		log.Warn().Err(err).Msg("Failed to get PVC for member")
-		return maskAny(err)
-	}
-	if k8sutil.IsPersistentVolumeClaimMarkedForDeletion(pvc) {
-		agentDataWillBeGone = true
-	}
-
-	// Is this a simple pod restart?
-	if !agentDataWillBeGone {
-		log.Debug().Msg("Pod is just being restarted, safe to remove agency serving finalizer")
-		return nil
-	}
-
-	// Inspect agency state
-	log.Debug().Msg("Agent data will be gone, so we will check agency serving status first")
-	ctx = agency.WithAllowNoLeader(ctx)                     // The ID we're checking may be the leader, so ignore situations where all other agents are followers
-	ctx, cancel := context.WithTimeout(ctx, time.Second*15) // Force a quick check
-	defer cancel()
-	agencyConns, err := r.context.GetAgencyClients(ctx, func(id string) bool { return id != memberStatus.ID })
-	if err != nil {
-		log.Debug().Err(err).Msg("Failed to create member client")
-		return maskAny(err)
-	}
-	if len(agencyConns) == 0 {
-		log.Debug().Err(err).Msg("No more remaining agents, we cannot delete this one")
-		return maskAny(fmt.Errorf("No more remaining agents"))
-	}
-	if err := agency.AreAgentsHealthy(ctx, agencyConns); err != nil {
-		log.Debug().Err(err).Msg("Remaining agents are not healthy")
+	if err := r.prepareAgencyPodTermination(ctx, log, p, memberStatus); err != nil {
+		// Pod cannot be terminated yet
 		return maskAny(err)
 	}
 
 	// Remaining agents are healthy, we can remove this one and trigger a delete of the PVC
+	pvcs := r.context.GetKubeCli().CoreV1().PersistentVolumeClaims(r.context.GetNamespace())
 	if err := pvcs.Delete(memberStatus.PersistentVolumeClaimName, &metav1.DeleteOptions{}); err != nil && !k8sutil.IsNotFound(err) {
 		log.Warn().Err(err).Msg("Failed to delete PVC for member")
 		return maskAny(err)

pkg/deployment/resources/pod_termination.go

@@ -0,0 +1,213 @@
+//
+// DISCLAIMER
+//
+// Copyright 2018 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 Ewout Prangsma
+//
+
+package resources
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/rs/zerolog"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/arangodb/go-driver/agency"
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+)
+
+// preparePodTermination checks if the given pod is allowed to terminate and if so,
+// prepares it for termination.
+// It returns nil if the pod is allowed to terminate yet, an error otherwise.
+func (r *Resources) preparePodTermination(ctx context.Context, log zerolog.Logger, p *v1.Pod, group api.ServerGroup, memberStatus api.MemberStatus, updateMember func(api.MemberStatus) error) error {
+	var err error
+	switch group {
+	case api.ServerGroupAgents:
+		err = r.prepareAgencyPodTermination(ctx, log, p, memberStatus)
+	case api.ServerGroupDBServers:
+		err = r.prepareDBServerPodTermination(ctx, log, p, memberStatus, updateMember)
+	default:
+		err = nil
+	}
+	return maskAny(err)
+}
+
+// prepareAgencyPodTermination checks if the given agency pod is allowed to terminate
+// and if so, prepares it for termination.
+// It returns nil if the pod is allowed to terminate, an error otherwise.
+func (r *Resources) prepareAgencyPodTermination(ctx context.Context, log zerolog.Logger, p *v1.Pod, memberStatus api.MemberStatus) error {
+	// Inspect member phase
+	if memberStatus.Phase.IsFailed() {
+		log.Debug().Msg("Pod is already failed, safe to remove agency serving finalizer")
+		return nil
+	}
+	// Inspect deployment deletion state
+	apiObject := r.context.GetAPIObject()
+	if apiObject.GetDeletionTimestamp() != nil {
+		log.Debug().Msg("Entire deployment is being deleted, safe to remove agency serving finalizer")
+		return nil
+	}
+
+	// Check node the pod is scheduled on
+	agentDataWillBeGone := false
+	if p.Spec.NodeName != "" {
+		node, err := r.context.GetKubeCli().CoreV1().Nodes().Get(p.Spec.NodeName, metav1.GetOptions{})
+		if err != nil {
+			log.Warn().Err(err).Msg("Failed to get node for member")
+			return maskAny(err)
+		}
+		if node.Spec.Unschedulable {
+			agentDataWillBeGone = true
+		}
+	}
+
+	// Check PVC
+	pvcs := r.context.GetKubeCli().CoreV1().PersistentVolumeClaims(apiObject.GetNamespace())
+	pvc, err := pvcs.Get(memberStatus.PersistentVolumeClaimName, metav1.GetOptions{})
+	if err != nil {
+		log.Warn().Err(err).Msg("Failed to get PVC for member")
+		return maskAny(err)
+	}
+	if k8sutil.IsPersistentVolumeClaimMarkedForDeletion(pvc) {
+		agentDataWillBeGone = true
+	}
+
+	// Is this a simple pod restart?
+	if !agentDataWillBeGone {
+		log.Debug().Msg("Pod is just being restarted, safe to terminate agency pod")
+		return nil
+	}
+
+	// Inspect agency state
+	log.Debug().Msg("Agent data will be gone, so we will check agency serving status first")
+	ctx = agency.WithAllowNoLeader(ctx)                     // The ID we're checking may be the leader, so ignore situations where all other agents are followers
+	ctx, cancel := context.WithTimeout(ctx, time.Second*15) // Force a quick check
+	defer cancel()
+	agencyConns, err := r.context.GetAgencyClients(ctx, func(id string) bool { return id != memberStatus.ID })
+	if err != nil {
+		log.Debug().Err(err).Msg("Failed to create member client")
+		return maskAny(err)
+	}
+	if len(agencyConns) == 0 {
+		log.Debug().Err(err).Msg("No more remaining agents, we cannot delete this one")
+		return maskAny(fmt.Errorf("No more remaining agents"))
+	}
+	if err := agency.AreAgentsHealthy(ctx, agencyConns); err != nil {
+		log.Debug().Err(err).Msg("Remaining agents are not healthy")
+		return maskAny(err)
+	}
+
+	return nil
+}
+
+// prepareDBServerPodTermination checks if the given dbserver pod is allowed to terminate
+// and if so, prepares it for termination.
+// It returns nil if the pod is allowed to terminate, an error otherwise.
+func (r *Resources) prepareDBServerPodTermination(ctx context.Context, log zerolog.Logger, p *v1.Pod, memberStatus api.MemberStatus, updateMember func(api.MemberStatus) error) error {
+	// Inspect member phase
+	if memberStatus.Phase.IsFailed() {
+		log.Debug().Msg("Pod is already failed, safe to remove dbserver pod")
+		return nil
+	}
+	// Inspect deployment deletion state
+	apiObject := r.context.GetAPIObject()
+	if apiObject.GetDeletionTimestamp() != nil {
+		log.Debug().Msg("Entire deployment is being deleted, safe to remove dbserver pod")
+		return nil
+	}
+
+	// Check node the pod is scheduled on
+	dbserverDataWillBeGone := false
+	if p.Spec.NodeName != "" {
+		node, err := r.context.GetKubeCli().CoreV1().Nodes().Get(p.Spec.NodeName, metav1.GetOptions{})
+		if err != nil {
+			log.Warn().Err(err).Msg("Failed to get node for member")
+			return maskAny(err)
+		}
+		if node.Spec.Unschedulable {
+			dbserverDataWillBeGone = true
+		}
+	}
+
+	// Check PVC
+	pvcs := r.context.GetKubeCli().CoreV1().PersistentVolumeClaims(apiObject.GetNamespace())
+	pvc, err := pvcs.Get(memberStatus.PersistentVolumeClaimName, metav1.GetOptions{})
+	if err != nil {
+		log.Warn().Err(err).Msg("Failed to get PVC for member")
+		return maskAny(err)
+	}
+	if k8sutil.IsPersistentVolumeClaimMarkedForDeletion(pvc) {
+		dbserverDataWillBeGone = true
+	}
+
+	// Is this a simple pod restart?
+	if !dbserverDataWillBeGone {
+		log.Debug().Msg("Pod is just being restarted, safe to remove dbserver pod")
+		return nil
+	}
+
+	// Inspect cleaned out state
+	log.Debug().Msg("DBServer data is being deleted, so we will cleanout the dbserver first")
+	c, err := r.context.GetDatabaseClient(ctx)
+	if err != nil {
+		log.Debug().Err(err).Msg("Failed to create member client")
+		return maskAny(err)
+	}
+	cluster, err := c.Cluster(ctx)
+	if err != nil {
+		log.Debug().Err(err).Msg("Failed to access cluster")
+		return maskAny(err)
+	}
+	cleanedOut, err := cluster.IsCleanedOut(ctx, memberStatus.ID)
+	if err != nil {
+		return maskAny(err)
+	}
+	if cleanedOut {
+		// Cleanout completed
+		if memberStatus.Conditions.Update(api.ConditionTypeCleanedOut, true, "CleanedOut", "") {
+			if err := updateMember(memberStatus); err != nil {
+				return maskAny(err)
+			}
+		}
+		// Trigger PVC removal
+		if err := pvcs.Delete(memberStatus.PersistentVolumeClaimName, &metav1.DeleteOptions{}); err != nil {
+			log.Warn().Err(err).Msg("Failed to delete PVC for member")
+			return maskAny(err)
+		}
+
+		log.Debug().Msg("Server is cleaned out. Save to remove drain dbserver finalizer")
+		return nil
+	}
+	// Not cleaned out yet, check member status
+	if memberStatus.Conditions.IsTrue(api.ConditionTypeTerminated) {
+		log.Warn().Msg("Member is already terminated before it could be cleaned out. Not good, but removing dbserver pod because we cannot do anything further")
+		return nil
+	}
+	// Ensure the cleanout is triggered
+	log.Debug().Msg("Server is not yet clean out. Triggering a clean out now")
+	if err := cluster.CleanOutServer(ctx, memberStatus.ID); err != nil {
+		log.Debug().Err(err).Msg("Failed to clean out server")
+		return maskAny(err)
+	}
+	return maskAny(fmt.Errorf("Server is not yet cleaned out"))
+}

pkg/deployment/resources/pvc_finalizers.go

@@ -36,14 +36,14 @@ import (
 )
 
 // runPVCFinalizers goes through the list of PVC finalizers to see if they can be removed.
-func (r *Resources) runPVCFinalizers(ctx context.Context, p *v1.PersistentVolumeClaim, group api.ServerGroup, memberStatus api.MemberStatus) error {
+func (r *Resources) runPVCFinalizers(ctx context.Context, p *v1.PersistentVolumeClaim, group api.ServerGroup, memberStatus api.MemberStatus, updateMember func(api.MemberStatus) error) error {
 	log := r.log.With().Str("pvc-name", p.GetName()).Logger()
 	var removalList []string
 	for _, f := range p.ObjectMeta.GetFinalizers() {
 		switch f {
 		case constants.FinalizerPVCMemberExists:
 			log.Debug().Msg("Inspecting member exists finalizer")
-			if err := r.inspectFinalizerPVCMemberExists(ctx, log, p, group, memberStatus); err == nil {
+			if err := r.inspectFinalizerPVCMemberExists(ctx, log, p, group, memberStatus, updateMember); err == nil {
 				removalList = append(removalList, f)
 			} else {
 				log.Debug().Err(err).Str("finalizer", f).Msg("Cannot remove PVC finalizer yet")
@@ -66,7 +66,7 @@ func (r *Resources) runPVCFinalizers(ctx context.Context, p *v1.PersistentVolume
 
 // inspectFinalizerPVCMemberExists checks the finalizer condition for member-exists.
 // It returns nil if the finalizer can be removed.
-func (r *Resources) inspectFinalizerPVCMemberExists(ctx context.Context, log zerolog.Logger, p *v1.PersistentVolumeClaim, group api.ServerGroup, memberStatus api.MemberStatus) error {
+func (r *Resources) inspectFinalizerPVCMemberExists(ctx context.Context, log zerolog.Logger, p *v1.PersistentVolumeClaim, group api.ServerGroup, memberStatus api.MemberStatus, updateMember func(api.MemberStatus) error) error {
 	// Inspect member phase
 	if memberStatus.Phase.IsFailed() {
 		log.Debug().Msg("Member is already failed, safe to remove member-exists finalizer")
@@ -93,10 +93,22 @@ func (r *Resources) inspectFinalizerPVCMemberExists(ctx context.Context, log zer
 		}
 	}
 
-	// Member still exists, let's trigger a delete of it
+	// Member still exists, let's trigger a delete of it, if we're allowed to do so
 	if memberStatus.PodName != "" {
-		log.Info().Msg("Removing Pod of member, because PVC is being removed")
 		pods := r.context.GetKubeCli().CoreV1().Pods(apiObject.GetNamespace())
+		log.Info().Msg("Checking in Pod of member can be removed, because PVC is being removed")
+		if pod, err := pods.Get(memberStatus.PodName, metav1.GetOptions{}); err != nil && !k8sutil.IsNotFound(err) {
+			log.Debug().Err(err).Msg("Failed to get pod for PVC")
+			return maskAny(err)
+		} else if err == nil {
+			// We've got the pod, check & prepare its termination
+			if err := r.preparePodTermination(ctx, log, pod, group, memberStatus, updateMember); err != nil {
+				log.Debug().Err(err).Msg("Not allowed to remove pod yet")
+				return maskAny(err)
+			}
+		}
+
+		log.Info().Msg("Removing Pod of member, because PVC is being removed")
 		if err := pods.Delete(memberStatus.PodName, &metav1.DeleteOptions{}); err != nil && !k8sutil.IsNotFound(err) {
 			log.Debug().Err(err).Msg("Failed to delete pod")
 			return maskAny(err)

pkg/deployment/resources/pvc_inspector.go

@@ -25,6 +25,7 @@ package resources
 import (
 	"context"
 
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
 	"github.com/arangodb/kube-arangodb/pkg/metrics"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 )
@@ -68,13 +69,23 @@ func (r *Resources) InspectPVCs(ctx context.Context) error {
 			continue
 		}
 
+		updateMemberStatusNeeded := false
 		if k8sutil.IsPersistentVolumeClaimMarkedForDeletion(&p) {
 			// Process finalizers
-			if err := r.runPVCFinalizers(ctx, &p, group, memberStatus); err != nil {
+			if err := r.runPVCFinalizers(ctx, &p, group, memberStatus, func(m api.MemberStatus) error {
+				updateMemberStatusNeeded = true
+				memberStatus = m
+				return nil
+			}); err != nil {
 				// Only log here, since we'll be called to try again.
 				log.Warn().Err(err).Msg("Failed to run PVC finalizers")
 			}
 		}
+		if updateMemberStatusNeeded {
+			if err := status.Members.Update(memberStatus, group); err != nil {
+				return maskAny(err)
+			}
+		}
 	}
 
 	return nil

tests/resilience_test.go

@@ -36,6 +36,7 @@ import (
 	driver "github.com/arangodb/go-driver"
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
 	"github.com/arangodb/kube-arangodb/pkg/client"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 	"github.com/arangodb/kube-arangodb/pkg/util/retry"
 )
 
@@ -191,14 +192,14 @@ func testResiliencePVC(testGroup api.ServerGroup, t *testing.T) {
 			if err := kubecli.CoreV1().PersistentVolumeClaims(ns).Delete(m.PersistentVolumeClaimName, &metav1.DeleteOptions{}); err != nil {
 				t.Fatalf("Failed to delete pvc %s: %v", m.PersistentVolumeClaimName, err)
 			}
-			// Now delete the pod as well, otherwise the PVC will only have a deletion timestamp but its finalizers will stay on.
-			if err := kubecli.CoreV1().Pods(ns).Delete(m.PodName, &metav1.DeleteOptions{}); err != nil {
-				t.Fatalf("Failed to delete pod %s: %v", m.PodName, err)
-			}
 			// Wait for pvc to return with different UID
 			op := func() error {
 				pvc, err := kubecli.CoreV1().PersistentVolumeClaims(ns).Get(m.PersistentVolumeClaimName, metav1.GetOptions{})
 				if err != nil {
+					if k8sutil.IsNotFound(err) && group == api.ServerGroupDBServers {
+						// DBServer member is completely replaced when cleaned out, so the PVC will have a different name also
+						return nil
+					}
 					return maskAny(err)
 				}
 				if pvc.GetUID() == originalPVC.GetUID() {
@@ -298,9 +299,9 @@ func TestResiliencePVDBServer(t *testing.T) {
 				t.Fatalf("Failed to delete pvc %s: %v", m.PersistentVolumeClaimName, err)
 			}
 			// Delete Pod
-			if err := kubecli.CoreV1().Pods(ns).Delete(m.PodName, &metav1.DeleteOptions{}); err != nil {
+			/*if err := kubecli.CoreV1().Pods(ns).Delete(m.PodName, &metav1.DeleteOptions{}); err != nil {
 				t.Fatalf("Failed to delete pod %s: %v", m.PodName, err)
-			}
+			}*/
 			// Wait for cluster to be healthy again with the same number of
 			// dbservers, but the current dbserver being replaced.
 			expectedDBServerCount := apiObject.Spec.DBServers.GetCount()