Resilience improvements

Author: ewoutp

pkg/deployment/deployment_inspector.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
+	"github.com/arangodb/kube-arangodb/pkg/util"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -87,9 +88,11 @@ func (d *Deployment) inspectDeployment(lastInterval time.Duration) time.Duration
 		}
 
 		// Inspection of generated resources needed
-		if err := d.resources.InspectPods(ctx); err != nil {
+		if x, err := d.resources.InspectPods(ctx); err != nil {
 			hasError = true
 			d.CreateEvent(k8sutil.NewErrorEvent("Pod inspection failed", err, d.apiObject))
+		} else {
+			nextInterval = util.MinDuration(nextInterval, x)
 		}
 		if err := d.resources.InspectPVCs(ctx); err != nil {
 			hasError = true

pkg/deployment/resources/pod_finalizers.go

@@ -37,8 +37,13 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 )
 
+const (
+	recheckPodFinalizerInterval = time.Second * 10
+)
+
 // runPodFinalizers goes through the list of pod finalizers to see if they can be removed.
-func (r *Resources) runPodFinalizers(ctx context.Context, p *v1.Pod, memberStatus api.MemberStatus, updateMember func(api.MemberStatus) error) error {
+// Returns: Interval_till_next_inspection, error
+func (r *Resources) runPodFinalizers(ctx context.Context, p *v1.Pod, memberStatus api.MemberStatus, updateMember func(api.MemberStatus) error) (time.Duration, error) {
 	log := r.log.With().Str("pod-name", p.GetName()).Logger()
 	var removalList []string
 	for _, f := range p.ObjectMeta.GetFinalizers() {
@@ -55,7 +60,7 @@ func (r *Resources) runPodFinalizers(ctx context.Context, p *v1.Pod, memberStatu
 			if err := r.inspectFinalizerPodDrainDBServer(ctx, log, p, memberStatus, updateMember); err == nil {
 				removalList = append(removalList, f)
 			} else {
-				log.Debug().Err(err).Str("finalizer", f).Msg("Cannot remove finalizer yet")
+				log.Debug().Err(err).Str("finalizer", f).Msg("Cannot remove Pod finalizer yet")
 			}
 		}
 	}
@@ -65,10 +70,15 @@ func (r *Resources) runPodFinalizers(ctx context.Context, p *v1.Pod, memberStatu
 		ignoreNotFound := false
 		if err := k8sutil.RemovePodFinalizers(log, kubecli, p, removalList, ignoreNotFound); err != nil {
 			log.Debug().Err(err).Msg("Failed to update pod (to remove finalizers)")
-			return maskAny(err)
+			return 0, maskAny(err)
+		} else {
+			log.Debug().Strs("finalizers", removalList).Msg("Removed finalizer(s) from Pod")
 		}
+	} else {
+		// Check again at given interval
+		return recheckPodFinalizerInterval, nil
 	}
-	return nil
+	return maxPodInspectorInterval, nil
 }
 
 // inspectFinalizerPodAgencyServing checks the finalizer condition for agency-serving.
@@ -131,14 +141,16 @@ func (r *Resources) inspectFinalizerPodAgencyServing(ctx context.Context, log ze
 		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 health")
+		log.Debug().Err(err).Msg("Remaining agents are not healthy")
 		return maskAny(err)
 	}
 
 	// Remaining agents are healthy, we can remove this one and trigger a delete of the PVC
-	if err := pvcs.Delete(memberStatus.PersistentVolumeClaimName, &metav1.DeleteOptions{}); err != nil {
+	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)
+	} else {
+		log.Debug().Str("pvc-name", memberStatus.PersistentVolumeClaimName).Msg("Removed PVC of member so agency can be completely replaced")
 	}
 
 	return nil

pkg/deployment/resources/pod_inspector.go

@@ -31,6 +31,7 @@ import (
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
 	"github.com/arangodb/kube-arangodb/pkg/metrics"
+	"github.com/arangodb/kube-arangodb/pkg/util"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 )
 
@@ -39,19 +40,22 @@ var (
 )
 
 const (
-	podScheduleTimeout = time.Minute // How long we allow the schedule to take scheduling a pod.
+	podScheduleTimeout      = time.Minute // How long we allow the schedule to take scheduling a pod.
+	maxPodInspectorInterval = time.Hour
 )
 
 // InspectPods lists all pods that belong to the given deployment and updates
 // the member status of the deployment accordingly.
-func (r *Resources) InspectPods(ctx context.Context) error {
+// Returns: Interval_till_next_inspection, error
+func (r *Resources) InspectPods(ctx context.Context) (time.Duration, error) {
 	log := r.log
 	var events []*k8sutil.Event
+	nextInterval := maxPodInspectorInterval // Large by default, will be made smaller if needed in the rest of the function
 
 	pods, err := r.context.GetOwnedPods()
 	if err != nil {
 		log.Debug().Err(err).Msg("Failed to get owned pods")
-		return maskAny(err)
+		return 0, maskAny(err)
 	}
 
 	// Update member status from all pods found
@@ -80,7 +84,7 @@ func (r *Resources) InspectPods(ctx context.Context) error {
 				ignoreNotFound := false
 				if err := k8sutil.RemovePodFinalizers(log, kubecli, &p, p.GetFinalizers(), ignoreNotFound); err != nil {
 					log.Debug().Err(err).Msg("Failed to update pod (to remove all finalizers)")
-					return maskAny(err)
+					return 0, maskAny(err)
 				}
 			}
 			continue
@@ -136,18 +140,20 @@ func (r *Resources) InspectPods(ctx context.Context) error {
 		}
 		if k8sutil.IsPodMarkedForDeletion(&p) {
 			// Process finalizers
-			if err := r.runPodFinalizers(ctx, &p, memberStatus, func(m api.MemberStatus) error {
+			if x, err := r.runPodFinalizers(ctx, &p, 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 pod finalizers")
+			} else {
+				nextInterval = util.MinDuration(nextInterval, x)
 			}
 		}
 		if updateMemberStatusNeeded {
 			if err := status.Members.Update(memberStatus, group); err != nil {
-				return maskAny(err)
+				return 0, maskAny(err)
 			}
 		}
 	}
@@ -238,14 +244,14 @@ func (r *Resources) InspectPods(ctx context.Context) error {
 
 	// Save status
 	if err := r.context.UpdateStatus(status, lastVersion); err != nil {
-		return maskAny(err)
+		return 0, maskAny(err)
 	}
 
 	// Create events
 	for _, evt := range events {
 		r.context.CreateEvent(evt)
 	}
-	return nil
+	return nextInterval, nil
 }
 
 // GetExpectedPodArguments creates command line arguments for a server in the given group with given ID.

pkg/deployment/resources/pvc_finalizers.go

@@ -46,7 +46,7 @@ func (r *Resources) runPVCFinalizers(ctx context.Context, p *v1.PersistentVolume
 			if err := r.inspectFinalizerPVCMemberExists(ctx, log, p, group, memberStatus); err == nil {
 				removalList = append(removalList, f)
 			} else {
-				log.Debug().Err(err).Str("finalizer", f).Msg("Cannot remove finalizer yet")
+				log.Debug().Err(err).Str("finalizer", f).Msg("Cannot remove PVC finalizer yet")
 			}
 		}
 	}
@@ -57,6 +57,8 @@ func (r *Resources) runPVCFinalizers(ctx context.Context, p *v1.PersistentVolume
 		if err := k8sutil.RemovePVCFinalizers(log, kubecli, p, removalList, ignoreNotFound); err != nil {
 			log.Debug().Err(err).Msg("Failed to update PVC (to remove finalizers)")
 			return maskAny(err)
+		} else {
+			log.Debug().Strs("finalizers", removalList).Msg("Removed finalizer(s) from PVC")
 		}
 	}
 	return nil

pkg/util/duration.go

@@ -0,0 +1,43 @@
+//
+// 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 util
+
+import (
+	"time"
+)
+
+// MaxDuration returns the largest of the given durations
+func MaxDuration(a, b time.Duration) time.Duration {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+// MinDuration returns the smallest of the given durations
+func MinDuration(a, b time.Duration) time.Duration {
+	if a < b {
+		return a
+	}
+	return b
+}

tests/resilience_test.go

@@ -122,16 +122,28 @@ func TestResiliencePod(t *testing.T) {
 	removeDeployment(c, depl.GetName(), ns)
 }
 
-// TestResiliencePVC
-// Tests handling of individual pod deletions
-func TestResiliencePVC(t *testing.T) {
+// TestResiliencePVCAgents
+// Tests handling of individual PVCs of agents being deleted
+func TestResiliencePVCAgents(t *testing.T) {
+	testResiliencePVC(api.ServerGroupAgents, t)
+}
+
+// TestResiliencePVCDBServers
+// Tests handling of individual PVCs of dbservers being deleted
+func TestResiliencePVCDBServers(t *testing.T) {
+	testResiliencePVC(api.ServerGroupDBServers, t)
+}
+
+// testResiliencePVC
+// Tests handling of individual PVCs of given group being deleted
+func testResiliencePVC(testGroup api.ServerGroup, t *testing.T) {
 	longOrSkip(t)
 	c := client.MustNewInCluster()
 	kubecli := mustNewKubeClient(t)
 	ns := getNamespace(t)
 
 	// Prepare deployment config
-	depl := newDeployment("test-pvc-resilience-" + uniuri.NewLen(4))
+	depl := newDeployment(fmt.Sprintf("test-pvc-resilience-%s-%s", testGroup.AsRoleAbbreviated(), uniuri.NewLen(4)))
 	depl.Spec.Mode = api.NewMode(api.DeploymentModeCluster)
 	depl.Spec.SetDefaults(depl.GetName()) // this must be last
 
@@ -166,9 +178,8 @@ func TestResiliencePVC(t *testing.T) {
 
 	// Delete one pvc after the other
 	apiObject.ForeachServerGroup(func(group api.ServerGroup, spec api.ServerGroupSpec, status *api.MemberStatusList) error {
-		if group != api.ServerGroupAgents {
-			// Coordinators have no PVC
-			// DBServers will be cleaned out and create a new member
+		if group != testGroup {
+			// We only test a specific group here
 			return nil
 		}
 		for _, m := range *status {
@@ -195,9 +206,13 @@ func TestResiliencePVC(t *testing.T) {
 				}
 				return nil
 			}
-			if err := retry.Retry(op, time.Minute); err != nil {
+			if err := retry.Retry(op, time.Minute*2); err != nil {
 				t.Fatalf("PVC did not restart: %v", err)
 			}
+			// Wait for deployment to be ready
+			if _, err = waitUntilDeployment(c, depl.GetName(), ns, deploymentIsReady()); err != nil {
+				t.Fatalf("Deployment not running in time: %v", err)
+			}
 			// Wait for cluster to be completely ready
 			if err := waitUntilClusterHealth(client, func(h driver.ClusterHealth) error {
 				return clusterHealthEqualsSpec(h, apiObject.Spec)