vmagent and vmanomaly: create PDB per shard to guarantee proper application protection, before there was a single PDB, which could lead to unpredicted application disruptions. fixes #1548

Author: AndrewChubatiuk

api/operator/v1beta1/vmagent_types.go

@@ -435,6 +435,14 @@ func (cr *VMAgent) UnmarshalJSON(src []byte) error {
 	return nil
 }
 
+// GetShardCount returns shard count for vmagent
+func (cr *VMAgent) GetShardCount() int {
+	if cr == nil || cr.Spec.ShardCount == nil {
+		return 0
+	}
+	return *cr.Spec.ShardCount
+}
+
 // UnmarshalJSON implements json.Unmarshaler interface
 func (cr *VMAgentSpec) UnmarshalJSON(src []byte) error {
 	type pcr VMAgentSpec

docs/CHANGELOG.md

@@ -66,6 +66,7 @@ This change could be reverted by providing env variable `VM_USECUSTOMCONFIGRELOA
 * BUGFIX: [vmoperator](https://docs.victoriametrics.com/operator/): fix an issue where the return value from a couple of controllers was always `nil`. See [#1532](https://github.com/VictoriaMetrics/operator/pull/1532) for details.
 * BUGFIX: [VMCluster](https://docs.victoriametrics.com/operator/resources/vmcluster/): emit warning if `vmcluster.spec.vmselect.persistentVolume` is set, previously it was emitted for `vmcluster.spec.vmselect.storage`.
 * BUGFIX: [vmoperator](https://docs.victoriametrics.com/operator/): Prevent endless Service reconcile loop by correctly track changes to Service.spec.LoadBalancerClass. See this issue [#1550](https://github.com/VictoriaMetrics/operator/issues/1550) for details.
+* BUGFIX: [vmagent](https://docs.victoriametrics.com/operator/resources/vmagent/) and [vmanomaly](https://docs.victoriametrics.com/operator/resources/vmanomaly/): create PDB per shard to guarantee proper application protection. See [#1548](https://github.com/VictoriaMetrics/operator/issues/1548).
 
 
 ## [v0.63.0](https://github.com/VictoriaMetrics/operator/releases/tag/v0.63.0)

internal/controller/operator/factory/build/pdb.go

@@ -1,6 +1,9 @@
 package build
 
 import (
+	"fmt"
+	"strconv"
+
 	policyv1 "k8s.io/api/policy/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -9,6 +12,11 @@ import (
 
 // PodDisruptionBudget creates object for given CRD
 func PodDisruptionBudget(cr builderOpts, spec *vmv1beta1.EmbeddedPodDisruptionBudgetSpec) *policyv1.PodDisruptionBudget {
+	return PodDisruptionBudgetSharded(cr, spec, nil)
+}
+
+// PodDisruptionBudgetSharded creates object for given CRD and shard num
+func PodDisruptionBudgetSharded(cr builderOpts, spec *vmv1beta1.EmbeddedPodDisruptionBudgetSpec, num *int) *policyv1.PodDisruptionBudget {
 	pdb := policyv1.PodDisruptionBudget{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            cr.PrefixedName(),
@@ -25,6 +33,10 @@ func PodDisruptionBudget(cr builderOpts, spec *vmv1beta1.EmbeddedPodDisruptionBu
 			},
 		},
 	}
+	if num != nil {
+		pdb.Name = fmt.Sprintf("%s-%d", pdb.Name, *num)
+		pdb.Spec.Selector.MatchLabels[shardLabelName] = strconv.Itoa(*num)
+	}
 	if len(spec.UnhealthyPodEvictionPolicy) > 0 {
 		p := policyv1.UnhealthyPodEvictionPolicyType(spec.UnhealthyPodEvictionPolicy)
 		pdb.Spec.UnhealthyPodEvictionPolicy = &p

internal/controller/operator/factory/build/shard.go

@@ -0,0 +1,72 @@
+package build
+
+import (
+	"fmt"
+	"iter"
+	"strconv"
+
+	"github.com/VictoriaMetrics/operator/internal/controller/operator/factory/k8stools"
+)
+
+const (
+	ShardNumPlaceholder = "%SHARD_NUM%"
+	shardLabelName      = "shard-num"
+)
+
+// ShardNumIter iterates over shardCount in order defined in backward
+func ShardNumIter(backward bool, shardCount int) iter.Seq[*int] {
+	if shardCount <= 1 {
+		return func(yield func(*int) bool) {
+			yield(nil)
+		}
+	}
+	if backward {
+		return func(yield func(*int) bool) {
+			for shardCount > 0 {
+				shardCount--
+				num := shardCount
+				if !yield(&num) {
+					return
+				}
+			}
+		}
+	}
+	return func(yield func(*int) bool) {
+		for i := 0; i < shardCount; i++ {
+			if !yield(&i) {
+				return
+			}
+		}
+	}
+}
+
+func isSharded(shardCount int) bool {
+	return shardCount > 1
+}
+
+// ShardName adds shard suffix to a name if shards count is > 1
+func ShardName(name string, shardCount int) string {
+	if isSharded(shardCount) {
+		name = fmt.Sprintf("%s-%s", name, ShardNumPlaceholder)
+	}
+	return name
+}
+
+// ShardLabels adds shard label pair if shards count is > 1
+func ShardLabels(labels map[string]string, shardCount int) map[string]string {
+	if isSharded(shardCount) {
+		labels[shardLabelName] = "%SHARD_NUM%"
+	}
+	return labels
+}
+
+// RenderShard renders resource's placeholders with optional shard number placeholder in num is defined
+func RenderShard[T any](resource *T, placeholders map[string]string, num *int) (*T, error) {
+	if placeholders == nil {
+		placeholders = make(map[string]string)
+	}
+	if num != nil {
+		placeholders[ShardNumPlaceholder] = strconv.Itoa(*num)
+	}
+	return k8stools.RenderPlaceholders(resource, placeholders)
+}

internal/controller/operator/factory/build/shard_test.go

@@ -0,0 +1,35 @@
+package build
+
+import (
+	"slices"
+	"testing"
+)
+
+func TestShardNumIter(t *testing.T) {
+	f := func(backward bool, upperBound int) {
+		output := slices.Collect(ShardNumIter(backward, upperBound))
+		if upperBound <= 1 {
+			if len(output) != 1 || output[0] != nil {
+				t.Errorf("invalid ShardNumIter() values, want: [nil]; got: %v", output)
+			}
+			return
+		}
+		if upperBound > 1 && len(output) != upperBound {
+			t.Errorf("invalid ShardNumIter() items count, want: %d, got: %d", upperBound, len(output))
+		}
+		var lowerBound int
+		if backward {
+			lowerBound = upperBound - 1
+			upperBound = 0
+		} else {
+			upperBound--
+		}
+		if *output[0] != lowerBound || *output[len(output)-1] != upperBound {
+			t.Errorf("invalid ShardNumIter() bounds, want: [%d, %d], got: [%d, %d]", lowerBound, upperBound, *output[0], *output[len(output)-1])
+		}
+	}
+	f(true, 9)
+	f(false, 5)
+	f(false, 1)
+	f(true, 0)
+}

internal/controller/operator/factory/finalize/orphaned.go

@@ -3,8 +3,9 @@ package finalize
 import (
 	"context"
 
-	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -14,84 +15,71 @@ type orphanedCRD interface {
 }
 
 // RemoveOrphanedDeployments removes deployments detached from given object
-func RemoveOrphanedDeployments(ctx context.Context, rclient client.Client, cr orphanedCRD, keepDeployments map[string]struct{}) error {
-	deployToRemove, err := discoverDeploymentsByLabels(ctx, rclient, cr.GetNamespace(), cr.SelectorLabels())
-	if err != nil {
-		return err
-	}
-	for i := range deployToRemove {
-		dep := deployToRemove[i]
-		if _, ok := keepDeployments[dep.Name]; !ok {
-			// need to remove
-			if err := RemoveFinalizer(ctx, rclient, dep); err != nil {
-				return err
-			}
-			if err := SafeDelete(ctx, rclient, dep); err != nil {
-				return err
-			}
-		}
+func RemoveOrphanedDeployments(ctx context.Context, rclient client.Client, cr orphanedCRD, keepNames map[string]struct{}) error {
+	gvk := schema.GroupVersionKind{
+		Group:   "apps",
+		Version: "v1",
+		Kind:    "Deployment",
 	}
-	return nil
+	return removeOrphaned(ctx, rclient, cr, gvk, keepNames)
 }
 
-// discoverDeploymentsByLabels - returns deployments with given args.
-func discoverDeploymentsByLabels(ctx context.Context, rclient client.Client, ns string, selector map[string]string) ([]*appsv1.Deployment, error) {
-	var deps appsv1.DeploymentList
-	opts := client.ListOptions{
-		Namespace:     ns,
-		LabelSelector: labels.SelectorFromSet(selector),
-	}
-	if err := rclient.List(ctx, &deps, &opts); err != nil {
-		return nil, err
-	}
-	resp := make([]*appsv1.Deployment, 0, len(deps.Items))
-	for i := range deps.Items {
-		resp = append(resp, &deps.Items[i])
+// RemoveOrphanedSTSs removes deployments detached from given object
+func RemoveOrphanedSTSs(ctx context.Context, rclient client.Client, cr orphanedCRD, keepNames map[string]struct{}) error {
+	gvk := schema.GroupVersionKind{
+		Group:   "apps",
+		Version: "v1",
+		Kind:    "StatefulSet",
 	}
-	return resp, nil
+	return removeOrphaned(ctx, rclient, cr, gvk, keepNames)
 }
 
-// RemoveSvcArgs defines interface for service deletion
-type RemoveSvcArgs struct {
-	PrefixedName   func() string
-	SelectorLabels func() map[string]string
-	GetNameSpace   func() string
+// RemoveOrphanedPDBs removes PDBs detached from given object
+func RemoveOrphanedPDBs(ctx context.Context, rclient client.Client, cr orphanedCRD, keepNames map[string]struct{}) error {
+	gvk := schema.GroupVersionKind{
+		Group:   "policy",
+		Version: "v1",
+		Kind:    "PodDisruptionBudget",
+	}
+	return removeOrphaned(ctx, rclient, cr, gvk, keepNames)
 }
 
-// RemoveOrphanedSTSs removes deployments detached from given object
-func RemoveOrphanedSTSs(ctx context.Context, rclient client.Client, cr orphanedCRD, keepSTSNames map[string]struct{}) error {
-	deployToRemove, err := discoverSTSsByLabels(ctx, rclient, cr.GetNamespace(), cr.SelectorLabels())
+// removeOrphaned removes orphaned resources
+func removeOrphaned(ctx context.Context, rclient client.Client, cr orphanedCRD, gvk schema.GroupVersionKind, keepNames map[string]struct{}) error {
+	toRemove, err := getOrphaned(ctx, rclient, gvk, cr)
 	if err != nil {
 		return err
 	}
-	for i := range deployToRemove {
-		dep := deployToRemove[i]
-		if _, ok := keepSTSNames[dep.Name]; !ok {
+	for i := range toRemove {
+		item := toRemove[i]
+		if _, ok := keepNames[item.GetName()]; !ok {
 			// need to remove
-			if err := RemoveFinalizer(ctx, rclient, dep); err != nil {
+			if err := RemoveFinalizer(ctx, rclient, item); err != nil {
 				return err
 			}
-			if err := SafeDelete(ctx, rclient, dep); err != nil {
+			if err := SafeDelete(ctx, rclient, item); err != nil {
 				return err
 			}
 		}
 	}
 	return nil
 }
 
-// discoverDeploymentsByLabels - returns deployments with given args.
-func discoverSTSsByLabels(ctx context.Context, rclient client.Client, ns string, selector map[string]string) ([]*appsv1.StatefulSet, error) {
-	var deps appsv1.StatefulSetList
+// getOrphaned - returns resources by orphaned CR selector.
+func getOrphaned(ctx context.Context, rclient client.Client, gvk schema.GroupVersionKind, cr orphanedCRD) ([]client.Object, error) {
+	var l unstructured.UnstructuredList
+	l.SetGroupVersionKind(gvk)
 	opts := client.ListOptions{
-		Namespace:     ns,
-		LabelSelector: labels.SelectorFromSet(selector),
+		Namespace:     cr.GetNamespace(),
+		LabelSelector: labels.SelectorFromSet(cr.SelectorLabels()),
 	}
-	if err := rclient.List(ctx, &deps, &opts); err != nil {
+	if err := rclient.List(ctx, &l, &opts); err != nil {
 		return nil, err
 	}
-	resp := make([]*appsv1.StatefulSet, 0, len(deps.Items))
-	for i := range deps.Items {
-		resp = append(resp, &deps.Items[i])
+	resp := make([]client.Object, 0, len(l.Items))
+	for i := range l.Items {
+		item := &l.Items[i]
+		resp = append(resp, item)
 	}
 	return resp, nil
 }