Repository: FoundationDB/fdb-kubernetes-operator
Pull request #1895 (2023-11-27 08:23:18)

Update the way the operator deployment is created and verified
# Description

We have seen cases where it took longer than the expected 60 seconds for the Pods to get scheduled and into a running state, so this PR increases the threshold to 120 seconds and the overall wait time. I removed some duplication code in `RecreateOperatorPods`. 

## Type of change

*Please select one of the options below.*

- Other

## Discussion

-

## Testing

CI will run tests.

## Documentation

-

## Follow-up

-


foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: d21946669b10d5c7bbbdb8c343a06374c6f6c05e
* Duration 1:46:55
* Result: :white_check_mark: **SUCCEEDED**
* Error: `N/A`
* [Build Log](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/7ac30b5b-e3fb-4859-a2a1-940e4c1b1267.gz) terminal output (available for 30 days)
* [Build Workspace](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/7ac30b5b-e3fb-4859-a2a1-940e4c1b1267/workspace.zip) zip file of the working directory (available for 30 days)


Patch:
diff --git a/api/v1beta2/foundationdbcluster_types.go b/api/v1beta2/foundationdbcluster_types.go
index b7dcd533..01619b83 100644
--- a/api/v1beta2/foundationdbcluster_types.go
+++ b/api/v1beta2/foundationdbcluster_types.go
@@ -584,7 +584,10 @@ func (processGroupStatus *ProcessGroupStatus) AllAddressesExcluded(logger logr.L
 	}
 
 	localityExclusionString := processGroupStatus.GetExclusionString()
