Repository: m3db/m3db-operator
Pull request #333 (2022-08-20 22:19:40)

Fix flaky tests
This fixes numerous flaky tests. Most of the flakes were caused because
the Indexer is updated async.

Thanks for contributing to the M3DB Operator! We'd love to accept your contribution, but we require that most issues
outside of trivial changes such as typo corrections have an issue associated with the pull request. So please [open an
issue](https://github.com/m3db/m3db-operator/issues/new) first!



Patch:
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index e519144..c46dccc 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -911,6 +911,12 @@ func (c *M3DBController) updateStatefulSetPods(
 			if err := c.kubeClient.CoreV1().
 				Pods(pod.Namespace).
 				Delete(ctx, pod.Name, metav1.DeleteOptions{}); err != nil {
+				if kerrors.IsNotFound(err) {
+					// This pod was already deleted in a previous iteration of the reconcile loop. This can happen
+					// since the index of the podLister is updated async.
+					logger.Info("pod already deleted", zap.String("pod", pod.Name))
+					continue
+				}
 				return false, err
 			}
 			names = append(names, pod.Name)
diff --git a/pkg/controller/update_cluster.go b/pkg/controller/update_cluster.go
index 5192aee..f80b3dd 100644
--- a/pkg/controller/update_cluster.go
+++ b/pkg/controller/update_cluster.go
@@ -55,10 +55,10 @@ import (
 )
 
 var (
-	errEmptyPodList         = errors.New("cannot find removal candidate in empty list")
-	errNoPodsInPlacement    = errors.New("no pods were found in the placement")
-	errNilNamespaceRegistry = errors.New("nil registry for namespaces")
-	errPodNotInPlacement    = errors.New("instance not found in placement")
+	errEmptyPodList              = errors.New("cannot find removal candidate in empty list")
+	errNilNamespaceRegistry      = errors.New("nil registry for namespaces")
+	errPodNotInPlacement         = errors.New("instance not found in placement")
+	errPodNotAbsentFromPlacement = errors.New("could not find pod absent from placement")
 )
 
 // reconcileNamespaces will delete any namespaces currently in the cluster that
@@ -418,7 +418,8 @@ func (c *M3DBController) addPodsToPlacement(
 func (c *M3DBController) checkPodsForReplacement(
 	cluster *myspec.M3DBCluster,
 	pods []*corev1.Pod,
-	pl placement.Placement) (string, *corev1.Pod, error) {
+	pl placement.Placement,
+) (string, *corev1.Pod, error) {
 	insts := pl.Instances()
 	sort.Sort(placement.ByIDAscending(insts))
 
@@ -515,6 +516,8 @@ func (c *M3DBController) expandPlacementForSet(
 	if err != nil {
 		return err
 	}
+	// give a deterministic order to the pods for easy testing with gomock comparisons.
+	sort.Sort(podsByName(pods))
 
 	podsToAdd := make([]*corev1.Pod, 0, len(pods))
 	for _, pod := range pods {
@@ -532,12 +535,26 @@ func (c *M3DBController) expandPlacementForSet(
 		}
 	}
 	if len(podsToAdd) == 0 {
-		return errors.New("could not find pod absent from placement")
+		return errPodNotAbsentFromPlacement
 	}
 
 	return c.addPodsToPlacement(ctx, cluster, podsToAdd)
 }
 
+type podsByName []*corev1.Pod
+
+func (p podsByName) Len() int {
+	return len(p)
+}
+
+func (p podsByName) Less(i, j int) bool {
+	return p[i].Name < p[j].Name
+}
+
+func (p podsByName) Swap(i, j int) {
+	p[i], p[j] = p[j], p[i]
+}
+
 // shrinkPlacementForSet takes a StatefulSet that needs to be shrunk and removes any pods
 // that are above desired instance count in the StatefulSet from the active placement, enabling
 // the StatefulSet size to be decreased once the remove completes.
diff --git a/pkg/k8sops/annotations/annotations.go b/pkg/k8sops/annotations/annotations.go
index 6544064..00b2c27 100644
--- a/pkg/k8sops/annotations/annotations.go
+++ b/pkg/k8sops/annotations/annotations.go
@@ -76,3 +76,12 @@ func PodAnnotations(cluster *myspec.M3DBCluster) map[string]string {
 
 	return base
 }
+
+// Copy the provided annotations to a new map.
+func Copy(annotations map[string]string) map[string]string {
+	m := make(map[string]string, len(annotations))
+	for k, v := range annotations {
+		m[k] = v
+	}
+	return m
+}
diff --git a/pkg/k8sops/m3db/statefulset.go b/pkg/k8sops/m3db/statefulset.go
index c8bc2fc..a9e8513 100644
--- a/pkg/k8sops/m3db/statefulset.go
+++ b/pkg/k8sops/m3db/statefulset.go
@@ -178,7 +178,7 @@ func NewBaseStatefulSet(ssName, isolationGroup string, cluster *myspec.M3DBClust
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        ssName,
 			Labels:      objLabels,
-			Annotations: objAnnotations,
+			Annotations: annotations.Copy(objAnnotations),
 		},
 		Spec: stsSpec,
 	}


Test patch:
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index c7008a3..9e1207e 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -24,7 +24,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"math"
 	"sort"
 	"strconv"
 	"sync"
@@ -47,6 +46,7 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -172,6 +172,7 @@ type waitForStatefulSetsResult struct {
 	updatedStatefulSets      []string
 	failedUpdateStatefulSets []string
 	podUpdateGroups          [][]string
+	done                     bool
 }
 
 func waitForStatefulSets(
@@ -180,12 +181,14 @@ func waitForStatefulSets(
 	cluster *myspec.M3DBCluster,
 	verb string,
 	opts waitForStatefulSetsOptions,
-) (waitForStatefulSetsResult, bool) {
+	until func(waitForStatefulSetsResult) bool,
+) waitForStatefulSetsResult {
 	var (
 		mu          sync.Mutex
 		updates     int
 		actualSets  = make(map[string]bool)
 		updatedPods []string
+		deletedPods = make(map[string]bool)
 		ctx         = context.Background()
 	)
 
@@ -235,86 +238,75 @@ func waitForStatefulSets(
 		"delete", "pods", func(action ktesting.Action) (bool, runtime.Object, error) {
 			podName := action.(kubetesting.DeleteActionImpl).GetName()
 			mu.Lock()
-			updatedPods = append(updatedPods, podName)
+			// delete can be called multiple times for the same pod since the podLister is async updated.
+			if !deletedPods[podName] {
+				deletedPods[podName] = true
+				updatedPods = append(updatedPods, podName)
+			}
 			mu.Unlock()
 
 			return false, nil, nil
 		})
 
-	// Iterate through the expected stateful sets twice (or at least 5 times) to make sure
-	// we see all stateful sets that we expect and also be able to catch any extra stateful
-	// sets that we don't.
 	var (
-		iters           = math.Max(float64(2*len(opts.expectedStatefulSets)), 10)
-		finalErr        error
 		podUpdateGroups [][]string
+		waitResult      waitForStatefulSetsResult
 	)
-	for i := 0; i < int(iters); i++ {
+	start := time.Now()
+	done := false
+	// Contiously update the cluster until the stop condition has been satisfied or a timeout of 5s.
+	for !done && time.Since(start) < time.Second*5 {
 		err := controller.handleClusterUpdate(ctx, cluster)
 		if opts.expectError != "" {
-			finalErr = err
+			if err != nil && err.Error() == opts.expectError {
+				return waitForStatefulSetsResult{}
+			}
 		} else {
 			require.NoError(t, err)
 		}
 
 		mu.Lock()
-		var seen int
-		for _, found := range actualSets {
-			if found {
-				seen++
-			}
-		}
 		if len(updatedPods) > 0 {
 			podUpdateGroups = append(podUpdateGroups, updatedPods)
 			updatedPods = nil
 		}
-		mu.Unlock()
 
-		if seen != len(opts.expectedStatefulSets) {
-			time.Sleep(100 * time.Millisecond)
-		}
-	}
+		var (
+			updated []string
+			failed  []string
+			seen    int
+		)
+		// Start with all failed.
+		failed = append(failed, opts.expectedStatefulSets...)
+		for name, found := range actualSets {
+			if found {
+				seen++
 
-	mu.Lock()
-	defer mu.Unlock()
+				// Updated.
+				updated = append(updated, name)
 
-	var (
-		updated []string
-		failed  []string
-		seen    int
-	)
-	// Start with all failed.
-	for _, name := range opts.expectedStatefulSets {
-		failed = append(failed, name)
-	}
-	for name, found := range actualSets {
-		if found {
-			seen++
-
-			// Updated.
-			updated = append(updated, name)
-
-			// Remove from failed.
-			filterFailed := failed[:]
-			failed = failed[:0]
-			for _, elem := range filterFailed {
-				if elem != name {
-					failed = append(failed, elem)
+				// Remove from failed.
+				filterFailed := failed
+				failed = failed[:0]
+				for _, elem := range filterFailed {
+					if elem != name {
+						failed = append(failed, elem)
+					}
 				}
 			}
 		}
-	}
+		waitResult = waitForStatefulSetsResult{
+			updatedStatefulSets:      updated,
+			failedUpdateStatefulSets: failed,
+			podUpdateGroups:          podUpdateGroups,
+			done:                     seen == len(opts.expectedStatefulSets),
+		}
+		mu.Unlock()
 
-	if opts.expectError != "" {
-		require.EqualError(t, finalErr, opts.expectError)
+		done = until(waitResult)
 	}
 
-	done := seen == len(opts.expectedStatefulSets)
-	return waitForStatefulSetsResult{
-		updatedStatefulSets:      updated,
-		failedUpdateStatefulSets: failed,
-		podUpdateGroups:          podUpdateGroups,
-	}, done
+	return waitResult
 }
 
 func TestNew(t *testing.T) {
@@ -739,11 +731,13 @@ func TestHandleUpdateClusterCreatesStatefulSets(t *testing.T) {
 			defer deps.cleanup()
 			c := deps.newController(t)
 
-			_, done := waitForStatefulSets(t, c, cluster, "create", waitForStatefulSetsOptions{
+			result := waitForStatefulSets(t, c, cluster, "create", waitForStatefulSetsOptions{
 				setReadyReplicas:     false,
 				expectedStatefulSets: test.expCreateStatefulSets,
+			}, func(result waitForStatefulSetsResult) bool {
+				return result.done
 			})
-			assert.True(t, done, "expected all sets to be created")
+			assert.True(t, result.done, "expected all sets to be created")
 		})
 	}
 }
@@ -931,17 +925,23 @@ func TestHandleUpdateClusterUpdatesStatefulSets(t *testing.T) {
 				})
 			}
 
-			result, done := waitForStatefulSets(t, c, cluster, "update", waitForStatefulSetsOptions{
+			result := waitForStatefulSets(t, c, cluster, "update", waitForStatefulSetsOptions{
 				setReadyReplicas:        true,
 				expectedStatefulSets:    test.expUpdateStatefulSets,
 				simulatePodsNotUpdated:  test.simulatePodsNotUpdated,
 				simulateStaleGeneration: test.simulateStaleGeneration,
 				expectError:             test.expError,
+			}, func(result waitForStatefulSetsResult) bool {
+				if len(test.expFailedUpdateStatefulSets) > 0 {
+					return assert.ObjectsAreEqual(test.expFailedUpdateStatefulSets,
+						result.failedUpdateStatefulSets)
+				}
+				return result.done
 			})
 			if test.expNotDone {
-				assert.False(t, done, "expected not all sets to be updated")
+				assert.False(t, result.done, "expected not all sets to be updated")
 			} else {
-				assert.True(t, done, "expected all sets to be updated")
+				assert.True(t, result.done, "expected all sets to be updated")
 			}
 			if len(test.expFailedUpdateStatefulSets) > 0 {
 				assert.Equal(t, test.expFailedUpdateStatefulSets, result.failedUpdateStatefulSets)
@@ -1075,12 +1075,14 @@ func TestHandleUpdateClusterOnDeleteStrategy(t *testing.T) {
 
 			cluster.Spec.Image = newImage
 
-			result, done := waitForStatefulSets(t, c, cluster, "update", waitForStatefulSetsOptions{
+			result := waitForStatefulSets(t, c, cluster, "update", waitForStatefulSetsOptions{
 				setReadyReplicas:     true,
 				expectedStatefulSets: test.expUpdateStatefulSets,
 				currentRevision:      currentRevision,
+			}, func(result waitForStatefulSetsResult) bool {
+				return result.done && assert.ObjectsAreEqual(test.expPodUpdateGroups, result.podUpdateGroups)
 			})
-			assert.True(t, done, "expected all sets to be updated")
+			assert.True(t, result.done, "expected all sets to be updated")
 			assert.Equal(t, test.expPodUpdateGroups, result.podUpdateGroups)
 		})
 	}
@@ -1189,7 +1191,11 @@ func TestHandleResizeClusterOnDeleteStrategy(t *testing.T) {
 				for k, v := range sts.Labels {
 					pod.Labels[k] = v
 				}
-				require.NoError(t, deps.kubeClient.Tracker().Add(pod))
+				err = deps.kubeClient.Tracker().Add(pod)
+				if kerrors.IsAlreadyExists(err) {
+					return true, sts, nil
+				}
+				require.NoError(t, err)
 				deps.idProvider.EXPECT().
 					Identity(pod, gomock.Any()).
 					AnyTimes().
@@ -1197,21 +1203,23 @@ func TestHandleResizeClusterOnDeleteStrategy(t *testing.T) {
 
 				return true, sts, nil
 			})
+	start := time.Now()
 
-	for i := 0; i < 10; i++ {
+	annotationExists := true
+	for annotationExists && time.Since(start) < time.Second*5 {
 		require.NoError(t, c.handleClusterUpdate(context.Background(), cluster))
+
+		sts, err := deps.statefulSetLister.StatefulSets("namespace").Get("cluster1-rep1")
+		require.NoError(t, err)
+
+		// Let's make sure that annotation is removed when everything is done
+		_, annotationExists = sts.Annotations[annotations.ParallelUpdateInProgress]
 	}
+	require.False(t, annotationExists)
 
 	requireStatefulSetReplicas(t, deps, "cluster1-rep0", 1)
 	requireStatefulSetReplicas(t, deps, "cluster1-rep1", 3)
 	requireStatefulSetReplicas(t, deps, "cluster1-rep2", 1)
-
-	sts, err := deps.statefulSetLister.StatefulSets("namespace").Get("cluster1-rep1")
-	require.NoError(t, err)
-
-	// Let's make sure that annotation is removed when everything is done
-	_, ok := sts.Annotations[annotations.ParallelUpdateInProgress]
-	require.False(t, ok)
 }
 
 func requireStatefulSetReplicas(
diff --git a/pkg/k8sops/podidentity/provider_test.go b/pkg/k8sops/podidentity/provider_test.go
index 6714ecd..8446973 100644
--- a/pkg/k8sops/podidentity/provider_test.go
+++ b/pkg/k8sops/podidentity/provider_test.go
@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
 	kubefake "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -211,6 +212,7 @@ func TestNodeForPod(t *testing.T) {
 	defer func() {
 		close(stopCh)
 	}()
+	cache.WaitForCacheSync(stopCh, kubeInformer.Core().V1().Nodes().Informer().HasSynced)
 
 	p := newTestProvider(t, WithNodeLister(nodeLister))