Repository: VictoriaMetrics/operator

Issue:
vmcluster: minimum downtime strategy
AFAIK, the operator currently supports only rolling updates where Pods of a StatefulSet get updated sequentially, this basically adheres to the "no dowtime strategy":
https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#no-downtime-strategy

However, it might be hard and quite expensive to achieve in practice, e.g. https://github.com/VictoriaMetrics/operator/issues/744, https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8591, etc. And the larger the cluster, the more painful it becomes, unfortunately.

Sometimes, it's more viable to keep VictoriaMetrics upgrades short and cheap at the cost of a small data loss. This is what the "minimum downtime strategy" allows to achieve:
https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#minimum-downtime-strategy

VictoriaMetrics is well-known for its flawless upgrade stability, meaning that upgrading from version A to version B of a given vmstorage node will almost certainly succeed without any compatibility issues regardless of the data inside. This is applicable even to downgrades, which is insane 🥇, thank you!

Thus, the idea is to make it possible to **upgrade all vmstorages at once**.

Hints from issue comments:
@f41gh7: The feature was included into [v0.61.0](https://docs.victoriametrics.com/operator/changelog/#v0610) release

Solution patch:
diff --git a/api/operator/v1/vlcluster_types.go b/api/operator/v1/vlcluster_types.go
index d2b6834c..9c9e662c 100644
--- a/api/operator/v1/vlcluster_types.go
+++ b/api/operator/v1/vlcluster_types.go
@@ -425,6 +425,9 @@ type VLStorage struct {
 	// Storage configures persistent volume for VLStorage
 	// +optional
 	Storage *vmv1beta1.StorageSpec `json:"storage,omitempty"`
+	// PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy
+	// +optional
+	PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"`
 
 	// MaintenanceInsertNodeIDs - excludes given node ids from insert requests routing, must contain pod suffixes - for pod-0, id will be 0 and etc.
 	// lets say, you have pod-0, pod-1, pod-2, pod-3. to exclude pod-0 and pod-3 from insert routing, define nodeIDs: [0,3].
diff --git a/api/operator/v1/vmanomaly_types.go b/api/operator/v1/vmanomaly_types.go
index ef57f58a..970338af 100644
--- a/api/operator/v1/vmanomaly_types.go
+++ b/api/operator/v1/vmanomaly_types.go
@@ -82,6 +82,9 @@ type VMAnomalySpec struct {
 	// Storage configures storage for StatefulSet
 	// +optional
 	Storage *vmv1beta1.StorageSpec `json:"storage,omitempty"`
+	// PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy
+	// +optional
+	PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"`
 	// RollingUpdateStrategy allows configuration for strategyType
 	// set it to RollingUpdate for disabling operator statefulSet rollingUpdate
 	// +optional
diff --git a/api/operator/v1/zz_generated.deepcopy.go b/api/operator/v1/zz_generated.deepcopy.go
index 5882a213..159cf429 100644
--- a/api/operator/v1/zz_generated.deepcopy.go
+++ b/api/operator/v1/zz_generated.deepcopy.go
@@ -537,6 +537,11 @@ func (in *VLStorage) DeepCopyInto(out *VLStorage) {
 		*out = new(v1beta1.StorageSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PersistentVolumeClaimRetentionPolicy != nil {
+		in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy
+		*out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy)
+		**out = **in
+	}
 	if in.MaintenanceInsertNodeIDs != nil {
 		in, out := &in.MaintenanceInsertNodeIDs, &out.MaintenanceInsertNodeIDs
 		*out = make([]int32, len(*in))
@@ -797,6 +802,11 @@ func (in *VMAnomalySpec) DeepCopyInto(out *VMAnomalySpec) {
 		*out = new(v1beta1.StorageSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PersistentVolumeClaimRetentionPolicy != nil {
+		in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy
+		*out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy)
+		**out = **in
+	}
 	if in.ClaimTemplates != nil {
 		in, out := &in.ClaimTemplates, &out.ClaimTemplates
 		*out = make([]corev1.PersistentVolumeClaim, len(*in))
diff --git a/api/operator/v1beta1/vmagent_types.go b/api/operator/v1beta1/vmagent_types.go
index ba0f7c41..deb77e42 100644
--- a/api/operator/v1beta1/vmagent_types.go
+++ b/api/operator/v1beta1/vmagent_types.go
@@ -301,6 +301,9 @@ type VMAgentSpec struct {
 	// set it to RollingUpdate for disabling operator statefulSet rollingUpdate
 	// +optional
 	StatefulRollingUpdateStrategy appsv1.StatefulSetUpdateStrategyType `json:"statefulRollingUpdateStrategy,omitempty"`
+	// PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy
+	// +optional
+	PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"`
 
 	// ClaimTemplates allows adding additional VolumeClaimTemplates for VMAgent in StatefulMode
 	ClaimTemplates []corev1.PersistentVolumeClaim `json:"claimTemplates,omitempty"`
@@ -676,10 +679,6 @@ func (cr *VMAgent) RelabelingAssetName() string {
 	return fmt.Sprintf("relabelings-assets-vmagent-%s", cr.Name)
 }
 
-func (cr *VMAgent) StreamAggrConfigName() string {
-	return fmt.Sprintf("stream-aggr-vmagent-%s", cr.Name)
-}
-
 func (cr *VMAgent) HealthPath() string {
 	return BuildPathWithPrefixFlag(cr.Spec.ExtraArgs, healthPath)
 }
diff --git a/api/operator/v1beta1/vmalertmanager_types.go b/api/operator/v1beta1/vmalertmanager_types.go
index 4fa39913..b42d8b3f 100644
--- a/api/operator/v1beta1/vmalertmanager_types.go
+++ b/api/operator/v1beta1/vmalertmanager_types.go
@@ -95,6 +95,9 @@ type VMAlertmanagerSpec struct {
 	// instances.
 	// +optional
 	Storage *StorageSpec `json:"storage,omitempty"`
+	// PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy
+	// +optional
+	PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"`
 
 	// ExternalURL the VMAlertmanager instances will be available under. This is
 	// necessary to generate correct URLs. This is necessary if VMAlertmanager is not
diff --git a/api/operator/v1beta1/vmcluster_types.go b/api/operator/v1beta1/vmcluster_types.go
index dee1c9f7..749f78d6 100644
--- a/api/operator/v1beta1/vmcluster_types.go
+++ b/api/operator/v1beta1/vmcluster_types.go
@@ -245,6 +245,9 @@ type VMSelect struct {
 	// +deprecated
 	// +optional
 	Storage *StorageSpec `json:"persistentVolume,omitempty"`
+	// PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy
+	// +optional
+	PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"`
 	// StorageSpec - add persistent volume claim for cacheMountPath
 	// its needed for persistent cache
 	// +optional
@@ -268,11 +271,17 @@ type VMSelect struct {
 	// Note, enabling this option disables vmselect to vmselect communication. In most cases it's not an issue.
 	// +optional
 	HPA *EmbeddedHPA `json:"hpa,omitempty"`
+
 	// RollingUpdateStrategy defines strategy for application updates
 	// Default is OnDelete, in this case operator handles update process
 	// Can be changed for RollingUpdate
 	// +optional
 	RollingUpdateStrategy appsv1.StatefulSetUpdateStrategyType `json:"rollingUpdateStrategy,omitempty"`
+	// RollingUpdateStrategyBehavior defines customized behavior for rolling updates.
+	// It applies if the RollingUpdateStrategy is set to OnDelete, which is the default.
+	// +optional
+	RollingUpdateStrategyBehavior *StatefulSetUpdateStrategyBehavior `json:"rollingUpdateStrategyBehavior,omitempty"`
+
 	// ClaimTemplates allows adding additional VolumeClaimTemplates for StatefulSet
 	ClaimTemplates []corev1.PersistentVolumeClaim `json:"claimTemplates,omitempty"`
 
@@ -408,6 +417,9 @@ type VMStorage struct {
 	// its useful for persistent cache
 	// +optional
 	Storage *StorageSpec `json:"storage,omitempty"`
+	// PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy
+	// +optional
+	PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"`
 
 	// VMInsertPort for VMInsert connections
 	// +optional
@@ -443,6 +455,10 @@ type VMStorage struct {
 	// Can be changed for RollingUpdate
 	// +optional
 	RollingUpdateStrategy appsv1.StatefulSetUpdateStrategyType `json:"rollingUpdateStrategy,omitempty"`
+	// RollingUpdateStrategyBehavior defines customized behavior for rolling updates.
+	// It applies if the RollingUpdateStrategy is set to OnDelete, which is the default.
+	// +optional
+	RollingUpdateStrategyBehavior *StatefulSetUpdateStrategyBehavior `json:"rollingUpdateStrategyBehavior,omitempty"`
 
 	// ClaimTemplates allows adding additional VolumeClaimTemplates for StatefulSet
 	ClaimTemplates []corev1.PersistentVolumeClaim `json:"claimTemplates,omitempty"`
@@ -595,6 +611,11 @@ func (cr *VMCluster) Validate() error {
 				return err
 			}
 		}
+		if vms.RollingUpdateStrategyBehavior != nil {
+			if err := vms.RollingUpdateStrategyBehavior.Validate(); err != nil {
+				return fmt.Errorf("vmselect: %w", err)
+			}
+		}
 	}
 	if cr.Spec.VMInsert != nil {
 		vmi := cr.Spec.VMInsert
@@ -617,6 +638,11 @@ func (cr *VMCluster) Validate() error {
 				return err
 			}
 		}
+		if vms.RollingUpdateStrategyBehavior != nil {
+			if err := vms.RollingUpdateStrategyBehavior.Validate(); err != nil {
+				return fmt.Errorf("vmstorage: %w", err)
+			}
+		}
 	}
 	if cr.Spec.RequestsLoadBalancer.Enabled {
 		rlb := cr.Spec.RequestsLoadBalancer.Spec
diff --git a/api/operator/v1beta1/vmextra_types.go b/api/operator/v1beta1/vmextra_types.go
index 504f37a7..37ce046a 100644
--- a/api/operator/v1beta1/vmextra_types.go
+++ b/api/operator/v1beta1/vmextra_types.go
@@ -1060,6 +1060,43 @@ func LastAppliedChangesAsPatch(crMeta metav1.ObjectMeta, spec any) (client.Patch
 
 }
 
+// StatefulSetUpdateStrategyBehavior customizes behavior for StatefulSet updates.
+type StatefulSetUpdateStrategyBehavior struct {
+	// MaxUnavailable defines the maximum number of pods that can be unavailable during the update.
+	// It can be specified as an absolute number (e.g. 2) or a percentage of the total pods (e.g. "50%").
+	// For example, if set to 100%, all pods will be upgraded at once, minimizing downtime when needed.
+	// +optional
+	MaxUnavailable *intstr.IntOrString `json:"maxUnavailable,omitempty"`
+}
+
+var percentPattern = regexp.MustCompile(`^\d+%$`)
+
+func (s *StatefulSetUpdateStrategyBehavior) Validate() error {
+	// If maxUnavailable is an integer, it must be >= 1
+	if s.MaxUnavailable.Type == intstr.Int {
+		if s.MaxUnavailable.IntValue() < 1 {
+			return fmt.Errorf("maxUnavailable must be greater than 0, got %d", s.MaxUnavailable.IntValue())
+		}
+	}
+
+	// If maxUnavailable is a percentage, it must be >= 1%
+	if s.MaxUnavailable.Type == intstr.String {
+		if !percentPattern.MatchString(s.MaxUnavailable.StrVal) {
+			return fmt.Errorf("maxUnavailable must be a percentage, got %s", s.MaxUnavailable.StrVal)
+		}
+
+		percent, err := strconv.Atoi(s.MaxUnavailable.StrVal[:len(s.MaxUnavailable.StrVal)-1])
+		if err != nil {
+			return err
+		}
+		if percent < 1 || percent > 100 {
+			return fmt.Errorf("maxUnavailable percentage must be between 1%% and 100%%, got %s", s.MaxUnavailable.StrVal)
+		}
+	}
+
+	return nil
+}
+
 // CommonDefaultableParams contains Application settings
 // with known values populated from operator configuration
 type CommonDefaultableParams struct {
diff --git a/api/operator/v1beta1/vmsingle_types.go b/api/operator/v1beta1/vmsingle_types.go
index 8860ab33..094ab6af 100644
--- a/api/operator/v1beta1/vmsingle_types.go
+++ b/api/operator/v1beta1/vmsingle_types.go
@@ -268,10 +268,6 @@ func (cr *VMSingle) PrefixedName() string {
 	return fmt.Sprintf("vmsingle-%s", cr.Name)
 }
 
-func (cr *VMSingle) StreamAggrConfigName() string {
-	return fmt.Sprintf("stream-aggr-vmsingle-%s", cr.Name)
-}
-
 // GetMetricPath returns prefixed path for metric requests
 func (cr *VMSingle) GetMetricPath() string {
 	return BuildPathWithPrefixFlag(cr.Spec.ExtraArgs, metricPath)
diff --git a/api/operator/v1beta1/zz_generated.deepcopy.go b/api/operator/v1beta1/zz_generated.deepcopy.go
index 6a144def..a31a3206 100644
--- a/api/operator/v1beta1/zz_generated.deepcopy.go
+++ b/api/operator/v1beta1/zz_generated.deepcopy.go
@@ -2969,6 +2969,26 @@ func (in *SnsConfig) DeepCopy() *SnsConfig {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *StatefulSetUpdateStrategyBehavior) DeepCopyInto(out *StatefulSetUpdateStrategyBehavior) {
+	*out = *in
+	if in.MaxUnavailable != nil {
+		in, out := &in.MaxUnavailable, &out.MaxUnavailable
+		*out = new(intstr.IntOrString)
+		**out = **in
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StatefulSetUpdateStrategyBehavior.
+func (in *StatefulSetUpdateStrategyBehavior) DeepCopy() *StatefulSetUpdateStrategyBehavior {
+	if in == nil {
+		return nil
+	}
+	out := new(StatefulSetUpdateStrategyBehavior)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *StaticConfig) DeepCopyInto(out *StaticConfig) {
 	*out = *in
@@ -4233,6 +4253,11 @@ func (in *VMAgentSpec) DeepCopyInto(out *VMAgentSpec) {
 		*out = new(StorageSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PersistentVolumeClaimRetentionPolicy != nil {
+		in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy
+		*out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy)
+		**out = **in
+	}
 	if in.ClaimTemplates != nil {
 		in, out := &in.ClaimTemplates, &out.ClaimTemplates
 		*out = make([]v1.PersistentVolumeClaim, len(*in))
@@ -4767,6 +4792,11 @@ func (in *VMAlertmanagerSpec) DeepCopyInto(out *VMAlertmanagerSpec) {
 		*out = new(StorageSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PersistentVolumeClaimRetentionPolicy != nil {
+		in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy
+		*out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy)
+		**out = **in
+	}
 	if in.AdditionalPeers != nil {
 		in, out := &in.AdditionalPeers, &out.AdditionalPeers
 		*out = make([]string, len(*in))
@@ -6096,6 +6126,11 @@ func (in *VMSelect) DeepCopyInto(out *VMSelect) {
 		*out = new(StorageSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PersistentVolumeClaimRetentionPolicy != nil {
+		in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy
+		*out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy)
+		**out = **in
+	}
 	if in.StorageSpec != nil {
 		in, out := &in.StorageSpec, &out.StorageSpec
 		*out = new(StorageSpec)
@@ -6126,6 +6161,11 @@ func (in *VMSelect) DeepCopyInto(out *VMSelect) {
 		*out = new(EmbeddedHPA)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.RollingUpdateStrategyBehavior != nil {
+		in, out := &in.RollingUpdateStrategyBehavior, &out.RollingUpdateStrategyBehavior
+		*out = new(StatefulSetUpdateStrategyBehavior)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.ClaimTemplates != nil {
 		in, out := &in.ClaimTemplates, &out.ClaimTemplates
 		*out = make([]v1.PersistentVolumeClaim, len(*in))
@@ -6487,6 +6527,11 @@ func (in *VMStorage) DeepCopyInto(out *VMStorage) {
 		*out = new(StorageSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.PersistentVolumeClaimRetentionPolicy != nil {
+		in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy
+		*out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy)
+		**out = **in
+	}
 	if in.VMBackup != nil {
 		in, out := &in.VMBackup, &out.VMBackup
 		*out = new(VMBackup)
@@ -6522,6 +6567,11 @@ func (in *VMStorage) DeepCopyInto(out *VMStorage) {
 		*out = make([]int32, len(*in))
 		copy(*out, *in)
 	}
+	if in.RollingUpdateStrategyBehavior != nil {
+		in, out := &in.RollingUpdateStrategyBehavior, &out.RollingUpdateStrategyBehavior
+		*out = new(StatefulSetUpdateStrategyBehavior)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.ClaimTemplates != nil {
 		in, out := &in.ClaimTemplates, &out.ClaimTemplates
 		*out = make([]v1.PersistentVolumeClaim, len(*in))
diff --git a/config/crd/overlay/crd.yaml b/config/crd/overlay/crd.yaml
index a180018f..255e79b4 100644
--- a/config/crd/overlay/crd.yaml
+++ b/config/crd/overlay/crd.yaml
@@ -2179,6 +2179,26 @@ spec:
                       Paused If set to true all actions on the underlying managed objects are not
                       going to be performed, except for delete actions.
                     type: boolean
+                  persistentVolumeClaimRetentionPolicy:
+                    description: PersistentVolumeClaimRetentionPolicy allows configuration
+                      of PVC rentention policy
+                    properties:
+                      whenDeleted:
+                        description: |-
+                          WhenDeleted specifies what happens to PVCs created from StatefulSet
+                          VolumeClaimTemplates when the StatefulSet is deleted. The default policy
+                          of `Retain` causes PVCs to not be affected by StatefulSet deletion. The
+                          `Delete` policy causes those PVCs to be deleted.
+                        type: string
+                      whenScaled:
+                        description: |-
+                          WhenScaled specifies what happens to PVCs created from StatefulSet
+                          VolumeClaimTemplates when the StatefulSet is scaled down. The default
+                          policy of `Retain` causes PVCs to not be affected by a scaledown. The
+                          `Delete` policy causes the associated PVCs for any excess pods above
+                          the replica count to be deleted.
+                        type: string
+                    type: object
                   podDisruptionBudget:
                     description: PodDisruptionBudget created by operator
                     properties:
@@ -5723,6 +5743,26 @@ spec:
                   Paused If set to true all actions on the underlying managed objects are not
                   going to be performed, except for delete actions.
                 type: boolean
+              persistentVolumeClaimRetentionPolicy:
+                description: PersistentVolumeClaimRetentionPolicy allows configuration
+                  of PVC rentention policy
+                properties:
+                  whenDeleted:
+                    description: |-
+                      WhenDeleted specifies what happens to PVCs created from StatefulSet
+                      VolumeClaimTemplates when the StatefulSet is deleted. The default policy
+                      of `Retain` causes PVCs to not be affected by StatefulSet deletion. The
+                      `Delete` policy causes those PVCs to be deleted.
+                    type: string
+                  whenScaled:
+                    description: |-
+                      WhenScaled specifies what happens to PVCs created from StatefulSet
+                      VolumeClaimTemplates when the StatefulSet is scaled down. The default
+                      policy of `Retain` causes PVCs to not be affected by a scaledown. The
+                      `Delete` policy causes the associated PVCs for any excess pods above
+                      the replica count to be deleted.
+                    type: string
+                type: object
               podDisruptionBudget:
                 description: PodDisruptionBudget created by operator
                 properties:
@@ -14580,6 +14620,26 @@ spec:
                   Paused If set to true all actions on the underlying managed objects are not
                   going to be performed, except for delete actions.
                 type: boolean
+              persistentVolumeClaimRetentionPolicy:
+                description: PersistentVolumeClaimRetentionPolicy allows configuration
+                  of PVC rentention policy
+                properties:
+                  whenDeleted:
+                    description: |-
+                      WhenDeleted specifies what happens to PVCs created from StatefulSet
+                      VolumeClaimTemplates when the StatefulSet is deleted. The default policy
+                      of `Retain` causes PVCs to not be affected by StatefulSet deletion. The
+                      `Delete` policy causes those PVCs to be deleted.
+                    type: string
+                  whenScaled:
+                    description: |-
+                      WhenScaled specifies what happens to PVCs created from StatefulSet
+                      VolumeClaimTemplates when the StatefulSet is scaled down. The default
+                      policy of `Retain` causes PVCs to not be affected by a scaledown. The
+                      `Delete` policy causes the associated PVCs for any excess pods above
+                      the replica count to be deleted.
+                    type: string
+                type: object
               podDisruptionBudget:
                 description: PodDisruptionBudget created by operator
                 properties:
@@ -18956,6 +19016,26 @@ spec:
                   Paused If set to true all actions on the underlying managed objects are not
                   going to be performed, except for delete actions.
                 type: boolean
+              persistentVolumeClaimRetentionPolicy:
+                description: PersistentVolumeClaimRetentionPolicy allows configuration
+                  of PVC rentention policy
+                properties:
+                  whenDeleted:
+                    description: |-
+                      WhenDeleted specifies what happens to PVCs created from StatefulSet
+                      VolumeClaimTemplates when the StatefulSet is deleted. The default policy
+                      of `Retain` causes PVCs to not be affected by StatefulSet deletion. The
+                      `Delete` policy causes those PVCs to be deleted.
+                    type: string
+                  whenScaled:
+                    description: |-
+                      WhenScaled specifies what happens to PVCs created from StatefulSet
+                      VolumeClaimTemplates when the StatefulSet is scaled down. The default
+                      policy of `Retain` causes PVCs to not be affected by a scaledown. The
+                      `Delete` policy causes the associated PVCs for any excess pods above
+                      the replica count to be deleted.
+                    type: string
+                type: object
               podDisruptionBudget:
                 description: PodDisruptionBudget created by operator
                 properties:
@@ -23605,6 +23685,26 @@ spec:
                         type: object
                         x-kubernetes-preserve-unknown-fields: true
                     type: object
+                  persistentVolumeClaimRetentionPolicy:
+                    description: PersistentVolumeClaimRetentionPolicy allows configuration
+                      of PVC rentention policy
+                    properties:
+                      whenDeleted:
+                        description: |-
+                          WhenDeleted specifies what happens to PVCs created from StatefulSet
+                          VolumeClaimTemplates when the StatefulSet is deleted. The default policy
+                          of `Retain` causes PVCs to not be affected by StatefulSet deletion. The
+                          `Delete` policy causes those PVCs to be deleted.
+                        type: string
+                      whenScaled:
+                        description: |-
+                          WhenScaled specifies what happens to PVCs created from StatefulSet
+                          VolumeClaimTemplates when the StatefulSet is scaled down. The default
+                          policy of `Retain` causes PVCs to not be affected by a scaledown. The
+                          `Delete` policy causes the associated PVCs for any excess pods above
+                          the replica count to be deleted.
+                        type: string
+                    type: object
                   podDisruptionBudget:
                     description: PodDisruptionBudget created by operator
                     properties:
@@ -23771,6 +23871,21 @@ spec:
                       Default is OnDelete, in this case operator handles update process
                       Can be changed for RollingUpdate
                     type: string
+                  rollingUpdateStrategyBehavior:
+                    description: |-
+                      RollingUpdateStrategyBehavior defines customized behavior for rolling updates.
+                      It applies if the RollingUpdateStrategy is set to OnDelete, which is the default.
+                    properties:
+                      maxUnavailable:
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        description: |-
+                          MaxUnavailable defines the maximum number of pods that can be unavailable during the update.
+                          It can be specified as an absolute number (e.g. 2) or a percentage of the total pods (e.g. "50%").
+                          For example, if set to 100%, all pods will be upgraded at once, minimizing downtime when needed.
+                        x-kubernetes-int-or-string: true
+                    type: object
                   runtimeClassName:
                     description: |-
                       RuntimeClassName - defines runtime class for kubernetes pod.
@@ -24834,6 +24949,26 @@ spec:
                       Paused If set to true all actions on the underlying managed objects are not
                       going to be performed, except for delete actions.
                     type: boolean
+                  persistentVolumeClaimRetentionPolicy:
+                    description: PersistentVolumeClaimRetentionPolicy allows configuration
+                      of PVC rentention policy
+                    properties:
+                      whenDeleted:
+                        description: |-
+                          WhenDeleted specifies what happens to PVCs created from StatefulSet
+                          VolumeClaimTemplates when the StatefulSet is deleted. The default policy
+                          of `Retain` causes PVCs to not be affected by StatefulSet deletion. The
+                          `Delete` policy causes those PVCs to be deleted.
+                        type: string
+                      whenScaled:
+                        description: |-
+                          WhenScaled specifies what happens to PVCs created from StatefulSet
+                          VolumeClaimTemplates when the StatefulSet is scaled down. The default
+                          policy of `Retain` causes PVCs to not be affected by a scaledown. The
+                          `Delete` policy causes the associated PVCs for any excess pods above
+                          the replica count to be deleted.
+                        type: string
+                    type: object
                   podDisruptionBudget:
                     description: PodDisruptionBudget created by operator
                     properties:
@@ -25000,6 +25135,21 @@ spec:
                       Default is OnDelete, in this case operator handles update process
                       Can be changed for RollingUpdate
                     type: string
+                  rollingUpdateStrategyBehavior:
+                    description: |-
+                      RollingUpdateStrategyBehavior defines customized behavior for rolling updates.
+                      It applies if the RollingUpdateStrategy is set to OnDelete, which is the default.
+                    properties:
+                      maxUnavailable:
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        description: |-
+                          MaxUnavailable defines the maximum number of pods that can be unavailable during the update.
+                          It can be specified as an absolute number (e.g. 2) or a percentage of the total pods (e.g. "50%").
+                          For example, if set to 100%, all pods will be upgraded at once, minimizing downtime when needed.
+                        x-kubernetes-int-or-string: true
+                    type: object
                   runtimeClassName:
                     description: |-
                       RuntimeClassName - defines runtime class for kubernetes pod.
diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml
index d7f7e1d3..f882577e 100644
--- a/config/rbac/role.yaml
+++ b/config/rbac/role.yaml
@@ -24,7 +24,7 @@ rules:
   - events
   - persistentvolumeclaims
   - persistentvolumeclaims/finalizers
-  - pods
+  - pods/eviction
   - secrets
   - secrets/finalizers
   - services
@@ -37,6 +37,7 @@ rules:
   - ""
   resources:
   - configmaps/status
+  - pods
   - nodes
   - nodes/proxy
   - nodes/metrics
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index d6981bac..e550f14d 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -14,11 +14,14 @@ aliases:
 ## tip
 
 * Dependency: [vmoperator](https://docs.victoriametrics.com/operator/): Updated default versions for VL apps to [v1.25.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.25.0).
+* Dependency: [vmoperator](https://docs.victoriametrics.com/operator/): Updated default versions for VM apps to [v1.121.0](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/victoriametrics/changelog/CHANGELOG.md#v11210) version.
 
 * FEATURE: [vmanomaly](https://docs.victoriametrics.com/anomaly-detection/): update shards concurrently. See [#1434](https://github.com/VictoriaMetrics/operator/issues/1434).
 * FEATURE: [vmoperator](https://docs.victoriametrics.com/operator/): deprecated acceptEula parameter in VMBackup. See [#1319](https://github.com/VictoriaMetrics/operator/issues/1319).
+* FEATURE: [vmoperator](https://docs.victoriametrics.com/operator/): add new field persistentVolumeClaimRetentionPolicy in VMAnomaly, VMCluster, VMAlertmanager and VMAgent CRs, that support StatefulSet mode. See [#1226](https://github.com/VictoriaMetrics/operator/issues/1226).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/operator/resources/vmagent/): added AWS section to VMAgent remoteWrite spec. See [#928](https://github.com/VictoriaMetrics/operator/issues/928).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/operator/resources/vmagent/): introduce global scrape config for VMAgent. See [#1179](https://github.com/VictoriaMetrics/operator/issues/1179).
+* FEATURE: [vmcluster](https://docs.victoriametrics.com/operator/resources/vmcluster/): added the `maxUnavailable` field to VMStorage and VMSelect specs to allow customization of rolling update behavior. See [#1457](https://github.com/VictoriaMetrics/operator/issues/1457).
 
 ## [v0.60.2](https://github.com/VictoriaMetrics/operator/releases/tag/v0.60.2)
 
diff --git a/docs/api.md b/docs/api.md
index 17bfeb34..83c86f60 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -385,6 +385,7 @@ Appears in: [VLClusterSpec](#vlclusterspec)
 | minReadySeconds<a href="#vlstorage-minreadyseconds" id="vlstorage-minreadyseconds">#</a><br/>_integer_ | _(Optional)_<br/>MinReadySeconds defines a minimum number of seconds to wait before starting update next pod<br />if previous in healthy state<br />Has no effect for VLogs and VMSingle |
 | nodeSelector<a href="#vlstorage-nodeselector" id="vlstorage-nodeselector">#</a><br/>_object (keys:string, values:string)_ | _(Optional)_<br/>NodeSelector Define which Nodes the Pods are scheduled on. |
 | paused<a href="#vlstorage-paused" id="vlstorage-paused">#</a><br/>_boolean_ | _(Optional)_<br/>Paused If set to true all actions on the underlying managed objects are not<br />going to be performed, except for delete actions. |
+| persistentVolumeClaimRetentionPolicy<a href="#vlstorage-persistentvolumeclaimretentionpolicy" id="vlstorage-persistentvolumeclaimretentionpolicy">#</a><br/>_[StatefulSetPersistentVolumeClaimRetentionPolicy](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetpersistentvolumeclaimretentionpolicy-v1-apps)_ | _(Optional)_<br/>PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy |
 | podDisruptionBudget<a href="#vlstorage-poddisruptionbudget" id="vlstorage-poddisruptionbudget">#</a><br/>_[EmbeddedPodDisruptionBudgetSpec](#embeddedpoddisruptionbudgetspec)_ | _(Optional)_<br/>PodDisruptionBudget created by operator |
 | podMetadata<a href="#vlstorage-podmetadata" id="vlstorage-podmetadata">#</a><br/>_[EmbeddedObjectMetadata](#embeddedobjectmetadata)_ | _(Required)_<br/>PodMetadata configures Labels and Annotations which are propagated to the VLStorage pods. |
 | port<a href="#vlstorage-port" id="vlstorage-port">#</a><br/>_string_ | _(Optional)_<br/>Port listen address |
@@ -564,6 +565,7 @@ Appears in: [VMAnomaly](#vmanomaly)
 | monitoring<a href="#vmanomalyspec-monitoring" id="vmanomalyspec-monitoring">#</a><br/>_[VMAnomalyMonitoringSpec](#vmanomalymonitoringspec)_ | _(Required)_<br/>Monitoring configures how expose anomaly metrics<br />See https://docs.victoriametrics.com/anomaly-detection/components/monitoring/ |
 | nodeSelector<a href="#vmanomalyspec-nodeselector" id="vmanomalyspec-nodeselector">#</a><br/>_object (keys:string, values:string)_ | _(Optional)_<br/>NodeSelector Define which Nodes the Pods are scheduled on. |
 | paused<a href="#vmanomalyspec-paused" id="vmanomalyspec-paused">#</a><br/>_boolean_ | _(Optional)_<br/>Paused If set to true all actions on the underlying managed objects are not<br />going to be performed, except for delete actions. |
+| persistentVolumeClaimRetentionPolicy<a href="#vmanomalyspec-persistentvolumeclaimretentionpolicy" id="vmanomalyspec-persistentvolumeclaimretentionpolicy">#</a><br/>_[StatefulSetPersistentVolumeClaimRetentionPolicy](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetpersistentvolumeclaimretentionpolicy-v1-apps)_ | _(Optional)_<br/>PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy |
 | podDisruptionBudget<a href="#vmanomalyspec-poddisruptionbudget" id="vmanomalyspec-poddisruptionbudget">#</a><br/>_[EmbeddedPodDisruptionBudgetSpec](#embeddedpoddisruptionbudgetspec)_ | _(Optional)_<br/>PodDisruptionBudget created by operator |
 | podMetadata<a href="#vmanomalyspec-podmetadata" id="vmanomalyspec-podmetadata">#</a><br/>_[EmbeddedObjectMetadata](#embeddedobjectmetadata)_ | _(Optional)_<br/>PodMetadata configures Labels and Annotations which are propagated to the vmanomaly pods. |
 | port<a href="#vmanomalyspec-port" id="vmanomalyspec-port">#</a><br/>_string_ | _(Optional)_<br/>Port listen address |
@@ -2318,6 +2320,19 @@ Appears in: [Receiver](#receiver)
 | topic_arn<a href="#snsconfig-topic_arn" id="snsconfig-topic_arn">#</a><br/>_string_ | _(Optional)_<br/>SNS topic ARN, either specify this, phone_number or target_arn |
 
 
+#### StatefulSetUpdateStrategyBehavior
+
+
+
+StatefulSetUpdateStrategyBehavior customizes behavior for StatefulSet updates.
+
+Appears in: [VMSelect](#vmselect), [VMStorage](#vmstorage)
+
+| Field | Description |
+| --- | --- |
+| maxUnavailable<a href="#statefulsetupdatestrategybehavior-maxunavailable" id="statefulsetupdatestrategybehavior-maxunavailable">#</a><br/>_[IntOrString](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#intorstring-intstr-util)_ | _(Optional)_<br/>MaxUnavailable defines the maximum number of pods that can be unavailable during the update.<br />It can be specified as an absolute number (e.g. 2) or a percentage of the total pods (e.g. "50%").<br />For example, if set to 100%, all pods will be upgraded at once, minimizing downtime when needed. |
+
+
 #### StaticConfig
 
 
@@ -2916,6 +2931,7 @@ Appears in: [VMAgent](#vmagent)
 | overrideHonorLabels<a href="#vmagentspec-overridehonorlabels" id="vmagentspec-overridehonorlabels">#</a><br/>_boolean_ | _(Optional)_<br/>OverrideHonorLabels if set to true overrides all user configured honor_labels.<br />If HonorLabels is set in scrape objects  to true, this overrides honor_labels to false. |
 | overrideHonorTimestamps<a href="#vmagentspec-overridehonortimestamps" id="vmagentspec-overridehonortimestamps">#</a><br/>_boolean_ | _(Optional)_<br/>OverrideHonorTimestamps allows to globally enforce honoring timestamps in all scrape configs. |
 | paused<a href="#vmagentspec-paused" id="vmagentspec-paused">#</a><br/>_boolean_ | _(Optional)_<br/>Paused If set to true all actions on the underlying managed objects are not<br />going to be performed, except for delete actions. |
+| persistentVolumeClaimRetentionPolicy<a href="#vmagentspec-persistentvolumeclaimretentionpolicy" id="vmagentspec-persistentvolumeclaimretentionpolicy">#</a><br/>_[StatefulSetPersistentVolumeClaimRetentionPolicy](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetpersistentvolumeclaimretentionpolicy-v1-apps)_ | _(Optional)_<br/>PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy |
 | podDisruptionBudget<a href="#vmagentspec-poddisruptionbudget" id="vmagentspec-poddisruptionbudget">#</a><br/>_[EmbeddedPodDisruptionBudgetSpec](#embeddedpoddisruptionbudgetspec)_ | _(Optional)_<br/>PodDisruptionBudget created by operator |
 | podMetadata<a href="#vmagentspec-podmetadata" id="vmagentspec-podmetadata">#</a><br/>_[EmbeddedObjectMetadata](#embeddedobjectmetadata)_ | _(Optional)_<br/>PodMetadata configures Labels and Annotations which are propagated to the vmagent pods. |
 | podScrapeNamespaceSelector<a href="#vmagentspec-podscrapenamespaceselector" id="vmagentspec-podscrapenamespaceselector">#</a><br/>_[LabelSelector](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#labelselector-v1-meta)_ | _(Optional)_<br/>PodScrapeNamespaceSelector defines Namespaces to be selected for VMPodScrape discovery.<br />Works in combination with Selector.<br />NamespaceSelector nil - only objects at VMAgent namespace.<br />Selector nil - only objects at NamespaceSelector namespaces.<br />If both nil - behaviour controlled by selectAllByDefault |
@@ -3243,6 +3259,7 @@ Appears in: [VMAlertmanager](#vmalertmanager)
 | minReadySeconds<a href="#vmalertmanagerspec-minreadyseconds" id="vmalertmanagerspec-minreadyseconds">#</a><br/>_integer_ | _(Optional)_<br/>MinReadySeconds defines a minimum number of seconds to wait before starting update next pod<br />if previous in healthy state<br />Has no effect for VLogs and VMSingle |
 | nodeSelector<a href="#vmalertmanagerspec-nodeselector" id="vmalertmanagerspec-nodeselector">#</a><br/>_object (keys:string, values:string)_ | _(Optional)_<br/>NodeSelector Define which Nodes the Pods are scheduled on. |
 | paused<a href="#vmalertmanagerspec-paused" id="vmalertmanagerspec-paused">#</a><br/>_boolean_ | _(Optional)_<br/>Paused If set to true all actions on the underlying managed objects are not<br />going to be performed, except for delete actions. |
+| persistentVolumeClaimRetentionPolicy<a href="#vmalertmanagerspec-persistentvolumeclaimretentionpolicy" id="vmalertmanagerspec-persistentvolumeclaimretentionpolicy">#</a><br/>_[StatefulSetPersistentVolumeClaimRetentionPolicy](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetpersistentvolumeclaimretentionpolicy-v1-apps)_ | _(Optional)_<br/>PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy |
 | podDisruptionBudget<a href="#vmalertmanagerspec-poddisruptionbudget" id="vmalertmanagerspec-poddisruptionbudget">#</a><br/>_[EmbeddedPodDisruptionBudgetSpec](#embeddedpoddisruptionbudgetspec)_ | _(Optional)_<br/>PodDisruptionBudget created by operator |
 | podMetadata<a href="#vmalertmanagerspec-podmetadata" id="vmalertmanagerspec-podmetadata">#</a><br/>_[EmbeddedObjectMetadata](#embeddedobjectmetadata)_ | _(Optional)_<br/>PodMetadata configures Labels and Annotations which are propagated to the alertmanager pods. |
 | port<a href="#vmalertmanagerspec-port" id="vmalertmanagerspec-port">#</a><br/>_string_ | _(Optional)_<br/>Port listen address |
@@ -3983,6 +4000,7 @@ Appears in: [VMClusterSpec](#vmclusterspec)
 | nodeSelector<a href="#vmselect-nodeselector" id="vmselect-nodeselector">#</a><br/>_object (keys:string, values:string)_ | _(Optional)_<br/>NodeSelector Define which Nodes the Pods are scheduled on. |
 | paused<a href="#vmselect-paused" id="vmselect-paused">#</a><br/>_boolean_ | _(Optional)_<br/>Paused If set to true all actions on the underlying managed objects are not<br />going to be performed, except for delete actions. |
 | persistentVolume<a href="#vmselect-persistentvolume" id="vmselect-persistentvolume">#</a><br/>_[StorageSpec](#storagespec)_ | _(Optional)_<br/>Storage - add persistent volume for cacheMountPath<br />its useful for persistent cache<br />use storage instead of persistentVolume. |
+| persistentVolumeClaimRetentionPolicy<a href="#vmselect-persistentvolumeclaimretentionpolicy" id="vmselect-persistentvolumeclaimretentionpolicy">#</a><br/>_[StatefulSetPersistentVolumeClaimRetentionPolicy](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetpersistentvolumeclaimretentionpolicy-v1-apps)_ | _(Optional)_<br/>PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy |
 | podDisruptionBudget<a href="#vmselect-poddisruptionbudget" id="vmselect-poddisruptionbudget">#</a><br/>_[EmbeddedPodDisruptionBudgetSpec](#embeddedpoddisruptionbudgetspec)_ | _(Optional)_<br/>PodDisruptionBudget created by operator |
 | podMetadata<a href="#vmselect-podmetadata" id="vmselect-podmetadata">#</a><br/>_[EmbeddedObjectMetadata](#embeddedobjectmetadata)_ | _(Required)_<br/>PodMetadata configures Labels and Annotations which are propagated to the VMSelect pods. |
 | port<a href="#vmselect-port" id="vmselect-port">#</a><br/>_string_ | _(Optional)_<br/>Port listen address |
@@ -3992,6 +4010,7 @@ Appears in: [VMClusterSpec](#vmclusterspec)
 | resources<a href="#vmselect-resources" id="vmselect-resources">#</a><br/>_[ResourceRequirements](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#resourcerequirements-v1-core)_ | _(Optional)_<br/>Resources container resource request and limits, https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/<br />if not defined default resources from operator config will be used |
 | revisionHistoryLimitCount<a href="#vmselect-revisionhistorylimitcount" id="vmselect-revisionhistorylimitcount">#</a><br/>_integer_ | _(Optional)_<br/>The number of old ReplicaSets to retain to allow rollback in deployment or<br />maximum number of revisions that will be maintained in the Deployment revision history.<br />Has no effect at StatefulSets<br />Defaults to 10. |
 | rollingUpdateStrategy<a href="#vmselect-rollingupdatestrategy" id="vmselect-rollingupdatestrategy">#</a><br/>_[StatefulSetUpdateStrategyType](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetupdatestrategytype-v1-apps)_ | _(Optional)_<br/>RollingUpdateStrategy defines strategy for application updates<br />Default is OnDelete, in this case operator handles update process<br />Can be changed for RollingUpdate |
+| rollingUpdateStrategyBehavior<a href="#vmselect-rollingupdatestrategybehavior" id="vmselect-rollingupdatestrategybehavior">#</a><br/>_[StatefulSetUpdateStrategyBehavior](#statefulsetupdatestrategybehavior)_ | _(Optional)_<br/>RollingUpdateStrategyBehavior defines customized behavior for rolling updates.<br />It applies if the RollingUpdateStrategy is set to OnDelete, which is the default. |
 | runtimeClassName<a href="#vmselect-runtimeclassname" id="vmselect-runtimeclassname">#</a><br/>_string_ | _(Optional)_<br/>RuntimeClassName - defines runtime class for kubernetes pod.<br />https://kubernetes.io/docs/concepts/containers/runtime-class/ |
 | schedulerName<a href="#vmselect-schedulername" id="vmselect-schedulername">#</a><br/>_string_ | _(Optional)_<br/>SchedulerName - defines kubernetes scheduler name |
 | secrets<a href="#vmselect-secrets" id="vmselect-secrets">#</a><br/>_string array_ | _(Optional)_<br/>Secrets is a list of Secrets in the same namespace as the Application<br />object, which shall be mounted into the Application container<br />at /etc/vm/secrets/SECRET_NAME folder |
@@ -4197,6 +4216,7 @@ Appears in: [VMClusterSpec](#vmclusterspec)
 | minReadySeconds<a href="#vmstorage-minreadyseconds" id="vmstorage-minreadyseconds">#</a><br/>_integer_ | _(Optional)_<br/>MinReadySeconds defines a minimum number of seconds to wait before starting update next pod<br />if previous in healthy state<br />Has no effect for VLogs and VMSingle |
 | nodeSelector<a href="#vmstorage-nodeselector" id="vmstorage-nodeselector">#</a><br/>_object (keys:string, values:string)_ | _(Optional)_<br/>NodeSelector Define which Nodes the Pods are scheduled on. |
 | paused<a href="#vmstorage-paused" id="vmstorage-paused">#</a><br/>_boolean_ | _(Optional)_<br/>Paused If set to true all actions on the underlying managed objects are not<br />going to be performed, except for delete actions. |
+| persistentVolumeClaimRetentionPolicy<a href="#vmstorage-persistentvolumeclaimretentionpolicy" id="vmstorage-persistentvolumeclaimretentionpolicy">#</a><br/>_[StatefulSetPersistentVolumeClaimRetentionPolicy](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetpersistentvolumeclaimretentionpolicy-v1-apps)_ | _(Optional)_<br/>PersistentVolumeClaimRetentionPolicy allows configuration of PVC rentention policy |
 | podDisruptionBudget<a href="#vmstorage-poddisruptionbudget" id="vmstorage-poddisruptionbudget">#</a><br/>_[EmbeddedPodDisruptionBudgetSpec](#embeddedpoddisruptionbudgetspec)_ | _(Optional)_<br/>PodDisruptionBudget created by operator |
 | podMetadata<a href="#vmstorage-podmetadata" id="vmstorage-podmetadata">#</a><br/>_[EmbeddedObjectMetadata](#embeddedobjectmetadata)_ | _(Required)_<br/>PodMetadata configures Labels and Annotations which are propagated to the VMStorage pods. |
 | port<a href="#vmstorage-port" id="vmstorage-port">#</a><br/>_string_ | _(Optional)_<br/>Port listen address |
@@ -4206,6 +4226,7 @@ Appears in: [VMClusterSpec](#vmclusterspec)
 | resources<a href="#vmstorage-resources" id="vmstorage-resources">#</a><br/>_[ResourceRequirements](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#resourcerequirements-v1-core)_ | _(Optional)_<br/>Resources container resource request and limits, https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/<br />if not defined default resources from operator config will be used |
 | revisionHistoryLimitCount<a href="#vmstorage-revisionhistorylimitcount" id="vmstorage-revisionhistorylimitcount">#</a><br/>_integer_ | _(Optional)_<br/>The number of old ReplicaSets to retain to allow rollback in deployment or<br />maximum number of revisions that will be maintained in the Deployment revision history.<br />Has no effect at StatefulSets<br />Defaults to 10. |
 | rollingUpdateStrategy<a href="#vmstorage-rollingupdatestrategy" id="vmstorage-rollingupdatestrategy">#</a><br/>_[StatefulSetUpdateStrategyType](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#statefulsetupdatestrategytype-v1-apps)_ | _(Optional)_<br/>RollingUpdateStrategy defines strategy for application updates<br />Default is OnDelete, in this case operator handles update process<br />Can be changed for RollingUpdate |
+| rollingUpdateStrategyBehavior<a href="#vmstorage-rollingupdatestrategybehavior" id="vmstorage-rollingupdatestrategybehavior">#</a><br/>_[StatefulSetUpdateStrategyBehavior](#statefulsetupdatestrategybehavior)_ | _(Optional)_<br/>RollingUpdateStrategyBehavior defines customized behavior for rolling updates.<br />It applies if the RollingUpdateStrategy is set to OnDelete, which is the default. |
 | runtimeClassName<a href="#vmstorage-runtimeclassname" id="vmstorage-runtimeclassname">#</a><br/>_string_ | _(Optional)_<br/>RuntimeClassName - defines runtime class for kubernetes pod.<br />https://kubernetes.io/docs/concepts/containers/runtime-class/ |
 | schedulerName<a href="#vmstorage-schedulername" id="vmstorage-schedulername">#</a><br/>_string_ | _(Optional)_<br/>SchedulerName - defines kubernetes scheduler name |
 | secrets<a href="#vmstorage-secrets" id="vmstorage-secrets">#</a><br/>_string array_ | _(Optional)_<br/>Secrets is a list of Secrets in the same namespace as the Application<br />object, which shall be mounted into the Application container<br />at /etc/vm/secrets/SECRET_NAME folder |
diff --git a/docs/env.md b/docs/env.md
index 30d12e6a..27f721f8 100644
--- a/docs/env.md
+++ b/docs/env.md
@@ -1,6 +1,6 @@
 | Environment variables |
 | --- |
-| VM_METRICS_VERSION: `v1.120.0` <a href="#variables-vm-metrics-version" id="variables-vm-metrics-version">#</a> |
+| VM_METRICS_VERSION: `v1.121.0` <a href="#variables-vm-metrics-version" id="variables-vm-metrics-version">#</a> |
 | VM_LOGS_VERSION: `v1.25.0` <a href="#variables-vm-logs-version" id="variables-vm-logs-version">#</a> |
 | VM_ANOMALY_VERSION: `v1.24.1` <a href="#variables-vm-anomaly-version" id="variables-vm-anomaly-version">#</a> |
 | VM_USECUSTOMCONFIGRELOADER: `false` <a href="#variables-vm-usecustomconfigreloader" id="variables-vm-usecustomconfigreloader">#</a><br>enables custom config reloader for vmauth and vmagent, it should speed-up config reloading process. |
diff --git a/go.mod b/go.mod
index 5238c295..b884ac12 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,7 @@ module github.com/VictoriaMetrics/operator
 go 1.24.4
 
 require (
+	github.com/Masterminds/semver/v3 v3.4.0
 	github.com/VictoriaMetrics/VictoriaMetrics v1.116.0
 	github.com/VictoriaMetrics/metrics v1.35.3
 	github.com/VictoriaMetrics/metricsql v0.84.5
@@ -34,7 +35,6 @@ require (
 )
 
 require (
-	github.com/Masterminds/semver/v3 v3.4.0 // indirect
 	github.com/aws/aws-sdk-go v1.55.6 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bmatcuk/doublestar/v4 v4.8.1 // indirect
diff --git a/internal/config/config.go b/internal/config/config.go
index ac5cf1d5..249c96c8 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -30,7 +30,7 @@ var (
 	//
 	// DO NOT FORGET TO MODIFY VERSIONS IN BaseOperatorConf
 	defaultEnvs = map[string]string{
-		"VM_METRICS_VERSION": "v1.120.0",
+		"VM_METRICS_VERSION": "v1.121.0",
 		"VM_LOGS_VERSION":    "v1.25.0",
 		"VM_ANOMALY_VERSION": "v1.24.1",
 	}
@@ -98,7 +98,7 @@ type BaseOperatorConf struct {
 	//
 	// DO NOT FORGET TO MODIFY VERSIONS IN defaultEnvs
 
-	MetricsVersion string `default:"v1.120.0" env:"METRICS_VERSION"`
+	MetricsVersion string `default:"v1.121.0" env:"METRICS_VERSION"`
 	LogsVersion    string `default:"v1.25.0" env:"LOGS_VERSION"`
 	AnomalyVersion string `default:"v1.24.1" env:"ANOMALY_VERSION"`
 
diff --git a/internal/controller/operator/factory/build/build.go b/internal/controller/operator/factory/build/build.go
index 790d7a3e..e5f7f1ab 100644
--- a/internal/controller/operator/factory/build/build.go
+++ b/internal/controller/operator/factory/build/build.go
@@ -3,6 +3,8 @@ package build
 import (
 	"fmt"
 	"path"
+	"strconv"
+	"strings"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -89,3 +91,114 @@ func LicenseVolumeTo(volumes []corev1.Volume, mounts []corev1.VolumeMount, l *vm
 	})
 	return volumes, mounts
 }
+
+func NewFlag(name, empty string) *Flag {
+	return &Flag{
+		name:  name,
+		empty: empty,
+	}
+}
+
+type Flag struct {
+	isSet bool
+	idx   int
+	name  string
+	value string
+	empty string
+}
+
+func (f *Flag) Add(value string, idx int) {
+	if idx > 0 {
+		if len(f.value) == 0 {
+			f.value += f.empty
+		}
+		f.value += ","
+		f.value += strings.Repeat(f.empty+",", idx-f.idx-1)
+		f.idx = idx
+	}
+	f.value += value
+	if value != f.empty && !f.isSet {
+		f.isSet = true
+	}
+}
+
+func (f *Flag) IsSet() bool {
+	return f.isSet
+}
+
+func AppendFlagsToArgs(args []string, total int, fss ...*Flag) []string {
+	for _, f := range fss {
+		if f.isSet {
+			args = append(args, f.name+"="+f.value+strings.Repeat(","+f.empty, total-f.idx-1))
+		}
+	}
+	return args
+}
+
+// StreamAggrArgsTo conditionally adds stream aggregation commandline args into given args
+func StreamAggrArgsTo(args []string, prefix string, keys []string, cs ...*vmv1beta1.StreamAggrConfig) []string {
+	if len(cs) == 0 {
+		return args
+	}
+
+	configFlag := NewFlag(fmt.Sprintf("-%s.config", prefix), "")
+	keepInputFlag := NewFlag(fmt.Sprintf("-%s.keepInput", prefix), "false")
+	dropInputFlag := NewFlag(fmt.Sprintf("-%s.dropInput", prefix), "false")
+	ignoreFirstIntervalsFlag := NewFlag(fmt.Sprintf("-%s.ignoreFirstIntervals", prefix), "0")
+	ignoreOldSamplesFlag := NewFlag(fmt.Sprintf("-%s.ignoreOldSamples", prefix), "false")
+	enableWindowsFlag := NewFlag(fmt.Sprintf("-%s.enableWindows", prefix), "false")
+	dedupIntervalFlag := NewFlag(fmt.Sprintf("-%s.dedupInterval", prefix), "")
+	dropInputLabelsFlag := NewFlag(fmt.Sprintf("-%s.dropInputLabels", prefix), "")
+
+	for i, c := range cs {
+		if c == nil {
+			continue
+		}
+		if c.HasAnyRule() {
+			configFlag.Add(path.Join(vmv1beta1.StreamAggrConfigDir, keys[i]), i)
+			keepInputFlag.Add(strconv.FormatBool(c.KeepInput), i)
+			dropInputFlag.Add(strconv.FormatBool(c.DropInput), i)
+			ignoreFirstIntervalsFlag.Add(strconv.Itoa(c.IgnoreFirstIntervals), i)
+			ignoreOldSamplesFlag.Add(strconv.FormatBool(c.IgnoreOldSamples), i)
+			enableWindowsFlag.Add(strconv.FormatBool(c.EnableWindows), i)
+		}
+		// deduplication can work without stream aggregation rules
+		dedupIntervalFlag.Add(c.DedupInterval, i)
+		dropInputLabelsFlag.Add(strings.Join(c.DropInputLabels, ","), i)
+	}
+	args = AppendFlagsToArgs(args, len(cs), configFlag, keepInputFlag, dropInputFlag, ignoreFirstIntervalsFlag, ignoreOldSamplesFlag, enableWindowsFlag)
+	args = AppendFlagsToArgs(args, len(cs), dedupIntervalFlag, dropInputLabelsFlag)
+	return args
+}
+
+// StreamAggrVolumeTo conditionally mounts configmap with stream aggregation config into given volumes and volume mounts
+func StreamAggrVolumeTo(volumes []corev1.Volume, mounts []corev1.VolumeMount, name string, cs ...*vmv1beta1.StreamAggrConfig) ([]corev1.Volume, []corev1.VolumeMount) {
+	if len(cs) == 0 {
+		return volumes, mounts
+	}
+	hasRule := false
+	for _, c := range cs {
+		if c.HasAnyRule() {
+			hasRule = true
+			break
+		}
+	}
+	if hasRule {
+		volumes = append(volumes, corev1.Volume{
+			Name: "stream-aggr-conf",
+			VolumeSource: corev1.VolumeSource{
+				ConfigMap: &corev1.ConfigMapVolumeSource{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: name,
+					},
+				},
+			},
+		})
+		mounts = append(mounts, corev1.VolumeMount{
+			Name:      "stream-aggr-conf",
+			ReadOnly:  true,
+			MountPath: vmv1beta1.StreamAggrConfigDir,
+		})
+	}
+	return volumes, mounts
+}
diff --git a/internal/controller/operator/factory/build/cache.go b/internal/controller/operator/factory/build/cache.go
index 24120691..8da1cb9f 100644
--- a/internal/controller/operator/factory/build/cache.go
+++ b/internal/controller/operator/factory/build/cache.go
@@ -42,13 +42,16 @@ const (
 	TLSAssetsResourceKind ResourceKind = "tls-assets"
 	// SecretConfigResourceKind defines build for configuration secret object
 	SecretConfigResourceKind ResourceKind = "config"
+	// StreamAggrConfigResourceKind defines build for stream aggregation configmap object
+	StreamAggrConfigResourceKind ResourceKind = "stream-aggr"
 )
 
 // ResourceName returns a name for provided resource and corresponding cr object
 func ResourceName(kind ResourceKind, cr builderOpts) string {
 	var parts []string
-	if kind == TLSAssetsResourceKind {
-		parts = append(parts, "tls-assets")
+	switch kind {
+	case TLSAssetsResourceKind, StreamAggrConfigResourceKind:
+		parts = append(parts, string(kind))
 	}
 	parts = append(parts, cr.PrefixedName())
 	return strings.Join(parts, "-")
diff --git a/internal/controller/operator/factory/finalize/vmagent.go b/internal/controller/operator/factory/finalize/vmagent.go
index a683d868..f9b4c5d6 100644
--- a/internal/controller/operator/factory/finalize/vmagent.go
+++ b/internal/controller/operator/factory/finalize/vmagent.go
@@ -56,7 +56,7 @@ func OnVMAgentDelete(ctx context.Context, rclient client.Client, cr *vmv1beta1.V
 	if err := removeFinalizeObjByName(ctx, rclient, &corev1.ConfigMap{}, cr.RelabelingAssetName(), cr.Namespace); err != nil {
 		return err
 	}
-	if err := removeFinalizeObjByName(ctx, rclient, &corev1.ConfigMap{}, cr.StreamAggrConfigName(), cr.Namespace); err != nil {
+	if err := removeFinalizeObjByName(ctx, rclient, &corev1.ConfigMap{}, build.ResourceName(build.StreamAggrConfigResourceKind, cr), cr.Namespace); err != nil {
 		return err
 	}
 
diff --git a/internal/controller/operator/factory/finalize/vmsingle.go b/internal/controller/operator/factory/finalize/vmsingle.go
index f9307185..d7f9cb0b 100644
--- a/internal/controller/operator/factory/finalize/vmsingle.go
+++ b/internal/controller/operator/factory/finalize/vmsingle.go
@@ -8,6 +8,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	vmv1beta1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
+	"github.com/VictoriaMetrics/operator/internal/controller/operator/factory/build"
 )
 
 // OnVMSingleDelete deletes all vmsingle related resources
@@ -30,7 +31,7 @@ func OnVMSingleDelete(ctx context.Context, rclient client.Client, cr *vmv1beta1.
 			return err
 		}
 	}
-	if err := removeFinalizeObjByName(ctx, rclient, &corev1.ConfigMap{}, cr.StreamAggrConfigName(), cr.Namespace); err != nil {
+	if err := removeFinalizeObjByName(ctx, rclient, &corev1.ConfigMap{}, build.ResourceName(build.StreamAggrConfigResourceKind, cr), cr.Namespace); err != nil {
 		return err
 	}
 	if err := deleteSA(ctx, rclient, cr); err != nil {
diff --git a/internal/controller/operator/factory/reconcile/statefulset.go b/internal/controller/operator/factory/reconcile/statefulset.go
index cf8eff66..9eebf7be 100644
--- a/internal/controller/operator/factory/reconcile/statefulset.go
+++ b/internal/controller/operator/factory/reconcile/statefulset.go
@@ -6,14 +6,18 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -32,6 +36,7 @@ type STSOptions struct {
 	SelectorLabels     func() map[string]string
 	HPA                *vmv1beta1.EmbeddedHPA
 	UpdateReplicaCount func(count *int32)
+	UpdateBehavior     *vmv1beta1.StatefulSetUpdateStrategyBehavior
 }
 
 func waitForStatefulSetReady(ctx context.Context, rclient client.Client, newSts *appsv1.StatefulSet) error {
@@ -145,9 +150,22 @@ func HandleSTSUpdate(ctx context.Context, rclient client.Client, cr STSOptions,
 			}
 		}
 
+		// determine manual update behavior only with OnDelete policy, one by one by default
+		podMaxUnavailable := 1
+		if cr.UpdateBehavior != nil {
+			if newSts.Spec.UpdateStrategy.Type == appsv1.OnDeleteStatefulSetStrategyType {
+				podMaxUnavailable, err = intstr.GetScaledValueFromIntOrPercent(cr.UpdateBehavior.MaxUnavailable, int(*newSts.Spec.Replicas), false)
+				if err != nil {
+					return err
+				}
+			} else {
+				logger.WithContext(ctx).Info(fmt.Sprintf("ignoring custom update behavior settings with update strategy=%s on sts: %s", newSts.Spec.UpdateStrategy.Type, newSts.Name))
+			}
+		}
+
 		// perform manual update only with OnDelete policy, which is default.
 		if newSts.Spec.UpdateStrategy.Type == appsv1.OnDeleteStatefulSetStrategyType {
-			if err := performRollingUpdateOnSts(ctx, podMustRecreate, rclient, newSts.Name, newSts.Namespace, cr.SelectorLabels()); err != nil {
+			if err := performRollingUpdateOnSts(ctx, podMustRecreate, rclient, newSts.Name, newSts.Namespace, cr.SelectorLabels(), podMaxUnavailable); err != nil {
 				return fmt.Errorf("cannot handle rolling-update on sts: %s, err: %w", newSts.Name, err)
 			}
 		} else {
@@ -187,14 +205,14 @@ func getLatestStsState(ctx context.Context, rclient client.Client, targetSTS typ
 	return &sts, nil
 }
 
-// we perform rolling update on sts by manually deleting pods one by one
+// we perform rolling update on sts by manually evicting pods one by one or in batches
 // we check sts revision (kubernetes controller-manager is responsible for that)
 // and compare pods revision label with sts revision
 // if it doesn't match - updated is needed
 //
 // we always check if sts.Status.CurrentRevision needs update, to keep it equal to UpdateRevision
 // see https://github.com/kubernetes/kube-state-metrics/issues/1324#issuecomment-1779751992
-func performRollingUpdateOnSts(ctx context.Context, podMustRecreate bool, rclient client.Client, stsName string, ns string, podLabels map[string]string) error {
+func performRollingUpdateOnSts(ctx context.Context, podMustRecreate bool, rclient client.Client, stsName string, ns string, podLabels map[string]string, podMaxUnavailable int) error {
 	time.Sleep(podWaitReadyIntervalCheck)
 	sts, err := getLatestStsState(ctx, rclient, types.NamespacedName{Name: stsName, Namespace: ns})
 	if err != nil {
@@ -278,6 +296,7 @@ func performRollingUpdateOnSts(ctx context.Context, podMustRecreate bool, rclien
 	}
 
 	l.Info(fmt.Sprintf("discovered already updated pods=%d, pods needed to be update=%d", len(updatedPods), len(podsForUpdate)))
+
 	// check updated, by not ready pods
 	for _, pod := range updatedPods {
 		l.Info(fmt.Sprintf("checking ready status for already updated pod %s to revision version=%q", pod.Name, stsVersion))
@@ -288,19 +307,46 @@ func performRollingUpdateOnSts(ctx context.Context, podMustRecreate bool, rclien
 		}
 	}
 
-	// perform update for not updated pods
-	for _, pod := range podsForUpdate {
-		l.Info(fmt.Sprintf("updating pod=%s revision label=%q", pod.Name, pod.Labels[podRevisionLabel]))
-		// we have to delete pod and wait for it readiness
-		err := rclient.Delete(ctx, &pod)
-		if err != nil {
-			return err
+	// perform update for not updated pods in batches according to podMaxUnavailable
+	for batchStart := 0; batchStart < len(podsForUpdate); batchStart += podMaxUnavailable {
+		var batch []corev1.Pod
+		var batchFailed atomic.Bool
+
+		// determine batch of pods to update
+		batchClose := batchStart + podMaxUnavailable
+		if batchClose > len(podsForUpdate) {
+			batchClose = len(podsForUpdate)
+		}
+		batch = podsForUpdate[batchStart:batchClose]
+
+		var wg sync.WaitGroup
+		for _, pod := range batch {
+			wg.Add(1)
+			go func(pod corev1.Pod) {
+				defer wg.Done()
+				l.Info(fmt.Sprintf("updating pod=%s revision label=%q", pod.Name, pod.Labels[podRevisionLabel]))
+				// evict pod to trigger re-creation
+				podEviction := policyv1.Eviction{ObjectMeta: pod.ObjectMeta}
+				if err := rclient.SubResource("eviction").Create(ctx, &pod, &podEviction); err != nil {
+					l.Error(err, fmt.Sprintf("cannot evict pod %s", pod.Name))
+					batchFailed.Store(true)
+					return
+				}
+				// wait for pod to be re-created
+				podNsn := types.NamespacedName{Namespace: ns, Name: pod.Name}
+				if err := waitForPodReady(ctx, rclient, podNsn, stsVersion, sts.Spec.MinReadySeconds); err != nil {
+					l.Error(err, fmt.Sprintf("cannot wait for pod ready state during re-creation for pod %s", pod.Name))
+					batchFailed.Store(true)
+					return
+				}
+				l.Info(fmt.Sprintf("pod %s was updated successfully", pod.Name))
+			}(pod)
 		}
-		podNsn := types.NamespacedName{Namespace: ns, Name: pod.Name}
-		if err = waitForPodReady(ctx, rclient, podNsn, stsVersion, sts.Spec.MinReadySeconds); err != nil {
-			return fmt.Errorf("cannot wait for pod ready state during re-creation: %w", err)
+		wg.Wait()
+
+		if batchFailed.Load() {
+			return fmt.Errorf("one or more pods in the batch failed, see logs for details")
 		}
-		l.Info(fmt.Sprintf("pod %s was updated successfully", pod.Name))
 	}
 
 	l.Info(fmt.Sprintf("finished statefulset update from revision=%q to revision=%q", sts.Status.CurrentRevision, stsVersion))
diff --git a/internal/controller/operator/factory/vlcluster/vlstorage.go b/internal/controller/operator/factory/vlcluster/vlstorage.go
index 44067a68..5b3b81cc 100644
--- a/internal/controller/operator/factory/vlcluster/vlstorage.go
+++ b/internal/controller/operator/factory/vlcluster/vlstorage.go
@@ -155,6 +155,9 @@ func buildVLStorageSTSSpec(cr *vmv1.VLCluster) (*appsv1.StatefulSet, error) {
 			ServiceName: cr.GetVLStorageName(),
 		},
 	}
+	if cr.Spec.VLStorage.PersistentVolumeClaimRetentionPolicy != nil {
+		stsSpec.Spec.PersistentVolumeClaimRetentionPolicy = cr.Spec.VLStorage.PersistentVolumeClaimRetentionPolicy
+	}
 	build.StatefulSetAddCommonParams(stsSpec, ptr.Deref(cr.Spec.VLStorage.UseStrictSecurity, false), &cr.Spec.VLStorage.CommonApplicationDeploymentParams)
 	storageSpec := cr.Spec.VLStorage.Storage
 	storageSpec.IntoSTSVolume(cr.Spec.VLStorage.GetStorageVolumeName(), &stsSpec.Spec)
diff --git a/internal/controller/operator/factory/vmagent/vmagent.go b/internal/controller/operator/factory/vmagent/vmagent.go
index 9a66f470..0f2da4d2 100644
--- a/internal/controller/operator/factory/vmagent/vmagent.go
+++ b/internal/controller/operator/factory/vmagent/vmagent.go
@@ -428,6 +428,9 @@ func newDeploy(cr *vmv1beta1.VMAgent, ac *build.AssetsCache) (runtime.Object, er
 				},
 			},
 		}
+		if cr.Spec.PersistentVolumeClaimRetentionPolicy != nil {
+			stsSpec.Spec.PersistentVolumeClaimRetentionPolicy = cr.Spec.PersistentVolumeClaimRetentionPolicy
+		}
 		build.StatefulSetAddCommonParams(stsSpec, useStrictSecurity, &cr.Spec.CommonApplicationDeploymentParams)
 		stsSpec.Spec.Template.Spec.Volumes = build.AddServiceAccountTokenVolume(stsSpec.Spec.Template.Spec.Volumes, &cr.Spec.CommonApplicationDeploymentParams)
 		cr.Spec.StatefulStorage.IntoSTSVolume(vmAgentPersistentQueueMountName, &stsSpec.Spec)
@@ -606,28 +609,7 @@ func makeSpec(cr *vmv1beta1.VMAgent, ac *build.AssetsCache) (*corev1.PodSpec, er
 		})
 
 	}
-	if cr.HasAnyStreamAggrRule() {
-		volumes = append(volumes,
-			corev1.Volume{
-				Name: "stream-aggr-conf",
-				VolumeSource: corev1.VolumeSource{
-					ConfigMap: &corev1.ConfigMapVolumeSource{
-						LocalObjectReference: corev1.LocalObjectReference{
-							Name: cr.StreamAggrConfigName(),
-						},
-					},
-				},
-			},
-		)
-		agentVolumeMounts = append(agentVolumeMounts,
-			corev1.VolumeMount{
-				Name:      "stream-aggr-conf",
-				ReadOnly:  true,
-				MountPath: vmv1beta1.StreamAggrConfigDir,
-			},
-		)
-	}
-
+	volumes, agentVolumeMounts = build.StreamAggrVolumeTo(volumes, agentVolumeMounts, build.ResourceName(build.StreamAggrConfigResourceKind, cr), cr.Spec.StreamAggrConfig)
 	if cr.HasAnyRelabellingConfigs() {
 		volumes = append(volumes,
 			corev1.Volume{
@@ -694,32 +676,10 @@ func makeSpec(cr *vmv1beta1.VMAgent, ac *build.AssetsCache) (*corev1.PodSpec, er
 		args = append(args, "-remoteWrite.relabelConfig="+path.Join(vmv1beta1.RelabelingConfigDir, globalRelabelingName))
 	}
 
-	if cr.Spec.StreamAggrConfig != nil {
-		if cr.Spec.StreamAggrConfig.HasAnyRule() {
-			args = append(args, "-streamAggr.config="+path.Join(vmv1beta1.StreamAggrConfigDir, globalAggregationConfigName))
-		}
-		if cr.Spec.StreamAggrConfig.KeepInput {
-			args = append(args, "-streamAggr.keepInput=true")
-		}
-		if cr.Spec.StreamAggrConfig.DropInput {
-			args = append(args, "-streamAggr.dropInput=true")
-		}
-		if cr.Spec.StreamAggrConfig.DedupInterval != "" {
-			args = append(args, fmt.Sprintf("-streamAggr.dedupInterval=%s", cr.Spec.StreamAggrConfig.DedupInterval))
-		}
-		if len(cr.Spec.StreamAggrConfig.DropInputLabels) > 0 {
-			args = append(args, fmt.Sprintf("-streamAggr.dropInputLabels=%s", strings.Join(cr.Spec.StreamAggrConfig.DropInputLabels, ",")))
-		}
-		if cr.Spec.StreamAggrConfig.IgnoreOldSamples {
-			args = append(args, "-streamAggr.ignoreOldSamples=true")
-		}
-		if cr.Spec.StreamAggrConfig.EnableWindows {
-			args = append(args, "-streamAggr.enableWindows=true")
-		}
-	}
-
+	streamAggrKeys := []string{globalAggregationConfigName}
+	streamAggrConfigs := []*vmv1beta1.StreamAggrConfig{cr.Spec.StreamAggrConfig}
+	args = build.StreamAggrArgsTo(args, "streamAggr", streamAggrKeys, streamAggrConfigs...)
 	args = build.AppendArgsForInsertPorts(args, cr.Spec.InsertPorts)
-
 	args = build.AddExtraArgsOverrideDefaults(args, cr.Spec.ExtraArgs, "-")
 	sort.Strings(args)
 
@@ -907,20 +867,10 @@ func createOrUpdateRelabelConfigsAssets(ctx context.Context, rclient client.Clie
 	return reconcile.ConfigMap(ctx, rclient, assestsCM, prevConfigMeta)
 }
 
-func buildStreamAggrConfigMeta(cr *vmv1beta1.VMAgent) metav1.ObjectMeta {
-	return metav1.ObjectMeta{
-		Namespace:       cr.Namespace,
-		Name:            cr.StreamAggrConfigName(),
-		Labels:          cr.AllLabels(),
-		Annotations:     cr.AnnotationsFiltered(),
-		OwnerReferences: cr.AsOwner(),
-	}
-}
-
 // buildStreamAggrConfig combines all possible stream aggregation configs and adding it to the configmap.
 func buildStreamAggrConfig(ctx context.Context, cr *vmv1beta1.VMAgent, rclient client.Client) (*corev1.ConfigMap, error) {
 	cfgCM := &corev1.ConfigMap{
-		ObjectMeta: buildStreamAggrConfigMeta(cr),
+		ObjectMeta: build.ResourceMeta(build.StreamAggrConfigResourceKind, cr),
 		Data:       make(map[string]string),
 	}
 	// global section
@@ -950,8 +900,9 @@ func buildStreamAggrConfig(ctx context.Context, cr *vmv1beta1.VMAgent, rclient c
 	for i := range cr.Spec.RemoteWrite {
 		rw := cr.Spec.RemoteWrite[i]
 		if rw.StreamAggrConfig != nil {
-			if len(rw.StreamAggrConfig.Rules) > 0 {
-				data, err := yaml.Marshal(rw.StreamAggrConfig.Rules)
+			c := rw.StreamAggrConfig
+			if len(c.Rules) > 0 {
+				data, err := yaml.Marshal(c.Rules)
 				if err != nil {
 					return nil, fmt.Errorf("cannot serialize relabelConfig as yaml: %w", err)
 				}
@@ -959,12 +910,12 @@ func buildStreamAggrConfig(ctx context.Context, cr *vmv1beta1.VMAgent, rclient c
 					cfgCM.Data[rw.AsConfigMapKey(i, "stream-aggr-conf")] = string(data)
 				}
 			}
-			if rw.StreamAggrConfig.RuleConfigMap != nil {
+			if c.RuleConfigMap != nil {
 				data, err := k8stools.FetchConfigMapContentByKey(ctx, rclient,
-					&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: rw.StreamAggrConfig.RuleConfigMap.Name, Namespace: cr.Namespace}},
-					rw.StreamAggrConfig.RuleConfigMap.Key)
+					&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: c.RuleConfigMap.Name, Namespace: cr.Namespace}},
+					c.RuleConfigMap.Key)
 				if err != nil {
-					return nil, fmt.Errorf("cannot fetch configmap: %s, err: %w", rw.StreamAggrConfig.RuleConfigMap.Name, err)
+					return nil, fmt.Errorf("cannot fetch configmap: %s, err: %w", c.RuleConfigMap.Name, err)
 				}
 				if len(data) > 0 {
 					cfgCM.Data[rw.AsConfigMapKey(i, "stream-aggr-conf")] += data
@@ -989,16 +940,11 @@ func createOrUpdateStreamAggrConfig(ctx context.Context, rclient client.Client,
 	}
 	var prevConfigMeta *metav1.ObjectMeta
 	if prevCR != nil {
-		prevConfigMeta = ptr.To(buildStreamAggrConfigMeta(prevCR))
+		prevConfigMeta = ptr.To(build.ResourceMeta(build.StreamAggrConfigResourceKind, cr))
 	}
 	return reconcile.ConfigMap(ctx, rclient, streamAggrCM, prevConfigMeta)
 }
 
-type remoteFlag struct {
-	isNotNull   bool
-	flagSetting string
-}
-
 func buildRemoteWriteSettings(cr *vmv1beta1.VMAgent) []string {
 	// limit to 1GB
 	// most people do not care about this setting,
@@ -1091,305 +1037,155 @@ func sortMap(m map[string]string) []item {
 }
 
 func buildRemoteWrites(cr *vmv1beta1.VMAgent, ac *build.AssetsCache) ([]string, error) {
-	var finalArgs []string
-	var remoteArgs []remoteFlag
 	remoteTargets := cr.Spec.RemoteWrite
 
-	url := remoteFlag{flagSetting: "-remoteWrite.url=", isNotNull: true}
-
-	authUser := remoteFlag{flagSetting: "-remoteWrite.basicAuth.username="}
-	authPasswordFile := remoteFlag{flagSetting: "-remoteWrite.basicAuth.passwordFile="}
-	bearerTokenFile := remoteFlag{flagSetting: "-remoteWrite.bearerTokenFile="}
-	urlRelabelConfig := remoteFlag{flagSetting: "-remoteWrite.urlRelabelConfig="}
-	sendTimeout := remoteFlag{flagSetting: "-remoteWrite.sendTimeout="}
-	tlsCAs := remoteFlag{flagSetting: "-remoteWrite.tlsCAFile="}
-	tlsCerts := remoteFlag{flagSetting: "-remoteWrite.tlsCertFile="}
-	tlsKeys := remoteFlag{flagSetting: "-remoteWrite.tlsKeyFile="}
-	tlsInsecure := remoteFlag{flagSetting: "-remoteWrite.tlsInsecureSkipVerify="}
-	tlsServerName := remoteFlag{flagSetting: "-remoteWrite.tlsServerName="}
-	oauth2ClientID := remoteFlag{flagSetting: "-remoteWrite.oauth2.clientID="}
-	oauth2ClientSecretFile := remoteFlag{flagSetting: "-remoteWrite.oauth2.clientSecretFile="}
-	oauth2Scopes := remoteFlag{flagSetting: "-remoteWrite.oauth2.scopes="}
-	oauth2TokenURL := remoteFlag{flagSetting: "-remoteWrite.oauth2.tokenUrl="}
-	headers := remoteFlag{flagSetting: "-remoteWrite.headers="}
-	streamAggrConfig := remoteFlag{flagSetting: "-remoteWrite.streamAggr.config="}
-	streamAggrKeepInput := remoteFlag{flagSetting: "-remoteWrite.streamAggr.keepInput="}
-	streamAggrDropInput := remoteFlag{flagSetting: "-remoteWrite.streamAggr.dropInput="}
-	streamAggrDedupInterval := remoteFlag{flagSetting: "-remoteWrite.streamAggr.dedupInterval="}
-	streamAggrDropInputLabels := remoteFlag{flagSetting: "-remoteWrite.streamAggr.dropInputLabels="}
-	streamAggrIgnoreFirstIntervals := remoteFlag{flagSetting: "-remoteWrite.streamAggr.ignoreFirstIntervals="}
-	streamAggrIgnoreOldSamples := remoteFlag{flagSetting: "-remoteWrite.streamAggr.ignoreOldSamples="}
-	streamAggrEnableWindows := remoteFlag{flagSetting: "-remoteWrite.streamAggr.enableWindows="}
-	maxDiskUsagePerURL := remoteFlag{flagSetting: "-remoteWrite.maxDiskUsagePerURL="}
-	forceVMProto := remoteFlag{flagSetting: "-remoteWrite.forceVMProto="}
-	proxyURL := remoteFlag{flagSetting: "-remoteWrite.proxyURL="}
-	awsEC2Endpoint := remoteFlag{flagSetting: "-remoteWrite.aws.ec2Endpoint="}
-	awsRegion := remoteFlag{flagSetting: "-remoteWrite.aws.region="}
-	awsRoleARN := remoteFlag{flagSetting: "-remoteWrite.aws.roleARN="}
-	awsService := remoteFlag{flagSetting: "-remoteWrite.aws.service="}
-	awsSTSEndpoint := remoteFlag{flagSetting: "-remoteWrite.aws.stsEndpoint="}
-	awsUseSigv4 := remoteFlag{flagSetting: "-remoteWrite.aws.useSigv4="}
-
 	maxDiskUsage := defaultMaxDiskUsage
 	if cr.Spec.RemoteWriteSettings != nil && cr.Spec.RemoteWriteSettings.MaxDiskUsagePerURL != nil {
 		maxDiskUsage = cr.Spec.RemoteWriteSettings.MaxDiskUsagePerURL.String()
 	}
+	url := build.NewFlag("-remoteWrite.url", "")
+	authUser := build.NewFlag("-remoteWrite.basicAuth.username", `""`)
+	authPasswordFile := build.NewFlag("-remoteWrite.basicAuth.passwordFile", "")
+	bearerTokenFile := build.NewFlag("-remoteWrite.bearerTokenFile", `""`)
+	urlRelabelConfig := build.NewFlag("-remoteWrite.urlRelabelConfig", "")
+	sendTimeout := build.NewFlag("-remoteWrite.sendTimeout", "")
+	tlsCAs := build.NewFlag("-remoteWrite.tlsCAFile", "")
+	tlsCerts := build.NewFlag("-remoteWrite.tlsCertFile", "")
+	tlsKeys := build.NewFlag("-remoteWrite.tlsKeyFile", "")
+	tlsInsecure := build.NewFlag("-remoteWrite.tlsInsecureSkipVerify", "false")
+	tlsServerName := build.NewFlag("-remoteWrite.tlsServerName", "")
+	oauth2ClientID := build.NewFlag("-remoteWrite.oauth2.clientID", "")
+	oauth2ClientSecretFile := build.NewFlag("-remoteWrite.oauth2.clientSecretFile", "")
+	oauth2Scopes := build.NewFlag("-remoteWrite.oauth2.scopes", "")
+	oauth2TokenURL := build.NewFlag("-remoteWrite.oauth2.tokenUrl", "")
+	headers := build.NewFlag("-remoteWrite.headers", "")
+	maxDiskUsagePerURL := build.NewFlag("-remoteWrite.maxDiskUsagePerURL", maxDiskUsage)
+	forceVMProto := build.NewFlag("-remoteWrite.forceVMProto", "false")
+	proxyURL := build.NewFlag("-remoteWrite.proxyURL", "")
+	awsEC2Endpoint := build.NewFlag("-remoteWrite.aws.ec2Endpoint", "")
+	awsRegion := build.NewFlag("-remoteWrite.aws.region", "")
+	awsRoleARN := build.NewFlag("-remoteWrite.aws.roleARN", "")
+	awsService := build.NewFlag("-remoteWrite.aws.service", "")
+	awsSTSEndpoint := build.NewFlag("-remoteWrite.aws.stsEndpoint", "")
+	awsUseSigv4 := build.NewFlag("-remoteWrite.aws.useSigv4", "false")
+
+	var streamAggrConfigs []*vmv1beta1.StreamAggrConfig
+	var streamAggrKeys []string
 
 	for i := range remoteTargets {
 		rws := remoteTargets[i]
-		url.flagSetting += fmt.Sprintf("%s,", rws.URL)
-
-		var caPath, certPath, keyPath, serverName string
-		var insecure bool
+		url.Add(rws.URL, i)
 		if rws.TLSConfig != nil {
 			creds, err := ac.BuildTLSCreds(cr.Namespace, rws.TLSConfig)
 			if err != nil {
 				return nil, err
 			}
-			if creds.CAFile != "" {
-				tlsCAs.isNotNull = true
-				caPath = creds.CAFile
-			}
-			if creds.CertFile != "" {
-				tlsCerts.isNotNull = true
-				certPath = creds.CertFile
-			}
-			if creds.KeyFile != "" {
-				tlsKeys.isNotNull = true
-				keyPath = creds.KeyFile
-			}
+			tlsCAs.Add(creds.CAFile, i)
+			tlsCerts.Add(creds.CertFile, i)
+			tlsKeys.Add(creds.KeyFile, i)
 			if rws.TLSConfig.InsecureSkipVerify {
-				tlsInsecure.isNotNull = true
-			}
-			if rws.TLSConfig.ServerName != "" {
-				serverName = rws.TLSConfig.ServerName
-				tlsServerName.isNotNull = true
+				tlsInsecure.Add("true", i)
 			}
-			insecure = rws.TLSConfig.InsecureSkipVerify
+			tlsServerName.Add(rws.TLSConfig.ServerName, i)
 		}
-		tlsCAs.flagSetting += fmt.Sprintf("%s,", caPath)
-		tlsCerts.flagSetting += fmt.Sprintf("%s,", certPath)
-		tlsKeys.flagSetting += fmt.Sprintf("%s,", keyPath)
-		tlsServerName.flagSetting += fmt.Sprintf("%s,", serverName)
-		tlsInsecure.flagSetting += fmt.Sprintf("%v,", insecure)
-
-		var user string
-		var passFile string
 		if rws.BasicAuth != nil {
-			authUser.isNotNull = true
-			var err error
-			user, err = ac.LoadKeyFromSecret(cr.Namespace, &rws.BasicAuth.Username)
+			user, err := ac.LoadKeyFromSecret(cr.Namespace, &rws.BasicAuth.Username)
 			if err != nil {
 				return nil, fmt.Errorf("cannot load BasicAuth username: %w", err)
 			}
+			authUser.Add(strconv.Quote(user), i)
 			if len(rws.BasicAuth.Password.Name) > 0 {
-				authPasswordFile.isNotNull = true
-				var err error
-				passFile, err = ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, &rws.BasicAuth.Password)
+				passFile, err := ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, &rws.BasicAuth.Password)
 				if err != nil {
 					return nil, fmt.Errorf("cannot load BasicAuth password: %w", err)
 				}
+				authPasswordFile.Add(passFile, i)
 			}
 			if len(rws.BasicAuth.PasswordFile) > 0 {
-				passFile = rws.BasicAuth.PasswordFile
-				authPasswordFile.isNotNull = true
+				authPasswordFile.Add(rws.BasicAuth.PasswordFile, i)
 			}
 		}
-		authUser.flagSetting += fmt.Sprintf("\"%s\",", strings.ReplaceAll(user, `"`, `\"`))
-		authPasswordFile.flagSetting += fmt.Sprintf("%s,", passFile)
-
-		var value string
 		if rws.BearerTokenSecret != nil && rws.BearerTokenSecret.Name != "" {
-			bearerTokenFile.isNotNull = true
-			var err error
-			value, err = ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, rws.BearerTokenSecret)
+			value, err := ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, rws.BearerTokenSecret)
 			if err != nil {
 				return nil, fmt.Errorf("cannot load BearerTokenSecret: %w", err)
 			}
+			bearerTokenFile.Add(strconv.Quote(value), i)
 		}
-		bearerTokenFile.flagSetting += fmt.Sprintf("\"%s\",", strings.ReplaceAll(value, `"`, `\"`))
-
-		value = ""
-
 		if rws.UrlRelabelConfig != nil || len(rws.InlineUrlRelabelConfig) > 0 {
-			urlRelabelConfig.isNotNull = true
-			value = path.Join(vmv1beta1.RelabelingConfigDir, fmt.Sprintf(urlRelabelingName, i))
+			urlRelabelConfig.Add(path.Join(vmv1beta1.RelabelingConfigDir, fmt.Sprintf(urlRelabelingName, i)), i)
 		}
-
-		urlRelabelConfig.flagSetting += fmt.Sprintf("%s,", value)
-
-		value = ""
 		if rws.SendTimeout != nil {
-			if !sendTimeout.isNotNull {
-				sendTimeout.isNotNull = true
-			}
-			value = *rws.SendTimeout
+			sendTimeout.Add(*rws.SendTimeout, i)
 		}
-		sendTimeout.flagSetting += fmt.Sprintf("%s,", value)
-
-		value = ""
 		if len(rws.Headers) > 0 {
-			headers.isNotNull = true
+			value := ""
 			for _, headerValue := range rws.Headers {
 				value += headerValue + "^^"
 			}
 			value = strings.TrimSuffix(value, "^^")
+			headers.Add(value, i)
 		}
-		headers.flagSetting += fmt.Sprintf("%s,", value)
-		var oaturl, oascopes, oaclientID, oaSecretKeyFile string
 		if rws.OAuth2 != nil {
 			if len(rws.OAuth2.TokenURL) > 0 {
-				oauth2TokenURL.isNotNull = true
-				oaturl = rws.OAuth2.TokenURL
+				oauth2TokenURL.Add(rws.OAuth2.TokenURL, i)
 			}
 
 			if len(rws.OAuth2.Scopes) > 0 {
-				oauth2Scopes.isNotNull = true
-				oascopes = strings.Join(rws.OAuth2.Scopes, ",")
+				oauth2Scopes.Add(strings.Join(rws.OAuth2.Scopes, ","), i)
 			}
 
 			if len(rws.OAuth2.ClientSecretFile) > 0 {
-				oauth2ClientSecretFile.isNotNull = true
-				oaSecretKeyFile = rws.OAuth2.ClientSecretFile
+				oauth2ClientSecretFile.Add(rws.OAuth2.ClientSecretFile, i)
 			}
 
 			if rws.OAuth2.ClientSecret != nil {
-				oauth2ClientSecretFile.isNotNull = true
-				var err error
-				oaSecretKeyFile, err = ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, rws.OAuth2.ClientSecret)
+				oaSecretKeyFile, err := ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, rws.OAuth2.ClientSecret)
 				if err != nil {
 					return nil, fmt.Errorf("cannot load OAuth2 ClientSecret: %w", err)
 				}
+				oauth2ClientSecretFile.Add(oaSecretKeyFile, i)
 			}
-			if secret, err := ac.LoadKeyFromSecretOrConfigMap(cr.Namespace, &rws.OAuth2.ClientID); err != nil {
+			secret, err := ac.LoadKeyFromSecretOrConfigMap(cr.Namespace, &rws.OAuth2.ClientID)
+			if err != nil {
 				return nil, err
-			} else if len(secret) > 0 {
-				oaclientID = secret
-				oauth2ClientID.isNotNull = true
 			}
+			oauth2ClientID.Add(secret, i)
 
 		}
-		oauth2TokenURL.flagSetting += fmt.Sprintf("%s,", oaturl)
-		oauth2ClientSecretFile.flagSetting += fmt.Sprintf("%s,", oaSecretKeyFile)
-		oauth2ClientID.flagSetting += fmt.Sprintf("%s,", oaclientID)
-		oauth2Scopes.flagSetting += fmt.Sprintf("%s,", oascopes)
 
-		var ec2Endpoint, region, service, roleARN, stsEndpoint string
-		var useSigv4 bool
 		if rws.AWS != nil {
-			if len(rws.AWS.EC2Endpoint) > 0 {
-				ec2Endpoint = rws.AWS.EC2Endpoint
-				awsEC2Endpoint.isNotNull = true
-			}
-			if len(rws.AWS.Region) > 0 {
-				region = rws.AWS.Region
-				awsRegion.isNotNull = true
-			}
-			if len(rws.AWS.RoleARN) > 0 {
-				roleARN = rws.AWS.RoleARN
-				awsRoleARN.isNotNull = true
-			}
-			if len(rws.AWS.Service) > 0 {
-				service = rws.AWS.Service
-				awsService.isNotNull = true
-			}
-			if len(rws.AWS.STSEndpoint) > 0 {
-				stsEndpoint = rws.AWS.STSEndpoint
-				awsSTSEndpoint.isNotNull = true
-			}
-			if rws.AWS.UseSigv4 {
-				useSigv4 = rws.AWS.UseSigv4
-				awsUseSigv4.isNotNull = true
-			}
+			awsEC2Endpoint.Add(rws.AWS.EC2Endpoint, i)
+			awsRegion.Add(rws.AWS.Region, i)
+			awsRoleARN.Add(rws.AWS.RoleARN, i)
+			awsService.Add(rws.AWS.Service, i)
+			awsSTSEndpoint.Add(rws.AWS.STSEndpoint, i)
+			awsUseSigv4.Add(strconv.FormatBool(rws.AWS.UseSigv4), i)
 		}
-		awsEC2Endpoint.flagSetting += fmt.Sprintf("%s,", ec2Endpoint)
-		awsRegion.flagSetting += fmt.Sprintf("%s,", region)
-		awsRoleARN.flagSetting += fmt.Sprintf("%s,", roleARN)
-		awsService.flagSetting += fmt.Sprintf("%s,", service)
-		awsSTSEndpoint.flagSetting += fmt.Sprintf("%s,", stsEndpoint)
-		awsUseSigv4.flagSetting += fmt.Sprintf("%v,", useSigv4)
-
-		var dedupIntVal, streamConfVal string
-		var keepInputVal, dropInputVal, ignoreOldSamples, enableWindows bool
-		var ignoreFirstIntervalsVal int
-		if rws.StreamAggrConfig != nil {
-			if rws.StreamAggrConfig.HasAnyRule() {
-				streamAggrConfig.isNotNull = true
-				streamConfVal = path.Join(vmv1beta1.StreamAggrConfigDir, rws.AsConfigMapKey(i, "stream-aggr-conf"))
-			}
 
-			dedupIntVal = rws.StreamAggrConfig.DedupInterval
-			if dedupIntVal != "" {
-				streamAggrDedupInterval.isNotNull = true
-			}
+		streamAggrConfigs = append(streamAggrConfigs, rws.StreamAggrConfig)
+		streamAggrKeys = append(streamAggrKeys, rws.AsConfigMapKey(i, "stream-aggr-conf"))
 
-			keepInputVal = rws.StreamAggrConfig.KeepInput
-			if keepInputVal {
-				streamAggrKeepInput.isNotNull = true
-			}
-			dropInputVal = rws.StreamAggrConfig.DropInput
-			if dropInputVal {
-				streamAggrDropInput.isNotNull = true
-			}
-			if len(rws.StreamAggrConfig.DropInputLabels) > 0 {
-				streamAggrDropInputLabels.isNotNull = true
-				streamAggrDropInputLabels.flagSetting += fmt.Sprintf("%s,", strings.Join(rws.StreamAggrConfig.DropInputLabels, ","))
-			}
-			ignoreFirstIntervalsVal = rws.StreamAggrConfig.IgnoreFirstIntervals
-			if ignoreFirstIntervalsVal > 0 {
-				streamAggrIgnoreFirstIntervals.isNotNull = true
-			}
-			ignoreOldSamples = rws.StreamAggrConfig.IgnoreOldSamples
-			if ignoreOldSamples {
-				streamAggrIgnoreOldSamples.isNotNull = true
-			}
-			enableWindows = rws.StreamAggrConfig.EnableWindows
-			if enableWindows {
-				streamAggrEnableWindows.isNotNull = true
-			}
-		}
-		streamAggrConfig.flagSetting += fmt.Sprintf("%s,", streamConfVal)
-		streamAggrKeepInput.flagSetting += fmt.Sprintf("%v,", keepInputVal)
-		streamAggrDropInput.flagSetting += fmt.Sprintf("%v,", dropInputVal)
-		streamAggrDedupInterval.flagSetting += fmt.Sprintf("%s,", dedupIntVal)
-		streamAggrIgnoreFirstIntervals.flagSetting += fmt.Sprintf("%d,", ignoreFirstIntervalsVal)
-		streamAggrIgnoreOldSamples.flagSetting += fmt.Sprintf("%v,", ignoreOldSamples)
-		streamAggrEnableWindows.flagSetting += fmt.Sprintf("%v", enableWindows)
-
-		value = maxDiskUsage
 		if rws.MaxDiskUsage != nil {
-			value = rws.MaxDiskUsage.String()
-			maxDiskUsagePerURL.isNotNull = true
-		}
-		maxDiskUsagePerURL.flagSetting += fmt.Sprintf("%s,", value)
-
-		if rws.ForceVMProto {
-			forceVMProto.isNotNull = true
+			maxDiskUsagePerURL.Add(rws.MaxDiskUsage.String(), i)
 		}
-		forceVMProto.flagSetting += fmt.Sprintf("%t,", rws.ForceVMProto)
 
-		value = ""
+		forceVMProto.Add(strconv.FormatBool(rws.ForceVMProto), i)
 		if rws.ProxyURL != nil {
-			proxyURL.isNotNull = true
-			value = *rws.ProxyURL
+			proxyURL.Add(*rws.ProxyURL, i)
 		}
-		proxyURL.flagSetting += fmt.Sprintf("%s,", value)
 	}
 
-	remoteArgs = append(remoteArgs, url, authUser, bearerTokenFile, urlRelabelConfig, tlsInsecure, sendTimeout, proxyURL)
-	remoteArgs = append(remoteArgs, tlsServerName, tlsKeys, tlsCerts, tlsCAs)
-	remoteArgs = append(remoteArgs, oauth2ClientID, oauth2ClientSecretFile, oauth2Scopes, oauth2TokenURL)
-	remoteArgs = append(remoteArgs, headers, authPasswordFile)
-	remoteArgs = append(remoteArgs, streamAggrConfig, streamAggrKeepInput, streamAggrDedupInterval, streamAggrDropInput, streamAggrDropInputLabels, streamAggrIgnoreFirstIntervals, streamAggrIgnoreOldSamples, streamAggrEnableWindows)
-	remoteArgs = append(remoteArgs, maxDiskUsagePerURL, forceVMProto)
-	remoteArgs = append(remoteArgs, awsEC2Endpoint, awsRegion, awsRoleARN, awsService, awsSTSEndpoint, awsUseSigv4)
-
-	for _, remoteArgType := range remoteArgs {
-		if remoteArgType.isNotNull {
-			finalArgs = append(finalArgs, strings.TrimSuffix(remoteArgType.flagSetting, ","))
-		}
-	}
-	return finalArgs, nil
+	var args []string
+	totalCount := len(remoteTargets)
+	args = build.AppendFlagsToArgs(args, totalCount, url, authUser, bearerTokenFile, urlRelabelConfig, tlsInsecure, sendTimeout, proxyURL)
+	args = build.AppendFlagsToArgs(args, totalCount, tlsServerName, tlsKeys, tlsCerts, tlsCAs)
+	args = build.AppendFlagsToArgs(args, totalCount, oauth2ClientID, oauth2ClientSecretFile, oauth2Scopes, oauth2TokenURL)
+	args = build.AppendFlagsToArgs(args, totalCount, headers, authPasswordFile)
+	args = build.StreamAggrArgsTo(args, "remoteWrite.streamAggr", streamAggrKeys, streamAggrConfigs...)
+	args = build.AppendFlagsToArgs(args, totalCount, maxDiskUsagePerURL, forceVMProto)
+	args = build.AppendFlagsToArgs(args, totalCount, awsEC2Endpoint, awsRegion, awsRoleARN, awsService, awsSTSEndpoint, awsUseSigv4)
+	return args, nil
 }
 
 func buildConfigReloaderContainer(cr *vmv1beta1.VMAgent, extraWatchsMounts []corev1.VolumeMount) corev1.Container {
diff --git a/internal/controller/operator/factory/vmalert/vmalert.go b/internal/controller/operator/factory/vmalert/vmalert.go
index 9473a29e..85cd2658 100644
--- a/internal/controller/operator/factory/vmalert/vmalert.go
+++ b/internal/controller/operator/factory/vmalert/vmalert.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path"
 	"sort"
+	"strconv"
 	"strings"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -529,46 +530,37 @@ func createOrUpdateAssets(ctx context.Context, rclient client.Client, cr, prevCR
 	return nil
 }
 
-type remoteFlag struct {
-	isNotNull   bool
-	flagSetting string
-}
-
 func buildNotifiersArgs(cr *vmv1beta1.VMAlert, ac *build.AssetsCache) ([]string, error) {
-	var finalArgs []string
-	var notifierArgs []remoteFlag
+	var args []string
 	notifierTargets := cr.Spec.Notifiers
 
 	if _, ok := cr.Spec.ExtraArgs["notifier.blackhole"]; ok {
 		// notifier.blackhole disables sending notifications completely, so we don't need to add any notifier args
 		// also no need to add notifier.blackhole to args as it will be added with ExtraArgs
-		return finalArgs, nil
+		return args, nil
 	}
 
 	if len(notifierTargets) == 0 && cr.Spec.NotifierConfigRef != nil {
-		return append(finalArgs, fmt.Sprintf("-notifier.config=%s/%s", notifierConfigMountPath, cr.Spec.NotifierConfigRef.Key)), nil
-	}
-
-	url := remoteFlag{flagSetting: "-notifier.url=", isNotNull: true}
-	authUser := remoteFlag{flagSetting: "-notifier.basicAuth.username="}
-	authPasswordFile := remoteFlag{flagSetting: "-notifier.basicAuth.passwordFile="}
-	tlsCAs := remoteFlag{flagSetting: "-notifier.tlsCAFile="}
-	tlsCerts := remoteFlag{flagSetting: "-notifier.tlsCertFile="}
-	tlsKeys := remoteFlag{flagSetting: "-notifier.tlsKeyFile="}
-	tlsServerName := remoteFlag{flagSetting: "-notifier.tlsServerName="}
-	tlsInSecure := remoteFlag{flagSetting: "-notifier.tlsInsecureSkipVerify="}
-	headers := remoteFlag{flagSetting: "-notifier.headers="}
-	bearerTokenPath := remoteFlag{flagSetting: "-notifier.bearerTokenFile="}
-	oauth2SecretFile := remoteFlag{flagSetting: "-notifier.oauth2.clientSecretFile="}
-	oauth2ClientID := remoteFlag{flagSetting: "-notifier.oauth2.clientID="}
-	oauth2Scopes := remoteFlag{flagSetting: "-notifier.oauth2.scopes="}
-	oauth2TokenURL := remoteFlag{flagSetting: "-notifier.oauth2.tokenUrl="}
-
-	for _, nt := range notifierTargets {
-		url.flagSetting += fmt.Sprintf("%s,", nt.URL)
-
-		var caPath, certPath, keyPath, ServerName string
-		var inSecure bool
+		return append(args, fmt.Sprintf("-notifier.config=%s/%s", notifierConfigMountPath, cr.Spec.NotifierConfigRef.Key)), nil
+	}
+
+	url := build.NewFlag("-notifier.url", "")
+	authUser := build.NewFlag("-notifier.basicAuth.username", `""`)
+	authPasswordFile := build.NewFlag("-notifier.basicAuth.passwordFile", "")
+	tlsCAs := build.NewFlag("-notifier.tlsCAFile", "")
+	tlsCerts := build.NewFlag("-notifier.tlsCertFile", "")
+	tlsKeys := build.NewFlag("-notifier.tlsKeyFile", "")
+	tlsServerName := build.NewFlag("-notifier.tlsServerName", "")
+	tlsInsecure := build.NewFlag("-notifier.tlsInsecureSkipVerify", "false")
+	headers := build.NewFlag("-notifier.headers", "")
+	bearerTokenPath := build.NewFlag("-notifier.bearerTokenFile", "")
+	oauth2SecretFile := build.NewFlag("-notifier.oauth2.clientSecretFile", "")
+	oauth2ClientID := build.NewFlag("-notifier.oauth2.clientID", "")
+	oauth2Scopes := build.NewFlag("-notifier.oauth2.scopes", "")
+	oauth2TokenURL := build.NewFlag("-notifier.oauth2.tokenUrl", "")
+
+	for i, nt := range notifierTargets {
+		url.Add(nt.URL, i)
 		ntTLS := nt.TLSConfig
 		if ntTLS != nil {
 			creds, err := ac.BuildTLSCreds(cr.Namespace, ntTLS)
@@ -576,123 +568,85 @@ func buildNotifiersArgs(cr *vmv1beta1.VMAlert, ac *build.AssetsCache) ([]string,
 				return nil, fmt.Errorf("failed to load notifier TLS credentials: %w", err)
 			}
 			if creds.CAFile != "" {
-				caPath = creds.CAFile
-				tlsCAs.isNotNull = true
+				tlsCAs.Add(creds.CAFile, i)
 			}
 			if creds.CertFile != "" {
-				certPath = creds.CertFile
-				tlsCerts.isNotNull = true
+				tlsCerts.Add(creds.CertFile, i)
 			}
 			if creds.KeyFile != "" {
-				keyPath = creds.KeyFile
-				tlsKeys.isNotNull = true
-			}
-			if ntTLS.InsecureSkipVerify {
-				tlsInSecure.isNotNull = true
-				inSecure = true
+				tlsKeys.Add(creds.KeyFile, i)
 			}
+			tlsInsecure.Add(strconv.FormatBool(ntTLS.InsecureSkipVerify), i)
 			if ntTLS.ServerName != "" {
-				ServerName = ntTLS.ServerName
-				tlsServerName.isNotNull = true
+				tlsServerName.Add(ntTLS.ServerName, i)
 			}
 		}
-		tlsCAs.flagSetting += fmt.Sprintf("%s,", caPath)
-		tlsCerts.flagSetting += fmt.Sprintf("%s,", certPath)
-		tlsKeys.flagSetting += fmt.Sprintf("%s,", keyPath)
-		tlsServerName.flagSetting += fmt.Sprintf("%s,", ServerName)
-		tlsInSecure.flagSetting += fmt.Sprintf("%v,", inSecure)
-		var headerFlagValue string
 		if len(nt.Headers) > 0 {
+			var value string
 			for _, headerKV := range nt.Headers {
-				headerFlagValue += headerKV + "^^"
+				value += headerKV + "^^"
 			}
-			headers.isNotNull = true
+			headers.Add(strings.TrimSuffix(value, "^^"), i)
 		}
-		headerFlagValue = strings.TrimSuffix(headerFlagValue, "^^")
-		headers.flagSetting += fmt.Sprintf("%s,", headerFlagValue)
-		var user, passFile string
 		if nt.BasicAuth != nil {
 			if len(nt.BasicAuth.PasswordFile) > 0 {
-				passFile = nt.BasicAuth.PasswordFile
-				authPasswordFile.isNotNull = true
+				authPasswordFile.Add(nt.BasicAuth.PasswordFile, i)
 			} else {
 				file, err := ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, &nt.BasicAuth.Password)
 				if err != nil {
 					return nil, err
 				}
-				passFile = file
-				authPasswordFile.isNotNull = true
+				authPasswordFile.Add(file, i)
 			}
-			authUser.isNotNull = true
 			secret, err := ac.LoadKeyFromSecret(cr.Namespace, &nt.BasicAuth.Username)
 			if err != nil {
 				return nil, err
 			}
-			user = secret
+			authUser.Add(strconv.Quote(secret), i)
 		}
-		authUser.flagSetting += fmt.Sprintf("\"%s\",", strings.ReplaceAll(user, `"`, `\"`))
-		authPasswordFile.flagSetting += fmt.Sprintf("%s,", passFile)
-
-		var tokenPath string
 		if nt.BearerAuth != nil {
 			if nt.TokenSecret != nil {
 				file, err := ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, nt.TokenSecret)
 				if err != nil {
 					return nil, err
 				}
-				bearerTokenPath.isNotNull = true
-				tokenPath = file
+				bearerTokenPath.Add(file, i)
 			}
 			if len(nt.TokenFilePath) > 0 {
-				bearerTokenPath.isNotNull = true
-				tokenPath = nt.TokenFilePath
+				bearerTokenPath.Add(nt.TokenFilePath, i)
 			}
 		}
-		bearerTokenPath.flagSetting += fmt.Sprintf("%s,", tokenPath)
-		var scopes, tokenURL, secretFile, clientID string
 		if nt.OAuth2 != nil {
 			if nt.OAuth2.ClientSecret != nil {
 				file, err := ac.LoadPathFromSecret(build.SecretConfigResourceKind, cr.Namespace, nt.OAuth2.ClientSecret)
 				if err != nil {
 					return nil, err
 				}
-				oauth2SecretFile.isNotNull = true
-				secretFile = file
+				oauth2SecretFile.Add(file, i)
 			} else {
-				oauth2SecretFile.isNotNull = true
-				secretFile = nt.OAuth2.ClientSecretFile
+				oauth2SecretFile.Add(nt.OAuth2.ClientSecretFile, i)
 			}
 			if len(nt.OAuth2.Scopes) > 0 {
-				oauth2Scopes.isNotNull = true
-				scopes = strings.Join(nt.OAuth2.Scopes, ",")
+				oauth2Scopes.Add(strings.Join(nt.OAuth2.Scopes, ","), i)
 			}
 			if len(nt.OAuth2.TokenURL) > 0 {
-				oauth2TokenURL.isNotNull = true
-				tokenURL = nt.OAuth2.TokenURL
+				oauth2TokenURL.Add(nt.OAuth2.TokenURL, i)
 			}
 			secret, err := ac.LoadKeyFromSecretOrConfigMap(cr.Namespace, &nt.OAuth2.ClientID)
 			if err != nil {
 				return nil, err
 			}
-			clientID = secret
-			oauth2ClientID.isNotNull = true
+			oauth2ClientID.Add(secret, i)
 		}
-		oauth2Scopes.flagSetting += fmt.Sprintf("%s,", scopes)
-		oauth2TokenURL.flagSetting += fmt.Sprintf("%s,", tokenURL)
-		oauth2ClientID.flagSetting += fmt.Sprintf("%s,", clientID)
-		oauth2SecretFile.flagSetting += fmt.Sprintf("%s,", secretFile)
 	}
-	notifierArgs = append(notifierArgs, url, authUser, authPasswordFile)
-	notifierArgs = append(notifierArgs, tlsServerName, tlsKeys, tlsCerts, tlsCAs, tlsInSecure, headers, bearerTokenPath)
-	notifierArgs = append(notifierArgs, oauth2SecretFile, oauth2ClientID, oauth2Scopes, oauth2TokenURL)
-
-	for _, remoteArgType := range notifierArgs {
-		if remoteArgType.isNotNull {
-			finalArgs = append(finalArgs, strings.TrimSuffix(remoteArgType.flagSetting, ","))
-		}
+	if !url.IsSet() {
+		args = append(args, "-notifier.url=")
 	}
-
-	return finalArgs, nil
+	totalCount := len(notifierTargets)
+	args = build.AppendFlagsToArgs(args, totalCount, url, authUser, authPasswordFile)
+	args = build.AppendFlagsToArgs(args, totalCount, tlsServerName, tlsKeys, tlsCerts, tlsCAs, tlsInsecure, headers, bearerTokenPath)
+	args = build.AppendFlagsToArgs(args, totalCount, oauth2SecretFile, oauth2ClientID, oauth2Scopes, oauth2TokenURL)
+	return args, nil
 }
 
 func buildConfigReloaderContainer(cr *vmv1beta1.VMAlert, ruleConfigMapNames []string, extraWatchVolumeMounts []corev1.VolumeMount) corev1.Container {
diff --git a/internal/controller/operator/factory/vmalertmanager/statefulset.go b/internal/controller/operator/factory/vmalertmanager/statefulset.go
index 74ce30b9..20dc6b2e 100644
--- a/internal/controller/operator/factory/vmalertmanager/statefulset.go
+++ b/internal/controller/operator/factory/vmalertmanager/statefulset.go
@@ -68,6 +68,10 @@ func newStsForAlertManager(cr *vmv1beta1.VMAlertmanager) (*appsv1.StatefulSet, e
 		},
 		Spec: *spec,
 	}
+	if cr.Spec.PersistentVolumeClaimRetentionPolicy != nil {
+
+		statefulset.Spec.PersistentVolumeClaimRetentionPolicy = cr.Spec.PersistentVolumeClaimRetentionPolicy
+	}
 	build.StatefulSetAddCommonParams(statefulset, ptr.Deref(cr.Spec.UseStrictSecurity, false), &cr.Spec.CommonApplicationDeploymentParams)
 	cr.Spec.Storage.IntoSTSVolume(cr.GetVolumeName(), &statefulset.Spec)
 	statefulset.Spec.Template.Spec.Volumes = append(statefulset.Spec.Template.Spec.Volumes, cr.Spec.Volumes...)
diff --git a/internal/controller/operator/factory/vmanomaly/statefulset.go b/internal/controller/operator/factory/vmanomaly/statefulset.go
index 27ac4024..c7cc1a16 100644
--- a/internal/controller/operator/factory/vmanomaly/statefulset.go
+++ b/internal/controller/operator/factory/vmanomaly/statefulset.go
@@ -170,7 +170,9 @@ func newStatefulSet(cr *vmv1.VMAnomaly, configHash string, ac *build.AssetsCache
 			},
 		},
 	}
-
+	if cr.Spec.PersistentVolumeClaimRetentionPolicy != nil {
+		app.Spec.PersistentVolumeClaimRetentionPolicy = cr.Spec.PersistentVolumeClaimRetentionPolicy
+	}
 	build.StatefulSetAddCommonParams(app, useStrictSecurity, &cr.Spec.CommonApplicationDeploymentParams)
 	app.Spec.Template.Spec.Volumes = append(app.Spec.Template.Spec.Volumes, cr.Spec.Volumes...)
 	cr.Spec.Storage.IntoSTSVolume(cr.GetVolumeName(), &app.Spec)
diff --git a/internal/controller/operator/factory/vmcluster/vmcluster.go b/internal/controller/operator/factory/vmcluster/vmcluster.go
index fc4f896c..061926ee 100644
--- a/internal/controller/operator/factory/vmcluster/vmcluster.go
+++ b/internal/controller/operator/factory/vmcluster/vmcluster.go
@@ -175,6 +175,7 @@ func createOrUpdateVMSelect(ctx context.Context, rclient client.Client, cr, prev
 				cr.Spec.VMSelect.ReplicaCount = count
 			}
 		},
+		UpdateBehavior: cr.Spec.VMSelect.RollingUpdateStrategyBehavior,
 	}
 	return reconcile.HandleSTSUpdate(ctx, rclient, stsOpts, newSts, prevSts)
 }
@@ -390,6 +391,7 @@ func createOrUpdateVMStorage(ctx context.Context, rclient client.Client, cr, pre
 	stsOpts := reconcile.STSOptions{
 		HasClaim:       len(newSts.Spec.VolumeClaimTemplates) > 0,
 		SelectorLabels: cr.VMStorageSelectorLabels,
+		UpdateBehavior: cr.Spec.VMStorage.RollingUpdateStrategyBehavior,
 	}
 	return reconcile.HandleSTSUpdate(ctx, rclient, stsOpts, newSts, prevSts)
 }
@@ -515,6 +517,9 @@ func genVMSelectSpec(cr *vmv1beta1.VMCluster) (*appsv1.StatefulSet, error) {
 			ServiceName: cr.GetVMSelectName(),
 		},
 	}
+	if cr.Spec.VMSelect.PersistentVolumeClaimRetentionPolicy != nil {
+		stsSpec.Spec.PersistentVolumeClaimRetentionPolicy = cr.Spec.VMSelect.PersistentVolumeClaimRetentionPolicy
+	}
 	build.StatefulSetAddCommonParams(stsSpec, ptr.Deref(cr.Spec.VMSelect.UseStrictSecurity, false), &cr.Spec.VMSelect.CommonApplicationDeploymentParams)
 	if cr.Spec.VMSelect.CacheMountPath != "" {
 		storageSpec := cr.Spec.VMSelect.Storage
@@ -932,6 +937,9 @@ func buildVMStorageSpec(ctx context.Context, cr *vmv1beta1.VMCluster) (*appsv1.S
 			ServiceName: cr.GetVMStorageName(),
 		},
 	}
+	if cr.Spec.VMStorage.PersistentVolumeClaimRetentionPolicy != nil {
+		stsSpec.Spec.PersistentVolumeClaimRetentionPolicy = cr.Spec.VMStorage.PersistentVolumeClaimRetentionPolicy
+	}
 	build.StatefulSetAddCommonParams(stsSpec, ptr.Deref(cr.Spec.VMStorage.UseStrictSecurity, false), &cr.Spec.VMStorage.CommonApplicationDeploymentParams)
 	storageSpec := cr.Spec.VMStorage.Storage
 	storageSpec.IntoSTSVolume(cr.Spec.VMStorage.GetStorageVolumeName(), &stsSpec.Spec)
diff --git a/internal/controller/operator/factory/vmsingle/vmsingle.go b/internal/controller/operator/factory/vmsingle/vmsingle.go
index fdd6f7b9..44d0ba92 100644
--- a/internal/controller/operator/factory/vmsingle/vmsingle.go
+++ b/internal/controller/operator/factory/vmsingle/vmsingle.go
@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"path"
 	"sort"
-	"strings"
 
 	"gopkg.in/yaml.v2"
 	appsv1 "k8s.io/api/apps/v1"
@@ -29,17 +28,17 @@ const (
 	streamAggrSecretKey = "config.yaml"
 )
 
-func createVMSingleStorage(ctx context.Context, rclient client.Client, cr, prevCR *vmv1beta1.VMSingle) error {
-	newPvc := makeVMSinglePvc(cr)
+func createStorage(ctx context.Context, rclient client.Client, cr, prevCR *vmv1beta1.VMSingle) error {
+	newPvc := makePvc(cr)
 	var prevPVC *corev1.PersistentVolumeClaim
 	if prevCR != nil && prevCR.Spec.Storage != nil {
-		prevPVC = makeVMSinglePvc(prevCR)
+		prevPVC = makePvc(prevCR)
 	}
 
 	return reconcile.PersistentVolumeClaim(ctx, rclient, newPvc, prevPVC)
 }
 
-func makeVMSinglePvc(cr *vmv1beta1.VMSingle) *corev1.PersistentVolumeClaim {
+func makePvc(cr *vmv1beta1.VMSingle) *corev1.PersistentVolumeClaim {
 	pvcObject := &corev1.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            cr.PrefixedName(),
@@ -84,7 +83,7 @@ func CreateOrUpdate(ctx context.Context, cr *vmv1beta1.VMSingle, rclient client.
 	}
 
 	if cr.Spec.Storage != nil && cr.Spec.StorageDataPath == "" {
-		if err := createVMSingleStorage(ctx, rclient, cr, prevCR); err != nil {
+		if err := createStorage(ctx, rclient, cr, prevCR); err != nil {
 			return fmt.Errorf("cannot create storage: %w", err)
 		}
 	}
@@ -101,12 +100,12 @@ func CreateOrUpdate(ctx context.Context, cr *vmv1beta1.VMSingle, rclient client.
 	}
 	var prevDeploy *appsv1.Deployment
 	if prevCR != nil {
-		prevDeploy, err = newDeployForVMSingle(ctx, prevCR)
+		prevDeploy, err = newDeploy(ctx, prevCR)
 		if err != nil {
 			return fmt.Errorf("cannot generate prev deploy spec: %w", err)
 		}
 	}
-	newDeploy, err := newDeployForVMSingle(ctx, cr)
+	newDeploy, err := newDeploy(ctx, cr)
 	if err != nil {
 		return fmt.Errorf("cannot generate new deploy for vmsingle: %w", err)
 	}
@@ -114,9 +113,9 @@ func CreateOrUpdate(ctx context.Context, cr *vmv1beta1.VMSingle, rclient client.
 	return reconcile.Deployment(ctx, rclient, newDeploy, prevDeploy, false)
 }
 
-func newDeployForVMSingle(ctx context.Context, cr *vmv1beta1.VMSingle) (*appsv1.Deployment, error) {
+func newDeploy(ctx context.Context, cr *vmv1beta1.VMSingle) (*appsv1.Deployment, error) {
 
-	podSpec, err := makeSpecForVMSingle(ctx, cr)
+	podSpec, err := makeSpec(ctx, cr)
 	if err != nil {
 		return nil, err
 	}
@@ -145,7 +144,7 @@ func newDeployForVMSingle(ctx context.Context, cr *vmv1beta1.VMSingle) (*appsv1.
 	return depSpec, nil
 }
 
-func makeSpecForVMSingle(ctx context.Context, cr *vmv1beta1.VMSingle) (*corev1.PodTemplateSpec, error) {
+func makeSpec(ctx context.Context, cr *vmv1beta1.VMSingle) (*corev1.PodTemplateSpec, error) {
 	var args []string
 
 	if cr.Spec.RetentionPeriod != "" {
@@ -235,43 +234,10 @@ func makeSpecForVMSingle(ctx context.Context, cr *vmv1beta1.VMSingle) (*corev1.P
 		})
 	}
 
-	if cr.HasAnyStreamAggrRule() {
-		volumes = append(volumes, corev1.Volume{
-			Name: k8stools.SanitizeVolumeName("stream-aggr-conf"),
-			VolumeSource: corev1.VolumeSource{
-				ConfigMap: &corev1.ConfigMapVolumeSource{
-					LocalObjectReference: corev1.LocalObjectReference{
-						Name: cr.StreamAggrConfigName(),
-					},
-				},
-			},
-		})
-		vmMounts = append(vmMounts, corev1.VolumeMount{
-			Name:      k8stools.SanitizeVolumeName("stream-aggr-conf"),
-			ReadOnly:  true,
-			MountPath: vmv1beta1.StreamAggrConfigDir,
-		})
-
-		args = append(args, fmt.Sprintf("--streamAggr.config=%s", path.Join(vmv1beta1.StreamAggrConfigDir, streamAggrSecretKey)))
-		if cr.Spec.StreamAggrConfig.KeepInput {
-			args = append(args, "--streamAggr.keepInput=true")
-		}
-		if cr.Spec.StreamAggrConfig.DropInput {
-			args = append(args, "--streamAggr.dropInput=true")
-		}
-		if len(cr.Spec.StreamAggrConfig.DropInputLabels) > 0 {
-			args = append(args, fmt.Sprintf("--streamAggr.dropInputLabels=%s", strings.Join(cr.Spec.StreamAggrConfig.DropInputLabels, ",")))
-		}
-		if cr.Spec.StreamAggrConfig.IgnoreFirstIntervals > 0 {
-			args = append(args, fmt.Sprintf("--streamAggr.ignoreFirstIntervals=%d", cr.Spec.StreamAggrConfig.IgnoreFirstIntervals))
-		}
-		if cr.Spec.StreamAggrConfig.IgnoreOldSamples {
-			args = append(args, "--streamAggr.ignoreOldSamples=true")
-		}
-		if cr.Spec.StreamAggrConfig.EnableWindows {
-			args = append(args, "--streamAggr.enableWindows=true")
-		}
-	}
+	volumes, vmMounts = build.StreamAggrVolumeTo(volumes, vmMounts, build.ResourceName(build.StreamAggrConfigResourceKind, cr), cr.Spec.StreamAggrConfig)
+	streamAggrKeys := []string{streamAggrSecretKey}
+	streamAggrConfigs := []*vmv1beta1.StreamAggrConfig{cr.Spec.StreamAggrConfig}
+	args = build.StreamAggrArgsTo(args, "streamAggr", streamAggrKeys, streamAggrConfigs...)
 
 	// deduplication can work without stream aggregation rules
 	if cr.Spec.StreamAggrConfig != nil && cr.Spec.StreamAggrConfig.DedupInterval != "" {
@@ -405,10 +371,10 @@ func createOrUpdateService(ctx context.Context, rclient client.Client, cr, prevC
 	return newService, nil
 }
 
-// buildVMSingleStreamAggrConfig build configmap with stream aggregation config for vmsingle.
-func buildVMSingleStreamAggrConfig(ctx context.Context, cr *vmv1beta1.VMSingle, rclient client.Client) (*corev1.ConfigMap, error) {
+// buildStreamAggrConfig build configmap with stream aggregation config for vmsingle.
+func buildStreamAggrConfig(ctx context.Context, cr *vmv1beta1.VMSingle, rclient client.Client) (*corev1.ConfigMap, error) {
 	cfgCM := &corev1.ConfigMap{
-		ObjectMeta: buildStreamAggrConfigMeta(cr),
+		ObjectMeta: build.ResourceMeta(build.StreamAggrConfigResourceKind, cr),
 		Data:       make(map[string]string),
 	}
 	if len(cr.Spec.StreamAggrConfig.Rules) > 0 {
@@ -434,28 +400,18 @@ func buildVMSingleStreamAggrConfig(ctx context.Context, cr *vmv1beta1.VMSingle,
 	return cfgCM, nil
 }
 
-func buildStreamAggrConfigMeta(cr *vmv1beta1.VMSingle) metav1.ObjectMeta {
-	return metav1.ObjectMeta{
-		Namespace:       cr.Namespace,
-		Name:            cr.StreamAggrConfigName(),
-		Labels:          cr.AllLabels(),
-		Annotations:     cr.AnnotationsFiltered(),
-		OwnerReferences: cr.AsOwner(),
-	}
-}
-
 // createOrUpdateStreamAggrConfig builds stream aggregation configs for vmsingle at separate configmap, serialized as yaml
 func createOrUpdateStreamAggrConfig(ctx context.Context, rclient client.Client, cr, prevCR *vmv1beta1.VMSingle) error {
 	if !cr.HasAnyStreamAggrRule() {
 		return nil
 	}
-	streamAggrCM, err := buildVMSingleStreamAggrConfig(ctx, cr, rclient)
+	streamAggrCM, err := buildStreamAggrConfig(ctx, cr, rclient)
 	if err != nil {
 		return err
 	}
 	var prevCMMeta *metav1.ObjectMeta
 	if prevCR != nil {
-		prevCMMeta = ptr.To(buildStreamAggrConfigMeta(prevCR))
+		prevCMMeta = ptr.To(build.ResourceMeta(build.StreamAggrConfigResourceKind, prevCR))
 	}
 	return reconcile.ConfigMap(ctx, rclient, streamAggrCM, prevCMMeta)
 }



Test patch:
diff --git a/internal/controller/operator/factory/reconcile/statefulset_test.go b/internal/controller/operator/factory/reconcile/statefulset_test.go
index b421147f..79d8481f 100644
--- a/internal/controller/operator/factory/reconcile/statefulset_test.go
+++ b/internal/controller/operator/factory/reconcile/statefulset_test.go
@@ -288,9 +288,10 @@ func Test_podIsReady(t *testing.T) {
 
 func Test_performRollingUpdateOnSts(t *testing.T) {
 	type args struct {
-		stsName   string
-		ns        string
-		podLabels map[string]string
+		stsName           string
+		ns                string
+		podLabels         map[string]string
+		podMaxUnavailable int
 	}
 	tests := []struct {
 		name              string
@@ -302,9 +303,10 @@ func Test_performRollingUpdateOnSts(t *testing.T) {
 		{
 			name: "rolling update is not needed",
 			args: args{
-				stsName:   "vmselect-sts",
-				ns:        "default",
-				podLabels: map[string]string{"app": "vmselect"},
+				stsName:           "vmselect-sts",
+				ns:                "default",
+				podLabels:         map[string]string{"app": "vmselect"},
+				podMaxUnavailable: 1,
 			},
 			predefinedObjects: []runtime.Object{
 				&appsv1.StatefulSet{
@@ -339,9 +341,10 @@ func Test_performRollingUpdateOnSts(t *testing.T) {
 		{
 			name: "rolling update is timeout",
 			args: args{
-				stsName:   "vmselect-sts",
-				ns:        "default",
-				podLabels: map[string]string{"app": "vmselect"},
+				stsName:           "vmselect-sts",
+				ns:                "default",
+				podLabels:         map[string]string{"app": "vmselect"},
+				podMaxUnavailable: 1,
 			},
 			predefinedObjects: []runtime.Object{
 				&appsv1.StatefulSet{
@@ -380,7 +383,7 @@ func Test_performRollingUpdateOnSts(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			fclient := k8stools.GetTestClientWithObjects(tt.predefinedObjects)
 
-			if err := performRollingUpdateOnSts(context.Background(), false, fclient, tt.args.stsName, tt.args.ns, tt.args.podLabels); (err != nil) != tt.wantErr {
+			if err := performRollingUpdateOnSts(context.Background(), false, fclient, tt.args.stsName, tt.args.ns, tt.args.podLabels, tt.args.podMaxUnavailable); (err != nil) != tt.wantErr {
 				t.Errorf("performRollingUpdateOnSts() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
diff --git a/internal/controller/operator/factory/vmagent/vmagent_test.go b/internal/controller/operator/factory/vmagent/vmagent_test.go
index 86cdf59b..f600fe80 100644
--- a/internal/controller/operator/factory/vmagent/vmagent_test.go
+++ b/internal/controller/operator/factory/vmagent/vmagent_test.go
@@ -1826,7 +1826,12 @@ func TestCreateOrUpdateStreamAggrConfig(t *testing.T) {
 				t.Fatalf("CreateOrUpdateStreamAggrConfig() error = %v, wantErr %v", err, tt.wantErr)
 			}
 			var createdCM corev1.ConfigMap
-			if err := cl.Get(ctx, types.NamespacedName{Namespace: tt.cr.Namespace, Name: tt.cr.StreamAggrConfigName()}, &createdCM); err != nil {
+			if err := cl.Get(ctx,
+				types.NamespacedName{
+					Namespace: tt.cr.Namespace,
+					Name:      build.ResourceName(build.StreamAggrConfigResourceKind, tt.cr),
+				}, &createdCM,
+			); err != nil {
 				t.Fatalf("cannot fetch created cm: %v", err)
 			}
 			if err := tt.validate(&createdCM); err != nil {
diff --git a/test/e2e/childobjects/vmalertmanagerconfig_test.go b/test/e2e/childobjects/vmalertmanagerconfig_test.go
index 98b3954c..bc951885 100644
--- a/test/e2e/childobjects/vmalertmanagerconfig_test.go
+++ b/test/e2e/childobjects/vmalertmanagerconfig_test.go
@@ -16,7 +16,7 @@ import (
 )
 
 //nolint:dupl,lll
-var _ = Describe("test vmalertmanagerconfig Controller", func() {
+var _ = Describe("test vmalertmanagerconfig Controller", Label("vm", "child", "alertmanager"), func() {
 	namespace := fmt.Sprintf("default-%d", GinkgoParallelProcess())
 	ctx := context.Background()
 	type opts struct {
diff --git a/test/e2e/childobjects/vmrule_test.go b/test/e2e/childobjects/vmrule_test.go
index 872aa308..ba39fb58 100644
--- a/test/e2e/childobjects/vmrule_test.go
+++ b/test/e2e/childobjects/vmrule_test.go
@@ -16,7 +16,7 @@ import (
 )
 
 //nolint:dupl,lll
-var _ = Describe("test vmrule Controller", func() {
+var _ = Describe("test vmrule Controller", Label("vm", "child", "alert"), func() {
 	namespace := fmt.Sprintf("default-%d", GinkgoParallelProcess())
 	ctx := context.Background()
 	type opts struct {
diff --git a/test/e2e/childobjects/vmuser_test.go b/test/e2e/childobjects/vmuser_test.go
index c112b855..69e2cace 100644
--- a/test/e2e/childobjects/vmuser_test.go
+++ b/test/e2e/childobjects/vmuser_test.go
@@ -17,7 +17,7 @@ import (
 )
 
 //nolint:dupl,lll
-var _ = Describe("test vmuser Controller", func() {
+var _ = Describe("test vmuser Controller", Label("vm", "child", "auth"), func() {
 	namespace := fmt.Sprintf("default-%d", GinkgoParallelProcess())
 	ctx := context.Background()
 	type opts struct {
diff --git a/test/e2e/vmcluster_test.go b/test/e2e/vmcluster_test.go
index 3743d163..5a1507f5 100644
--- a/test/e2e/vmcluster_test.go
+++ b/test/e2e/vmcluster_test.go
@@ -11,6 +11,7 @@ import (
 	policyv1 "k8s.io/api/policy/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/ptr"
@@ -1249,6 +1250,238 @@ up{baz="bar"} 123
 			),
 		)
 	})
+	Context("update", func() {
+		AfterEach(func() {
+			Expect(k8sClient.Delete(ctx, &vmv1beta1.VMCluster{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: namespace,
+					Name:      namespacedName.Name,
+				},
+			})).To(Succeed())
+			Eventually(func() error {
+				return k8sClient.Get(context.Background(), types.NamespacedName{
+					Name:      namespacedName.Name,
+					Namespace: namespace,
+				}, &vmv1beta1.VMCluster{})
+
+			}, eventualDeletionTimeout).WithContext(ctx).Should(MatchError(k8serrors.IsNotFound, "want not found error"))
+		})
+
+		type testStep struct {
+			setup  func(*vmv1beta1.VMCluster)
+			modify func(*vmv1beta1.VMCluster)
+			verify func(*vmv1beta1.VMCluster)
+		}
+
+		DescribeTable("should respect customized behavior",
+			func(name string, initCR *vmv1beta1.VMCluster, steps ...testStep) {
+				namespacedName.Name = name
+				initCR.Namespace = namespace
+				initCR.Name = name
+				ctx = context.Background()
+				Expect(k8sClient.Create(ctx, initCR)).To(Succeed())
+				Eventually(func() error {
+					return expectObjectStatusOperational(ctx, k8sClient, initCR, namespacedName)
+				}, eventualStatefulsetAppReadyTimeout).WithContext(ctx).Should(Succeed())
+				for _, step := range steps {
+					if step.setup != nil {
+						step.setup(initCR)
+					}
+					// update and verify immediately
+					Eventually(func() error {
+						var toUpdate vmv1beta1.VMCluster
+						if err := k8sClient.Get(ctx, namespacedName, &toUpdate); err != nil {
+							return err
+						}
+						step.modify(&toUpdate)
+						return k8sClient.Update(ctx, &toUpdate)
+					}, eventualExpandingTimeout).WithContext(ctx).Should(Succeed())
+					var updated vmv1beta1.VMCluster
+					Expect(k8sClient.Get(ctx, namespacedName, &updated)).To(Succeed())
+					step.verify(&updated)
+					Eventually(func() error {
+						return expectObjectStatusOperational(ctx, k8sClient, initCR, namespacedName)
+					}, eventualStatefulsetAppReadyTimeout).WithContext(ctx).Should(Succeed())
+				}
+			},
+			Entry("configures vmstorage with MaxUnavailable 2", "maxunavailable-2-integer",
+				&vmv1beta1.VMCluster{
+					Spec: vmv1beta1.VMClusterSpec{
+						RetentionPeriod: "1",
+						VMStorage: &vmv1beta1.VMStorage{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](5),
+							},
+							PodMetadata: &vmv1beta1.EmbeddedObjectMetadata{
+								Labels: map[string]string{"version": "old"},
+							},
+							RollingUpdateStrategyBehavior: &vmv1beta1.StatefulSetUpdateStrategyBehavior{
+								MaxUnavailable: ptr.To(intstr.FromInt32(2)),
+							},
+						},
+						VMSelect: &vmv1beta1.VMSelect{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](1),
+							},
+						},
+						VMInsert: &vmv1beta1.VMInsert{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](1),
+							},
+						},
+					},
+				},
+				testStep{
+					modify: func(cr *vmv1beta1.VMCluster) {
+						By("triggering a rolling update")
+						cr.Spec.VMStorage.PodMetadata.Labels["version"] = "new"
+					},
+					verify: func(cr *vmv1beta1.VMCluster) {
+						By("checking that update process runs as configured")
+						Eventually(func() int {
+							podList := &corev1.PodList{}
+							k8sClient.List(ctx, podList, &client.ListOptions{
+								Namespace:     namespace,
+								LabelSelector: labels.SelectorFromSet(cr.VMStorageSelectorLabels()),
+							})
+							podsUpdated := 0
+							podsUnavailable := 0
+							for _, pod := range podList.Items {
+								if pod.Status.Phase == corev1.PodRunning && pod.Labels["version"] == "new" {
+									podsUpdated++
+								}
+								if pod.Status.Phase != corev1.PodRunning {
+									podsUnavailable++
+								}
+							}
+							Expect(podsUnavailable).To(BeNumerically("<=", 2), "no more than 2 pods should be unavailable during the update")
+							return podsUpdated
+						}, eventualStatefulsetAppReadyTimeout).Should(BeNumerically("==", 5))
+					},
+				},
+			),
+			Entry("configures vmstorage with MaxUnavailable 100%", "maxunavailable-100-percent",
+				&vmv1beta1.VMCluster{
+					Spec: vmv1beta1.VMClusterSpec{
+						RetentionPeriod: "1",
+						VMStorage: &vmv1beta1.VMStorage{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](5),
+							},
+							PodMetadata: &vmv1beta1.EmbeddedObjectMetadata{
+								Labels: map[string]string{"version": "old"},
+							},
+							RollingUpdateStrategyBehavior: &vmv1beta1.StatefulSetUpdateStrategyBehavior{
+								MaxUnavailable: ptr.To(intstr.FromString("100%")),
+							},
+						},
+						VMSelect: &vmv1beta1.VMSelect{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](1),
+							},
+						},
+						VMInsert: &vmv1beta1.VMInsert{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](1),
+							},
+						},
+					},
+				},
+				testStep{
+					modify: func(cr *vmv1beta1.VMCluster) {
+						By("triggering a rolling update")
+						cr.Spec.VMStorage.PodMetadata.Labels["version"] = "new"
+					},
+					verify: func(cr *vmv1beta1.VMCluster) {
+						By("checking that update process runs as configured")
+						Eventually(func() int {
+							podList := &corev1.PodList{}
+							k8sClient.List(ctx, podList, &client.ListOptions{
+								Namespace:     namespace,
+								LabelSelector: labels.SelectorFromSet(cr.VMStorageSelectorLabels()),
+							})
+							podsUpdated := 0
+							podsOutdated := 0
+							for _, pod := range podList.Items {
+								if pod.Status.Phase == corev1.PodRunning && pod.Labels["version"] == "new" {
+									podsUpdated++
+								}
+								if pod.Status.Phase == corev1.PodRunning && pod.Labels["version"] == "old" {
+									podsOutdated++
+								}
+							}
+							// consider the update is in progress if at least one pod is updated and running,
+							// this is to avoid catching false positives due to async pod evictions/creations
+							if podsUpdated >= 1 {
+								Expect(podsOutdated).To(BeNumerically("==", 0), "no pods are running with old version when the update is in progress")
+							}
+							return podsUpdated
+						}, eventualStatefulsetAppReadyTimeout).Should(BeNumerically("==", 5))
+					},
+				},
+			),
+			Entry("configures vmstorage with MaxUnavailable 100% but limited by a PDB", "maxunavailable-100-percent-pdb",
+				&vmv1beta1.VMCluster{
+					Spec: vmv1beta1.VMClusterSpec{
+						RetentionPeriod: "1",
+						VMStorage: &vmv1beta1.VMStorage{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](5),
+							},
+							PodMetadata: &vmv1beta1.EmbeddedObjectMetadata{
+								Labels: map[string]string{"version": "old"},
+							},
+							RollingUpdateStrategyBehavior: &vmv1beta1.StatefulSetUpdateStrategyBehavior{
+								MaxUnavailable: ptr.To(intstr.FromString("100%")),
+							},
+							PodDisruptionBudget: &vmv1beta1.EmbeddedPodDisruptionBudgetSpec{
+								MinAvailable: ptr.To(intstr.FromInt32(4)),
+							},
+						},
+						VMSelect: &vmv1beta1.VMSelect{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](1),
+							},
+						},
+						VMInsert: &vmv1beta1.VMInsert{
+							CommonApplicationDeploymentParams: vmv1beta1.CommonApplicationDeploymentParams{
+								ReplicaCount: ptr.To[int32](1),
+							},
+						},
+					},
+				},
+				testStep{
+					modify: func(cr *vmv1beta1.VMCluster) {
+						By("triggering a rolling update")
+						cr.Spec.VMStorage.PodMetadata.Labels["version"] = "new"
+					},
+					verify: func(cr *vmv1beta1.VMCluster) {
+						By("checking that update process runs as configured")
+						Eventually(func() int {
+							podList := &corev1.PodList{}
+							k8sClient.List(ctx, podList, &client.ListOptions{
+								Namespace:     namespace,
+								LabelSelector: labels.SelectorFromSet(cr.VMStorageSelectorLabels()),
+							})
+							podsUpdated := 0
+							for _, pod := range podList.Items {
+								if pod.Status.Phase == corev1.PodRunning && pod.Labels["version"] == "new" {
+									podsUpdated++
+								}
+							}
+							var pdb policyv1.PodDisruptionBudget
+							k8sClient.Get(ctx, types.NamespacedName{
+								Namespace: namespace,
+								Name:      cr.GetVMStorageName(),
+							}, &pdb)
+							Expect(pdb.Status.CurrentHealthy).To(BeNumerically(">=", 4), "at least 4 pods should be healthy during the update")
+							return podsUpdated
+						}, eventualStatefulsetAppReadyTimeout).Should(BeNumerically("==", 5))
+					},
+				},
+			),
+		)
+	})
 })
 
 func assertStrictSecurity(podSpec corev1.PodSpec) {