-	if isRemaining, isPresent := remainingMap[localityExclusionString]; isPresent && !isRemaining {
+	if isRemaining, isPresent := remainingMap[localityExclusionString]; isPresent {
+		if isRemaining {
+			return false, fmt.Errorf("process has missing exclusion string in exclusion results: %s", localityExclusionString)
+		}
 		logger.V(1).Info("process group is fully excluded based on locality based exclusions", "processGroupID", processGroupStatus.ProcessGroupID, "exclusionString", localityExclusionString)
 		return true, nil
 	}
diff --git a/fdbclient/common.go b/fdbclient/common.go
index ddd1a906..32ea2c56 100644
--- a/fdbclient/common.go
+++ b/fdbclient/common.go
@@ -59,12 +59,13 @@ func parseMachineReadableStatus(logger logr.Logger, contents []byte) (*fdbv1beta
 	}
 
 	if !status.Client.DatabaseStatus.Available {
-		logger.Info("database is unavailable")
+		logger.Info("database is unavailable", "status", status)
 		return nil, fdbv1beta2.TimeoutError{Err: fmt.Errorf("database is unavailable")}
 	}
 
 	if len(status.Cluster.Messages) > 0 {
 		logger.Info("found cluster message(s) in the machine-readable status", "messages", status.Cluster.Messages)
+		logger.V(1).Info("current status with cluster messages", "status", status)
 
 		// If the status is incomplete because of a timeout, return an error. This will force a new reconciliation.
 		for _, message := range status.Cluster.Messages {
diff --git a/kubectl-fdb/cmd/analyze.go b/kubectl-fdb/cmd/analyze.go
index ff15ae35..c36b6cff 100644
--- a/kubectl-fdb/cmd/analyze.go
+++ b/kubectl-fdb/cmd/analyze.go
@@ -72,6 +72,11 @@ func newAnalyzeCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
+			shouldAnalyzeStatus, err := cmd.Flags().GetBool("analyze-status")
+			if err != nil {
+				return err
+			}
+
 			ignoreConditions, err := cmd.Flags().GetStringArray("ignore-condition")
 			if err != nil {
 				return err
@@ -90,7 +95,7 @@ func newAnalyzeCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return cmd.Help()
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
@@ -139,6 +144,10 @@ func newAnalyzeCmd(streams genericclioptions.IOStreams) *cobra.Command {
 					errs = append(errs, err)
 				}
 
+				if !shouldAnalyzeStatus {
+					continue
+				}
+
 				err = analyzeStatus(cmd, config, clientSet, kubeClient, cluster, autoFix)
 				if err != nil {
 					errs = append(errs, err)
@@ -182,6 +191,7 @@ kubectl fdb analyze --ignore-removals=false sample-cluster-1
 
 	cmd.Flags().Bool("auto-fix", false, "defines if the analyze tasks should try to fix the found issues (e.g. replace Pods).")
 	cmd.Flags().Bool("all-clusters", false, "defines all clusters in the given namespace should be analyzed.")
+	cmd.Flags().Bool("analyze-status", false, "defines if the command should also analyze the machine-readable status of the provided cluster(s).")
 	// We might want to move this into the root cmd if we need this in multiple places
 	cmd.Flags().Bool("no-color", false, "Disable color output.")
 	cmd.Flags().StringArray("ignore-condition", nil, "specify which process group conditions should be ignored and not be printed to stdout.")
diff --git a/kubectl-fdb/cmd/buggify_crash_loop.go b/kubectl-fdb/cmd/buggify_crash_loop.go
index 943cf7b4..23a351ad 100644
--- a/kubectl-fdb/cmd/buggify_crash_loop.go
+++ b/kubectl-fdb/cmd/buggify_crash_loop.go
@@ -61,7 +61,7 @@ func newBuggifyCrashLoop(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/buggify_empty_monitor_conf.go b/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
index 9ae6baf8..5ca001f1 100644
--- a/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
+++ b/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
@@ -52,7 +52,7 @@ func newBuggifyEmptyMonitorConf(streams genericclioptions.IOStreams) *cobra.Comm
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/buggify_no_schedule.go b/kubectl-fdb/cmd/buggify_no_schedule.go
index 48656521..ca178040 100644
--- a/kubectl-fdb/cmd/buggify_no_schedule.go
+++ b/kubectl-fdb/cmd/buggify_no_schedule.go
@@ -56,7 +56,7 @@ func newBuggifyNoSchedule(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/configuration.go b/kubectl-fdb/cmd/configuration.go
index e3cda9b8..7bf6cc79 100644
--- a/kubectl-fdb/cmd/configuration.go
+++ b/kubectl-fdb/cmd/configuration.go
@@ -52,7 +52,7 @@ func newConfigurationCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/cordon.go b/kubectl-fdb/cmd/cordon.go
index db063c8e..27248356 100644
--- a/kubectl-fdb/cmd/cordon.go
+++ b/kubectl-fdb/cmd/cordon.go
@@ -64,7 +64,7 @@ func newCordonCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/deprecation.go b/kubectl-fdb/cmd/deprecation.go
index fefe1920..5f45d4a3 100644
--- a/kubectl-fdb/cmd/deprecation.go
+++ b/kubectl-fdb/cmd/deprecation.go
@@ -57,7 +57,7 @@ Deprecated settings that should be replace by a newer setting (or removed) are p
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/exclusion_status.go b/kubectl-fdb/cmd/exclusion_status.go
index 54fc5813..39c60de6 100644
--- a/kubectl-fdb/cmd/exclusion_status.go
+++ b/kubectl-fdb/cmd/exclusion_status.go
@@ -41,7 +41,7 @@ func newExclusionStatusCmd(streams genericclioptions.IOStreams) *cobra.Command {
 		Use:   "exclusion-status",
 		Short: "Get the exclusion status for all excluded processes.",
 		Long:  "Get the exclusion status for all excluded processes.",
-		Args:  cobra.ExactValidArgs(1),
+		Args:  cobra.MatchAll(cobra.ExactArgs(1), cobra.OnlyValidArgs),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			ignoreFullyExcluded, err := cmd.Flags().GetBool("ignore-fully-excluded")
 			if err != nil {
@@ -63,7 +63,7 @@ func newExclusionStatusCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/exec.go b/kubectl-fdb/cmd/exec.go
index dfde0c40..cba42df5 100644
--- a/kubectl-fdb/cmd/exec.go
+++ b/kubectl-fdb/cmd/exec.go
@@ -53,7 +53,7 @@ func newExecCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/fix_coordinator_ips.go b/kubectl-fdb/cmd/fix_coordinator_ips.go
index 07952c31..1991ceac 100644
--- a/kubectl-fdb/cmd/fix_coordinator_ips.go
+++ b/kubectl-fdb/cmd/fix_coordinator_ips.go
@@ -59,7 +59,7 @@ func newFixCoordinatorIPsCmd(streams genericclioptions.IOStreams) *cobra.Command
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/k8s_client.go b/kubectl-fdb/cmd/k8s_client.go
index a4ba9716..e2898c55 100644
--- a/kubectl-fdb/cmd/k8s_client.go
+++ b/kubectl-fdb/cmd/k8s_client.go
@@ -22,10 +22,13 @@ package cmd
 
 import (
 	"bytes"
-	ctx "context"
+	"context"
 	"fmt"
+	"github.com/go-logr/logr"
 	"io"
+	"k8s.io/klog/v2"
 	"math/rand"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"strings"
 
 	fdbv1beta1 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta1"
@@ -45,7 +48,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func getKubeClient(o *fdbBOptions) (client.Client, error) {
+func getKubeClient(ctx context.Context, o *fdbBOptions) (client.Client, error) {
 	config, err := o.configFlags.ToRESTConfig()
 	if err != nil {
 		return nil, err
@@ -56,7 +59,46 @@ func getKubeClient(o *fdbBOptions) (client.Client, error) {
 	utilruntime.Must(fdbv1beta1.AddToScheme(scheme))
 	utilruntime.Must(fdbv1beta2.AddToScheme(scheme))
 
-	return client.New(config, client.Options{Scheme: scheme})
+	// Don't printout any log messages from client-go.
+	klog.SetLogger(logr.Discard())
+
+	internalClient, err := client.NewWithWatch(config, client.Options{
+		Scheme: scheme,
+		Opts: client.WarningHandlerOptions{
+			SuppressWarnings: true,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	namespace, err := getNamespace(*o.configFlags.Namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	cacheBuilder := cache.MultiNamespacedCacheBuilder([]string{namespace})
+	internalCache, err := cacheBuilder(config, cache.Options{
+		Scheme: scheme,
+		Mapper: internalClient.RESTMapper(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// Make sure the internal cache is started.
+	go func() {
+		_ = internalCache.Start(ctx)
+	}()
+
+	internalCache.WaitForCacheSync(ctx)
+
+	return client.NewDelegatingClient(client.NewDelegatingClientInput{
+		CacheReader:       internalCache,
+		Client:            internalClient,
+		UncachedObjects:   nil,
+		CacheUnstructured: false,
+	})
 }
 
 func getNamespace(namespace string) (string, error) {
@@ -69,9 +111,9 @@ func getNamespace(namespace string) (string, error) {
 		return "", err
 	}
 
-	if context, ok := clientCfg.Contexts[clientCfg.CurrentContext]; ok {
-		if context.Namespace != "" {
-			return context.Namespace, nil
+	if ctx, ok := clientCfg.Contexts[clientCfg.CurrentContext]; ok {
+		if ctx.Namespace != "" {
+			return ctx.Namespace, nil
 		}
 	}
 
@@ -80,13 +122,13 @@ func getNamespace(namespace string) (string, error) {
 
 func getOperator(kubeClient client.Client, operatorName string, namespace string) (*appsv1.Deployment, error) {
 	operator := &appsv1.Deployment{}
-	err := kubeClient.Get(ctx.Background(), client.ObjectKey{Namespace: namespace, Name: operatorName}, operator)
+	err := kubeClient.Get(context.Background(), client.ObjectKey{Namespace: namespace, Name: operatorName}, operator)
 	return operator, err
 }
 
 func loadCluster(kubeClient client.Client, namespace string, clusterName string) (*fdbv1beta2.FoundationDBCluster, error) {
 	cluster := &fdbv1beta2.FoundationDBCluster{}
-	err := kubeClient.Get(ctx.Background(), types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster)
+	err := kubeClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster)
 	if err != nil {
 		return nil, err
 	}
@@ -94,12 +136,13 @@ func loadCluster(kubeClient client.Client, namespace string, clusterName string)
 	if err != nil {
 		return nil, err
 	}
+
 	return cluster, err
 }
 
 func getNodes(kubeClient client.Client, nodeSelector map[string]string) ([]string, error) {
 	var nodesList corev1.NodeList
-	err := kubeClient.List(ctx.Background(), &nodesList, client.MatchingLabels(nodeSelector))
+	err := kubeClient.List(context.Background(), &nodesList, client.MatchingLabels(nodeSelector))
 	if err != nil {
 		return []string{}, err
 	}
@@ -116,7 +159,7 @@ func getNodes(kubeClient client.Client, nodeSelector map[string]string) ([]strin
 func getPodsForCluster(kubeClient client.Client, cluster *fdbv1beta2.FoundationDBCluster) (*corev1.PodList, error) {
 	var podList corev1.PodList
 	err := kubeClient.List(
-		ctx.Background(),
+		context.Background(),
 		&podList,
 		client.MatchingLabels(cluster.GetMatchLabels()),
 		client.InNamespace(cluster.GetNamespace()))
diff --git a/kubectl-fdb/cmd/remove_process_group.go b/kubectl-fdb/cmd/remove_process_group.go
index fcade941..2e7919cc 100644
--- a/kubectl-fdb/cmd/remove_process_group.go
+++ b/kubectl-fdb/cmd/remove_process_group.go
@@ -60,7 +60,7 @@ func newRemoveProcessGroupCmd(streams genericclioptions.IOStreams) *cobra.Comman
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/restart.go b/kubectl-fdb/cmd/restart.go
index 82e0fa8a..457fd50c 100644
--- a/kubectl-fdb/cmd/restart.go
+++ b/kubectl-fdb/cmd/restart.go
@@ -74,7 +74,7 @@ func newRestartCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/version.go b/kubectl-fdb/cmd/version.go
index 2f83c916..fd6df61f 100644
--- a/kubectl-fdb/cmd/version.go
+++ b/kubectl-fdb/cmd/version.go
@@ -55,7 +55,7 @@ func newVersionCmd(streams genericclioptions.IOStreams) *cobra.Command {
 			}
 
 			if !clientOnly {
-				kubeClient, err := getKubeClient(o)
+				kubeClient, err := getKubeClient(cmd.Context(), o)
 				if err != nil {
 					return err
 				}


Test patch:
diff --git a/api/v1beta2/foundationdbcluster_types_test.go b/api/v1beta2/foundationdbcluster_types_test.go
index aa0cac2f..9f3820ba 100644
--- a/api/v1beta2/foundationdbcluster_types_test.go
+++ b/api/v1beta2/foundationdbcluster_types_test.go
@@ -5552,6 +5552,26 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			false,
 			fmt.Errorf("process has missing address in exclusion results: 192.168.0.2"),
 		),
+		Entry("process group is excluded using locality and it is present in the remaining map as excluded",
+			&ProcessGroupStatus{
+				ProcessGroupID: "storage-1",
+			},
+			map[string]bool{
+				fmt.Sprintf("%s:storage-1", FDBLocalityExclusionPrefix): false,
+			},
+			true,
+			nil,
+		),
+		Entry("process group is excluded using locality and it is present in the remaining map as not excluded",
+			&ProcessGroupStatus{
+				ProcessGroupID: "storage-1",
+			},
+			map[string]bool{
+				fmt.Sprintf("%s:storage-1", FDBLocalityExclusionPrefix): true,
+			},
+			false,
+			fmt.Errorf("process has missing exclusion string in exclusion results: %s", fmt.Sprintf("%s:storage-1", FDBLocalityExclusionPrefix)),
+		),
 	)
 
 	DescribeTable("when getting the removal mode", func(cluster *FoundationDBCluster, expected PodUpdateMode) {
diff --git a/e2e/fixtures/fdb_operator_client.go b/e2e/fixtures/fdb_operator_client.go
index 0d5512ea..5783c6da 100644
--- a/e2e/fixtures/fdb_operator_client.go
+++ b/e2e/fixtures/fdb_operator_client.go
@@ -40,8 +40,6 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer/yaml"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
 	yamlutil "k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -507,80 +505,38 @@ func (factory *Factory) WaitUntilOperatorPodsRunning(namespace string) {
 	).NotTo(gomega.HaveOccurred())
 
 	expectedReplicas := int(pointer.Int32Deref(deployment.Spec.Replicas, 1))
-
-	gomega.Eventually(func() (done bool) {
+	gomega.Eventually(func(g gomega.Gomega) int {
 		pods := factory.GetOperatorPods(namespace)
-		if len(pods.Items) != expectedReplicas {
-			return false
-		}
-
-		allRunning := true
+		var runningReplicas int
 		for _, pod := range pods.Items {
-			if pod.Status.Phase == corev1.PodRunning {
+			if pod.Status.Phase == corev1.PodRunning && pod.DeletionTimestamp.IsZero() {
+				runningReplicas++
 				continue
 			}
 
-			allRunning = false
-			// If the Pod is not running after 60 seconds we delete it and let the Deployment controller create a new Pod.
-			if time.Since(pod.CreationTimestamp.Time).Seconds() > 60.0 {
-				log.Println("operator Pod", pod.Name, "not running after 60 seconds, going to delete this Pod, status:", pod.Status)
+			// If the Pod is not running after 120 seconds we delete it and let the Deployment controller create a new Pod.
+			if time.Since(pod.CreationTimestamp.Time).Seconds() > 120.0 {
+				log.Println("operator Pod", pod.Name, "not running after 120 seconds, going to delete this Pod, status:", pod.Status)
 				err := factory.GetControllerRuntimeClient().Delete(ctx.TODO(), &pod)
 				if k8serrors.IsNotFound(err) {
 					continue
 				}
 
-				gomega.Expect(err).NotTo(gomega.HaveOccurred())
+				g.Expect(err).NotTo(gomega.HaveOccurred())
 			}
 		}
 
-		return allRunning
-	}).WithTimeout(5 * time.Minute).WithPolling(2 * time.Second).Should(gomega.BeTrue())
+		return runningReplicas
+	}).WithTimeout(10 * time.Minute).WithPolling(2 * time.Second).Should(gomega.BeNumerically(">=", expectedReplicas))
 }
 
 // RecreateOperatorPods will recreate all operator Pods in the specified namespace and wait until the new Pods are
 // up and running.
 func (factory *Factory) RecreateOperatorPods(namespace string) {
-	initialList := factory.GetOperatorPods(namespace)
-
-	oldOperatorIds := map[types.UID]fdbv1beta2.None{}
-	for _, pod := range initialList.Items {
-		oldOperatorIds[pod.UID] = fdbv1beta2.None{}
-	}
-
 	gomega.Expect(
 		factory.GetControllerRuntimeClient().
 			DeleteAllOf(ctx.TODO(), &corev1.Pod{}, client.InNamespace(namespace), client.MatchingLabels(map[string]string{"app": operatorDeploymentName})),
 	).NotTo(gomega.HaveOccurred())
 
-	deployment := &appsv1.Deployment{}
-	gomega.Expect(
-		factory.GetControllerRuntimeClient().
-			Get(ctx.TODO(), client.ObjectKey{Name: operatorDeploymentName, Namespace: namespace}, deployment),
-	).NotTo(gomega.HaveOccurred())
-
-	expectedReplicas := int(pointer.Int32Deref(deployment.Spec.Replicas, 1))
-
-	gomega.Expect(wait.PollImmediate(1*time.Second, 10*time.Minute, func() (done bool, err error) {
-		podList := factory.GetOperatorPods(namespace)
-
-		if len(podList.Items) == 0 {
-			return false, nil
-		}
-
-		// Wait until the Pods are recreated
-		var runningPods int
-		for _, pod := range podList.Items {
-			if _, ok := oldOperatorIds[pod.UID]; ok {
-				continue
-			}
-
-			if pod.Status.Phase != corev1.PodRunning {
-				continue
-			}
-
-			runningPods++
-		}
-
-		return runningPods == expectedReplicas, nil
-	})).NotTo(gomega.HaveOccurred())
+	factory.WaitUntilOperatorPodsRunning(namespace)
 }