fix(logstorage): warn when node count only exceeds replicas by 1

Author: tianfeng92

pkg/controller/logstorage/initializer/initializing_controller.go

@@ -168,25 +168,31 @@ func FillDefaults(opr *operatorv1.LogStorage) {
 	}
 }
 
-func validateLogStorage(spec *operatorv1.LogStorageSpec) error {
-	if err := validateReplicasForNodeCount(spec); err != nil {
-		return err
+func validateLogStorage(spec *operatorv1.LogStorageSpec) (error, string) {
+	err, warning := validateReplicasForNodeCount(spec)
+	if err != nil {
+		return err, ""
 	}
-	return validateComponentResources(spec)
+	if err := validateComponentResources(spec); err != nil {
+		return err, ""
+	}
+	return nil, warning
 }
 
-func validateReplicasForNodeCount(spec *operatorv1.LogStorageSpec) error {
+func validateReplicasForNodeCount(spec *operatorv1.LogStorageSpec) (error, string) {
 	if spec.Nodes == nil || spec.Indices == nil || spec.Indices.Replicas == nil {
-		return nil
+		return nil, ""
 	}
 	replicas := int(*spec.Indices.Replicas)
 	nodeCount := int(spec.Nodes.Count)
 	if replicas > 0 && nodeCount <= replicas {
-		return fmt.Errorf("LogStorage spec.indices.replicas (%d) must be less than spec.nodes.count (%d); "+
-			"replica shards cannot be allocated when there are not enough nodes. "+
-			"For a single-node Elasticsearch cluster, set spec.indices.replicas to 0", replicas, nodeCount)
+		return fmt.Errorf("LogStorage spec.indices.replicas (%d) must be less than spec.nodes.count (%d); replica shards cannot be allocated when there are not enough nodes. For a single-node Elasticsearch cluster, set spec.indices.replicas to 0", replicas, nodeCount), ""
 	}
-	return nil
+
+	if replicas > 0 && nodeCount == replicas+1 {
+		return nil, fmt.Sprintf("LogStorage spec.nodes.count (%d) is only 1 more than spec.indices.replicas (%d); this may prevent voluntary pod evictions (e.g., node repaving) due to PodDisruptionBudget constraints. If this is expected for your environment, no action is needed. Otherwise, consider setting spec.nodes.count to at least %d", nodeCount, replicas, replicas+2)
+	}
+	return nil, ""
 }
 
 func validateComponentResources(spec *operatorv1.LogStorageSpec) error {
@@ -253,13 +259,16 @@ func (r *LogStorageInitializer) Reconcile(ctx context.Context, request reconcile
 
 	// Default and validate the object.
 	FillDefaults(ls)
-	err = validateLogStorage(&ls.Spec)
+	err, warning := validateLogStorage(&ls.Spec)
 	if err != nil {
 		// Invalid - mark it as such and return.
 		r.setConditionDegraded(ctx, ls, reqLogger)
 		r.status.SetDegraded(operatorv1.ResourceValidationError, "An error occurred while validating LogStorage", err, reqLogger)
 		return reconcile.Result{}, err
 	}
+	if warning != "" {
+		reqLogger.Info(warning)
+	}
 
 	pullSecrets, err := utils.GetNetworkingPullSecrets(install, r.client)
 	if err != nil {

pkg/controller/logstorage/initializer/initializing_controller_test.go

@@ -128,6 +128,8 @@ var _ = Describe("LogStorage Initializing controller", func() {
 			mockStatus.On("ClearDegraded")
 			mockStatus.On("SetDegraded", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 			mockStatus.On("OnCRNotFound")
+			mockStatus.On("SetWarning", mock.Anything, mock.Anything)
+			mockStatus.On("ClearWarning", mock.Anything)
 		})
 
 		It("fills defaults on an empty LogStorage", func() {
@@ -204,6 +206,27 @@ var _ = Describe("LogStorage Initializing controller", func() {
 			Expect(ls.Status.State).Should(Equal(operatorv1.TigeraStatusDegraded))
 		})
 
+		It("sets a warning when node count only exceeds replicas by 1", func() {
+			var replicas int32 = 1
+			ls := &operatorv1.LogStorage{}
+			ls.Name = "tigera-secure"
+			FillDefaults(ls)
+			ls.Spec.Indices.Replicas = &replicas
+			ls.Spec.Nodes.Count = 2
+			Expect(cli.Create(ctx, ls)).ShouldNot(HaveOccurred())
+
+			r, err := NewTestInitializer(cli, scheme, mockStatus, operatorv1.ProviderNone, dns.DefaultClusterDomain)
+			Expect(err).ShouldNot(HaveOccurred())
+			_, err = r.Reconcile(ctx, reconcile.Request{})
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(mockStatus.AssertNumberOfCalls(GinkgoT(), "SetDegraded", 0)).Should(BeTrue())
+			mockStatus.AssertCalled(GinkgoT(), "SetWarning", "replicaNodeCount", mock.Anything)
+
+			ls = &operatorv1.LogStorage{}
+			Expect(cli.Get(ctx, client.ObjectKey{Name: "tigera-secure"}, ls)).ShouldNot(HaveOccurred())
+			Expect(ls.Status.State).Should(Equal(operatorv1.TigeraStatusReady))
+		})
+
 		It("handles LogStorage deletion", func() {
 			// Create a LogStorage instance.
 			ls := &operatorv1.LogStorage{}
@@ -379,7 +402,9 @@ var _ = Describe("LogStorage Initializing controller", func() {
 				Nodes:   &operatorv1.Nodes{Count: 1},
 				Indices: &operatorv1.Indices{Replicas: &replicas},
 			}
-			Expect(validateReplicasForNodeCount(spec)).NotTo(BeNil())
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).NotTo(BeNil())
+			Expect(warning).To(BeEmpty())
 		})
 
 		It("should return an error when replicas equals node count", func() {
@@ -388,40 +413,72 @@ var _ = Describe("LogStorage Initializing controller", func() {
 				Nodes:   &operatorv1.Nodes{Count: 2},
 				Indices: &operatorv1.Indices{Replicas: &replicas},
 			}
-			Expect(validateReplicasForNodeCount(spec)).NotTo(BeNil())
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).NotTo(BeNil())
+			Expect(warning).To(BeEmpty())
 		})
 
-		It("should return nil when replicas is 0 and node count is 1", func() {
-			var replicas int32 = 0
+		It("should return a warning when node count is only 1 more than replicas", func() {
+			var replicas int32 = 1
 			spec := &operatorv1.LogStorageSpec{
-				Nodes:   &operatorv1.Nodes{Count: 1},
+				Nodes:   &operatorv1.Nodes{Count: 2},
+				Indices: &operatorv1.Indices{Replicas: &replicas},
+			}
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).To(BeNil())
+			Expect(warning).To(ContainSubstring("only 1 more than"))
+		})
+
+		It("should return a warning when node count is 3 and replicas is 2", func() {
+			var replicas int32 = 2
+			spec := &operatorv1.LogStorageSpec{
+				Nodes:   &operatorv1.Nodes{Count: 3},
 				Indices: &operatorv1.Indices{Replicas: &replicas},
 			}
-			Expect(validateReplicasForNodeCount(spec)).To(BeNil())
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).To(BeNil())
+			Expect(warning).To(ContainSubstring("only 1 more than"))
 		})
 
-		It("should return nil when replicas is 1 and node count is 2", func() {
+		It("should return no error or warning when node count exceeds replicas by 2 or more", func() {
 			var replicas int32 = 1
 			spec := &operatorv1.LogStorageSpec{
-				Nodes:   &operatorv1.Nodes{Count: 2},
+				Nodes:   &operatorv1.Nodes{Count: 3},
+				Indices: &operatorv1.Indices{Replicas: &replicas},
+			}
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).To(BeNil())
+			Expect(warning).To(BeEmpty())
+		})
+
+		It("should return no error or warning when replicas is 0 and node count is 1", func() {
+			var replicas int32 = 0
+			spec := &operatorv1.LogStorageSpec{
+				Nodes:   &operatorv1.Nodes{Count: 1},
 				Indices: &operatorv1.Indices{Replicas: &replicas},
 			}
-			Expect(validateReplicasForNodeCount(spec)).To(BeNil())
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).To(BeNil())
+			Expect(warning).To(BeEmpty())
 		})
 
-		It("should return nil when indices is nil", func() {
+		It("should return no error or warning when indices is nil", func() {
 			spec := &operatorv1.LogStorageSpec{
 				Nodes: &operatorv1.Nodes{Count: 1},
 			}
-			Expect(validateReplicasForNodeCount(spec)).To(BeNil())
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).To(BeNil())
+			Expect(warning).To(BeEmpty())
 		})
 
-		It("should return nil when nodes is nil", func() {
+		It("should return no error or warning when nodes is nil", func() {
 			var replicas int32 = 1
 			spec := &operatorv1.LogStorageSpec{
 				Indices: &operatorv1.Indices{Replicas: &replicas},
 			}
-			Expect(validateReplicasForNodeCount(spec)).To(BeNil())
+			err, warning := validateReplicasForNodeCount(spec)
+			Expect(err).To(BeNil())
+			Expect(warning).To(BeEmpty())
 		})
 	})