Merge pull request #174 from arangodb/improvement/allow-early-cluster-inspections-2-fail

Avoid useless warnings in log

Author: ewoutp

pkg/deployment/cleanup.go

@@ -35,7 +35,8 @@ func (d *Deployment) removePodFinalizers() error {
 		return maskAny(err)
 	}
 	for _, p := range pods {
-		if err := k8sutil.RemovePodFinalizers(log, kubecli, &p, p.GetFinalizers()); err != nil {
+		ignoreNotFound := true
+		if err := k8sutil.RemovePodFinalizers(log, kubecli, &p, p.GetFinalizers(), ignoreNotFound); err != nil {
 			log.Warn().Err(err).Msg("Failed to remove pod finalizers")
 		}
 	}
@@ -51,7 +52,8 @@ func (d *Deployment) removePVCFinalizers() error {
 		return maskAny(err)
 	}
 	for _, p := range pvcs {
-		if err := k8sutil.RemovePVCFinalizers(log, kubecli, &p, p.GetFinalizers()); err != nil {
+		ignoreNotFound := true
+		if err := k8sutil.RemovePVCFinalizers(log, kubecli, &p, p.GetFinalizers(), ignoreNotFound); err != nil {
 			log.Warn().Err(err).Msg("Failed to remove PVC finalizers")
 		}
 	}

pkg/deployment/cluster_scaling_integration.go

@@ -50,6 +50,10 @@ type clusterScalingIntegration struct {
 	}
 }
 
+const (
+	maxClusterBootstrapTime = time.Minute * 2 // Time we allow a cluster bootstrap to take, before we can do cluster inspections.
+)
+
 // newClusterScalingIntegration creates a new clusterScalingIntegration.
 func newClusterScalingIntegration(depl *Deployment) *clusterScalingIntegration {
 	return &clusterScalingIntegration{
@@ -67,20 +71,29 @@ func (ci *clusterScalingIntegration) SendUpdateToCluster(spec api.DeploymentSpec
 
 // listenForClusterEvents keep listening for changes entered in the UI of the cluster.
 func (ci *clusterScalingIntegration) ListenForClusterEvents(stopCh <-chan struct{}) {
+	start := time.Now()
+	goodInspections := 0
 	for {
 		delay := time.Second * 2
 
 		// Is deployment in running state
 		if ci.depl.GetPhase() == api.DeploymentPhaseRunning {
 			// Update cluster with our state
 			ctx := context.Background()
-			safeToAskCluster, err := ci.updateClusterServerCount(ctx)
+			expectSuccess := goodInspections > 0 || time.Since(start) > maxClusterBootstrapTime
+			safeToAskCluster, err := ci.updateClusterServerCount(ctx, expectSuccess)
 			if err != nil {
-				ci.log.Debug().Err(err).Msg("Cluster update failed")
+				if expectSuccess {
+					ci.log.Debug().Err(err).Msg("Cluster update failed")
+				}
 			} else if safeToAskCluster {
 				// Inspect once
-				if err := ci.inspectCluster(ctx); err != nil {
-					ci.log.Debug().Err(err).Msg("Cluster inspection failed")
+				if err := ci.inspectCluster(ctx, expectSuccess); err != nil {
+					if expectSuccess {
+						ci.log.Debug().Err(err).Msg("Cluster inspection failed")
+					}
+				} else {
+					goodInspections++
 				}
 			}
 		}
@@ -96,15 +109,17 @@ func (ci *clusterScalingIntegration) ListenForClusterEvents(stopCh <-chan struct
 }
 
 // Perform a single inspection of the cluster
-func (ci *clusterScalingIntegration) inspectCluster(ctx context.Context) error {
+func (ci *clusterScalingIntegration) inspectCluster(ctx context.Context, expectSuccess bool) error {
 	log := ci.log
 	c, err := ci.depl.clientCache.GetDatabase(ctx)
 	if err != nil {
 		return maskAny(err)
 	}
 	req, err := arangod.GetNumberOfServers(ctx, c.Connection())
 	if err != nil {
-		log.Debug().Err(err).Msg("Failed to get number of servers")
+		if expectSuccess {
+			log.Debug().Err(err).Msg("Failed to get number of servers")
+		}
 		return maskAny(err)
 	}
 	if req.Coordinators == nil && req.DBServers == nil {
@@ -150,7 +165,7 @@ func (ci *clusterScalingIntegration) inspectCluster(ctx context.Context) error {
 
 // updateClusterServerCount updates the intended number of servers of the cluster.
 // Returns true when it is safe to ask the cluster for updates.
-func (ci *clusterScalingIntegration) updateClusterServerCount(ctx context.Context) (bool, error) {
+func (ci *clusterScalingIntegration) updateClusterServerCount(ctx context.Context, expectSuccess bool) (bool, error) {
 	// Any update needed?
 	ci.pendingUpdate.mutex.Lock()
 	spec := ci.pendingUpdate.spec
@@ -168,7 +183,9 @@ func (ci *clusterScalingIntegration) updateClusterServerCount(ctx context.Contex
 	coordinatorCount := spec.Coordinators.GetCount()
 	dbserverCount := spec.DBServers.GetCount()
 	if err := arangod.SetNumberOfServers(ctx, c.Connection(), coordinatorCount, dbserverCount); err != nil {
-		log.Debug().Err(err).Msg("Failed to set number of servers")
+		if expectSuccess {
+			log.Debug().Err(err).Msg("Failed to set number of servers")
+		}
 		return false, maskAny(err)
 	}
 

pkg/deployment/deployment_finalizers.go

@@ -109,7 +109,8 @@ func removeDeploymentFinalizers(log zerolog.Logger, cli versioned.Interface, dep
 		*depl = *result
 		return nil
 	}
-	if err := k8sutil.RemoveFinalizers(log, finalizers, getFunc, updateFunc); err != nil {
+	ignoreNotFound := false
+	if err := k8sutil.RemoveFinalizers(log, finalizers, getFunc, updateFunc, ignoreNotFound); err != nil {
 		return maskAny(err)
 	}
 	return nil

pkg/deployment/resources/pod_creator.go

@@ -32,6 +32,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/arangodb/go-driver/jwt"
@@ -428,7 +429,7 @@ func (r *Resources) createPodTolerations(group api.ServerGroup, groupSpec api.Se
 }
 
 // createPodForMember creates all Pods listed in member status
-func (r *Resources) createPodForMember(spec api.DeploymentSpec, memberID string) error {
+func (r *Resources) createPodForMember(spec api.DeploymentSpec, memberID string, imageNotFoundOnce *sync.Once) error {
 	kubecli := r.context.GetKubeCli()
 	log := r.log
 	apiObject := r.context.GetAPIObject()
@@ -453,7 +454,9 @@ func (r *Resources) createPodForMember(spec api.DeploymentSpec, memberID string)
 	// Find image ID
 	imageInfo, imageFound := status.Images.GetByImage(spec.GetImage())
 	if !imageFound {
-		log.Debug().Str("image", spec.GetImage()).Msg("Image ID is not known yet for image")
+		imageNotFoundOnce.Do(func() {
+			log.Debug().Str("image", spec.GetImage()).Msg("Image ID is not known yet for image")
+		})
 		return nil
 	}
 	// Create pod
@@ -602,6 +605,7 @@ func (r *Resources) createPodForMember(spec api.DeploymentSpec, memberID string)
 func (r *Resources) EnsurePods() error {
 	iterator := r.context.GetServerGroupIterator()
 	status, _ := r.context.GetStatus()
+	imageNotFoundOnce := &sync.Once{}
 	if err := iterator.ForeachServerGroup(func(group api.ServerGroup, groupSpec api.ServerGroupSpec, status *api.MemberStatusList) error {
 		for _, m := range *status {
 			if m.Phase != api.MemberPhaseNone {
@@ -611,7 +615,7 @@ func (r *Resources) EnsurePods() error {
 				continue
 			}
 			spec := r.context.GetSpec()
-			if err := r.createPodForMember(spec, m.ID); err != nil {
+			if err := r.createPodForMember(spec, m.ID, imageNotFoundOnce); err != nil {
 				return maskAny(err)
 			}
 		}

pkg/deployment/resources/pod_finalizers.go

@@ -62,7 +62,8 @@ func (r *Resources) runPodFinalizers(ctx context.Context, p *v1.Pod, memberStatu
 	// Remove finalizers (if needed)
 	if len(removalList) > 0 {
 		kubecli := r.context.GetKubeCli()
-		if err := k8sutil.RemovePodFinalizers(log, kubecli, p, removalList); err != nil {
+		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)
 		}

pkg/deployment/resources/pod_inspector.go

@@ -78,7 +78,8 @@ func (r *Resources) InspectPods(ctx context.Context) error {
 				// Remove all finalizers, so it can be removed.
 				log.Warn().Msg("Pod belongs to this deployment, but we don't know the member. Removing all finalizers")
 				kubecli := r.context.GetKubeCli()
-				if err := k8sutil.RemovePodFinalizers(log, kubecli, &p, p.GetFinalizers()); err != nil {
+				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)
 				}

pkg/deployment/resources/pvc_finalizers.go

@@ -53,7 +53,8 @@ func (r *Resources) runPVCFinalizers(ctx context.Context, p *v1.PersistentVolume
 	// Remove finalizers (if needed)
 	if len(removalList) > 0 {
 		kubecli := r.context.GetKubeCli()
-		if err := k8sutil.RemovePVCFinalizers(log, kubecli, p, removalList); err != nil {
+		ignoreNotFound := false
+		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)
 		}

pkg/deployment/resources/pvc_inspector.go

@@ -59,7 +59,8 @@ func (r *Resources) InspectPVCs(ctx context.Context) error {
 				// Remove all finalizers, so it can be removed.
 				log.Warn().Msg("PVC belongs to this deployment, but we don't know the member. Removing all finalizers")
 				kubecli := r.context.GetKubeCli()
-				if err := k8sutil.RemovePVCFinalizers(log, kubecli, &p, p.GetFinalizers()); err != nil {
+				ignoreNotFound := false
+				if err := k8sutil.RemovePVCFinalizers(log, kubecli, &p, p.GetFinalizers(), ignoreNotFound); err != nil {
 					log.Debug().Err(err).Msg("Failed to update PVC (to remove all finalizers)")
 					return maskAny(err)
 				}

pkg/replication/finalizers.go

@@ -78,7 +78,8 @@ func (dr *DeploymentReplication) runFinalizers(ctx context.Context, p *api.Arang
 	}
 	// Remove finalizers (if needed)
 	if len(removalList) > 0 {
-		if err := removeDeploymentReplicationFinalizers(log, dr.deps.CRCli, p, removalList); err != nil {
+		ignoreNotFound := false
+		if err := removeDeploymentReplicationFinalizers(log, dr.deps.CRCli, p, removalList, ignoreNotFound); err != nil {
 			log.Debug().Err(err).Msg("Failed to update deployment replication (to remove finalizers)")
 			return maskAny(err)
 		}
@@ -165,7 +166,7 @@ func (dr *DeploymentReplication) inspectFinalizerDeplReplStopSync(ctx context.Co
 }
 
 // removeDeploymentReplicationFinalizers removes the given finalizers from the given DeploymentReplication.
-func removeDeploymentReplicationFinalizers(log zerolog.Logger, crcli versioned.Interface, p *api.ArangoDeploymentReplication, finalizers []string) error {
+func removeDeploymentReplicationFinalizers(log zerolog.Logger, crcli versioned.Interface, p *api.ArangoDeploymentReplication, finalizers []string, ignoreNotFound bool) error {
 	repls := crcli.ReplicationV1alpha().ArangoDeploymentReplications(p.GetNamespace())
 	getFunc := func() (metav1.Object, error) {
 		result, err := repls.Get(p.GetName(), metav1.GetOptions{})
@@ -183,7 +184,7 @@ func removeDeploymentReplicationFinalizers(log zerolog.Logger, crcli versioned.I
 		*p = *result
 		return nil
 	}
-	if err := k8sutil.RemoveFinalizers(log, finalizers, getFunc, updateFunc); err != nil {
+	if err := k8sutil.RemoveFinalizers(log, finalizers, getFunc, updateFunc, ignoreNotFound); err != nil {
 		return maskAny(err)
 	}
 	return nil

pkg/util/k8sutil/finalizers.go

@@ -34,7 +34,7 @@ const (
 )
 
 // RemovePodFinalizers removes the given finalizers from the given pod.
-func RemovePodFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1.Pod, finalizers []string) error {
+func RemovePodFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1.Pod, finalizers []string, ignoreNotFound bool) error {
 	pods := kubecli.CoreV1().Pods(p.GetNamespace())
 	getFunc := func() (metav1.Object, error) {
 		result, err := pods.Get(p.GetName(), metav1.GetOptions{})
@@ -52,14 +52,14 @@ func RemovePodFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1
 		*p = *result
 		return nil
 	}
-	if err := RemoveFinalizers(log, finalizers, getFunc, updateFunc); err != nil {
+	if err := RemoveFinalizers(log, finalizers, getFunc, updateFunc, ignoreNotFound); err != nil {
 		return maskAny(err)
 	}
 	return nil
 }
 
 // RemovePVCFinalizers removes the given finalizers from the given PVC.
-func RemovePVCFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1.PersistentVolumeClaim, finalizers []string) error {
+func RemovePVCFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1.PersistentVolumeClaim, finalizers []string, ignoreNotFound bool) error {
 	pvcs := kubecli.CoreV1().PersistentVolumeClaims(p.GetNamespace())
 	getFunc := func() (metav1.Object, error) {
 		result, err := pvcs.Get(p.GetName(), metav1.GetOptions{})
@@ -77,7 +77,7 @@ func RemovePVCFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1
 		*p = *result
 		return nil
 	}
-	if err := RemoveFinalizers(log, finalizers, getFunc, updateFunc); err != nil {
+	if err := RemoveFinalizers(log, finalizers, getFunc, updateFunc, ignoreNotFound); err != nil {
 		return maskAny(err)
 	}
 	return nil
@@ -87,12 +87,16 @@ func RemovePVCFinalizers(log zerolog.Logger, kubecli kubernetes.Interface, p *v1
 // The functions tries to get the object using the provided get function,
 // then remove the given finalizers and update the update using the given update function.
 // In case of an update conflict, the functions tries again.
-func RemoveFinalizers(log zerolog.Logger, finalizers []string, getFunc func() (metav1.Object, error), updateFunc func(metav1.Object) error) error {
+func RemoveFinalizers(log zerolog.Logger, finalizers []string, getFunc func() (metav1.Object, error), updateFunc func(metav1.Object) error, ignoreNotFound bool) error {
 	attempts := 0
 	for {
 		attempts++
 		obj, err := getFunc()
 		if err != nil {
+			if IsNotFound(err) && ignoreNotFound {
+				// Object no longer found and we're allowed to ignore that.
+				return nil
+			}
 			log.Warn().Err(err).Msg("Failed to get resource")
 			return maskAny(err)
 		}
@@ -125,6 +129,9 @@ func RemoveFinalizers(log zerolog.Logger, finalizers []string, getFunc func() (m
 					// Try again
 					continue
 				}
+			} else if IsNotFound(err) && ignoreNotFound {
+				// Object no longer found and we're allowed to ignore that.
+				return nil
 			} else if err != nil {
 				log.Warn().Err(err).Msg("Failed to update resource with fewer finalizers")
 				return maskAny(err)