vmagent and vmanomaly: create PDB per shard

Author: AndrewChubatiuk

Makefile

@@ -362,13 +362,13 @@ OPM = $(LOCALBIN)/opm-$(OPM_VERSION)
 YQ = $(LOCALBIN)/yq-$(YQ_VERSION)
 CRD_REF_DOCS = $(LOCALBIN)/crd-ref-docs-$(CRD_REF_DOCS_VERSION)
 GINKGO_BIN ?= $(LOCALBIN)/ginkgo-$(GINKGO_VERSION)
-GINKGO_VERSION ?= v2.23.0
+GINKGO_VERSION ?= v2.26.0
 
 ## Tool Versions
 KUSTOMIZE_VERSION ?= v5.6.0
 CONTROLLER_TOOLS_VERSION ?= v0.18.0
 ENVTEST_VERSION ?= release-0.20
-GOLANGCI_LINT_VERSION ?= v2.4.0
+GOLANGCI_LINT_VERSION ?= v2.5.0
 CODEGENERATOR_VERSION ?= v0.32.2
 KIND_VERSION ?= v0.27.0
 OLM_VERSION ?= 0.31.0

api/operator/v1beta1/vmagent_types.go

@@ -402,6 +402,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

@@ -25,6 +25,7 @@ aliases:
 
 * 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: [vmagent](https://docs.victoriametrics.com/operator/resources/vmagent/) and [vmanomaly](https://docs.victoriametrics.com/operator/resources/vmanomaly/): create PDB per shard. See [#1548](https://github.com/VictoriaMetrics/operator/issues/1548).
 
 
 ## [v0.63.0](https://github.com/VictoriaMetrics/operator/releases/tag/v0.63.0)

go.mod

@@ -15,14 +15,14 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/go-version v1.7.0
-	github.com/onsi/ginkgo/v2 v2.23.0
-	github.com/onsi/gomega v1.36.2
+	github.com/onsi/ginkgo/v2 v2.26.0
+	github.com/onsi/gomega v1.38.2
 	github.com/pires/go-proxyproto v0.8.0
 	github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring v0.84.0
 	github.com/prometheus/client_golang v1.22.0
 	github.com/stretchr/testify v1.10.0
 	go.uber.org/zap v1.27.0
-	golang.org/x/net v0.41.0
+	golang.org/x/net v0.43.0
 	golang.org/x/sync v0.16.0
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.33.2
@@ -53,7 +53,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/btree v1.1.3 // indirect
 	github.com/google/gnostic-models v0.6.9 // indirect
-	github.com/google/pprof v0.0.0-20250302191652-9094ed2288e7 // indirect
+	github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/jpillora/backoff v1.0.0 // indirect
@@ -80,16 +80,19 @@ require (
 	github.com/valyala/histogram v1.2.0 // indirect
 	github.com/valyala/quicktemplate v1.8.0 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
+	golang.org/x/mod v0.27.0 // indirect
 	golang.org/x/oauth2 v0.28.0 // indirect
-	golang.org/x/sys v0.33.0 // indirect
-	golang.org/x/term v0.32.0 // indirect
-	golang.org/x/text v0.27.0 // indirect
+	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/term v0.34.0 // indirect
+	golang.org/x/text v0.28.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
-	golang.org/x/tools v0.34.0 // indirect
+	golang.org/x/tools v0.36.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/protobuf v1.36.6 // indirect
+	google.golang.org/protobuf v1.36.7 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

go.sum

@@ -1,6 +1,7 @@
 package build
 
 import (
+	"iter"
 	"sort"
 )
 
@@ -36,3 +37,30 @@ func (s *keysSorter[T]) Swap(i, j int) {
 	s.target[i], s.target[j] = s.target[j], s.target[i]
 	s.sorter[i], s.sorter[j] = s.sorter[j], s.sorter[i]
 }
+
+// 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
+			}
+		}
+	}
+}

internal/controller/operator/factory/build/common.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/build/common_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	appsv1 "k8s.io/api/apps/v1"
+	policyv1 "k8s.io/api/policy/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -14,14 +15,14 @@ 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 {
+func RemoveOrphanedDeployments(ctx context.Context, rclient client.Client, cr orphanedCRD, keepNames 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 {
+		if _, ok := keepNames[dep.Name]; !ok {
 			// need to remove
 			if err := RemoveFinalizer(ctx, rclient, dep); err != nil {
 				return err
@@ -59,14 +60,14 @@ type RemoveSvcArgs struct {
 }
 
 // RemoveOrphanedSTSs removes deployments detached from given object
-func RemoveOrphanedSTSs(ctx context.Context, rclient client.Client, cr orphanedCRD, keepSTSNames map[string]struct{}) error {
+func RemoveOrphanedSTSs(ctx context.Context, rclient client.Client, cr orphanedCRD, keepNames map[string]struct{}) error {
 	deployToRemove, err := discoverSTSsByLabels(ctx, rclient, cr.GetNamespace(), cr.SelectorLabels())
 	if err != nil {
 		return err
 	}
 	for i := range deployToRemove {
 		dep := deployToRemove[i]
-		if _, ok := keepSTSNames[dep.Name]; !ok {
+		if _, ok := keepNames[dep.Name]; !ok {
 			// need to remove
 			if err := RemoveFinalizer(ctx, rclient, dep); err != nil {
 				return err
@@ -79,7 +80,7 @@ func RemoveOrphanedSTSs(ctx context.Context, rclient client.Client, cr orphanedC
 	return nil
 }
 
-// discoverDeploymentsByLabels - returns deployments with given args.
+// discoverSTSsByLabels - returns statefulsets with given args.
 func discoverSTSsByLabels(ctx context.Context, rclient client.Client, ns string, selector map[string]string) ([]*appsv1.StatefulSet, error) {
 	var deps appsv1.StatefulSetList
 	opts := client.ListOptions{
@@ -95,3 +96,41 @@ func discoverSTSsByLabels(ctx context.Context, rclient client.Client, ns string,
 	}
 	return resp, nil
 }
+
+// RemoveOrphanedPDBs removes PDBs detached from given object
+func RemoveOrphanedPDBs(ctx context.Context, rclient client.Client, cr orphanedCRD, keepNames map[string]struct{}) error {
+	pdbsToRemove, err := discoverPDBsByLabels(ctx, rclient, cr.GetNamespace(), cr.SelectorLabels())
+	if err != nil {
+		return err
+	}
+	for i := range pdbsToRemove {
+		dep := pdbsToRemove[i]
+		if _, ok := keepNames[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
+			}
+		}
+	}
+	return nil
+}
+
+// discoverPDBsByLabels - returns PDBs with given args.
+func discoverPDBsByLabels(ctx context.Context, rclient client.Client, ns string, selector map[string]string) ([]*policyv1.PodDisruptionBudget, error) {
+	var pdbs policyv1.PodDisruptionBudgetList
+	opts := client.ListOptions{
+		Namespace:     ns,
+		LabelSelector: labels.SelectorFromSet(selector),
+	}
+	if err := rclient.List(ctx, &pdbs, &opts); err != nil {
+		return nil, err
+	}
+	resp := make([]*policyv1.PodDisruptionBudget, 0, len(pdbs.Items))
+	for i := range pdbs.Items {
+		resp = append(resp, &pdbs.Items[i])
+	}
+	return resp, nil
+}