feat: Support setting ParallelPodManagement concurrency (#7879)

Signed-off-by: Liang Deng <283304489@qq.com>
apecloud · Jul 26, 2024 · daeb239 · daeb239
1 parent 220ede3
commit daeb239
Show file tree

Hide file tree

Showing 24 changed files with 412 additions and 83 deletions.
diff --git a/apis/apps/v1alpha1/cluster_types.go b/apis/apps/v1alpha1/cluster_types.go
@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
@@ -783,6 +784,13 @@ type ClusterComponentSpec struct {
 	// +optional
 	UpdateStrategy *UpdateStrategy `json:"updateStrategy,omitempty"`
 
+	// Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+	// or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+	// The default Concurrency is 100%.
+	//
+	// +optional
+	ParallelPodManagementConcurrency *intstr.IntOrString `json:"parallelPodManagementConcurrency,omitempty"`
+
 	// PodUpdatePolicy indicates how pods should be updated
 	//
 	// - `StrictInPlace` indicates that only allows in-place upgrades.

diff --git a/apis/apps/v1alpha1/component_types.go b/apis/apps/v1alpha1/component_types.go
@@ -19,6 +19,7 @@ package v1alpha1
 import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
 )
@@ -172,6 +173,13 @@ type ComponentSpec struct {
 	// +optional
 	ServiceAccountName string `json:"serviceAccountName,omitempty"`
 
+	// Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+	// or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+	// The default Concurrency is 100%.
+	//
+	// +optional
+	ParallelPodManagementConcurrency *intstr.IntOrString `json:"parallelPodManagementConcurrency,omitempty"`
+
 	// PodUpdatePolicy indicates how pods should be updated
 	//
 	// - `StrictInPlace` indicates that only allows in-place upgrades.

diff --git a/apis/apps/v1alpha1/zz_generated.deepcopy.go b/apis/apps/v1alpha1/zz_generated.deepcopy.go
diff --git a/apis/workloads/v1alpha1/instanceset_types.go b/apis/workloads/v1alpha1/instanceset_types.go
@@ -20,6 +20,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
@@ -304,6 +305,13 @@ type InstanceSetSpec struct {
 	// +optional
 	PodManagementPolicy appsv1.PodManagementPolicyType `json:"podManagementPolicy,omitempty"`
 
+	// Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+	// or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+	// The default Concurrency is 100%.
+	//
+	// +optional
+	ParallelPodManagementConcurrency *intstr.IntOrString `json:"parallelPodManagementConcurrency,omitempty"`
+
 	// PodUpdatePolicy indicates how pods should be updated
 	//
 	// - `StrictInPlace` indicates that only allows in-place upgrades.

diff --git a/apis/workloads/v1alpha1/zz_generated.deepcopy.go b/apis/workloads/v1alpha1/zz_generated.deepcopy.go
diff --git a/config/crd/bases/apps.kubeblocks.io_clusters.yaml b/config/crd/bases/apps.kubeblocks.io_clusters.yaml
@@ -3808,6 +3808,15 @@ spec:
                       items:
                         type: string
                       type: array
+                    parallelPodManagementConcurrency:
+                      anyOf:
+                      - type: integer
+                      - type: string
+                      description: |-
+                        Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                        or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                        The default Concurrency is 100%.
+                      x-kubernetes-int-or-string: true
                     podUpdatePolicy:
                       description: |-
                         PodUpdatePolicy indicates how pods should be updated
@@ -10366,6 +10375,15 @@ spec:
                           items:
                             type: string
                           type: array
+                        parallelPodManagementConcurrency:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          description: |-
+                            Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                            or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                            The default Concurrency is 100%.
+                          x-kubernetes-int-or-string: true
                         podUpdatePolicy:
                           description: |-
                             PodUpdatePolicy indicates how pods should be updated

diff --git a/config/crd/bases/apps.kubeblocks.io_components.yaml b/config/crd/bases/apps.kubeblocks.io_components.yaml
@@ -3615,6 +3615,15 @@ spec:
                 items:
                   type: string
                 type: array
+              parallelPodManagementConcurrency:
+                anyOf:
+                - type: integer
+                - type: string
+                description: |-
+                  Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                  or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                  The default Concurrency is 100%.
+                x-kubernetes-int-or-string: true
               podUpdatePolicy:
                 description: |-
                   PodUpdatePolicy indicates how pods should be updated

diff --git a/config/crd/bases/workloads.kubeblocks.io_instancesets.yaml b/config/crd/bases/workloads.kubeblocks.io_instancesets.yaml
@@ -3899,6 +3899,15 @@ spec:
                 items:
                   type: string
                 type: array
+              parallelPodManagementConcurrency:
+                anyOf:
+                - type: integer
+                - type: string
+                description: |-
+                  Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                  or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                  The default Concurrency is 100%.
+                x-kubernetes-int-or-string: true
               paused:
                 description: Indicates that the InstanceSet is paused, meaning the
                   reconciliation of this InstanceSet object will be paused.

diff --git a/controllers/apps/transformer_cluster_component.go b/controllers/apps/transformer_cluster_component.go
@@ -219,6 +219,7 @@ func copyAndMergeComponent(oldCompObj, newCompObj *appsv1alpha1.Component) *apps
 	// compObjCopy.Spec.Monitor = compProto.Spec.Monitor
 	compObjCopy.Spec.EnabledLogs = compProto.Spec.EnabledLogs
 	compObjCopy.Spec.ServiceAccountName = compProto.Spec.ServiceAccountName
+	compObjCopy.Spec.ParallelPodManagementConcurrency = compProto.Spec.ParallelPodManagementConcurrency
 	compObjCopy.Spec.PodUpdatePolicy = compProto.Spec.PodUpdatePolicy
 	compObjCopy.Spec.Affinity = compProto.Spec.Affinity
 	compObjCopy.Spec.Tolerations = compProto.Spec.Tolerations

diff --git a/controllers/apps/transformer_component_workload.go b/controllers/apps/transformer_component_workload.go
@@ -334,6 +334,7 @@ func copyAndMergeITS(oldITS, newITS *workloads.InstanceSet, synthesizeComp *comp
 	itsObjCopy.Spec.OfflineInstances = itsProto.Spec.OfflineInstances
 	itsObjCopy.Spec.MinReadySeconds = itsProto.Spec.MinReadySeconds
 	itsObjCopy.Spec.VolumeClaimTemplates = itsProto.Spec.VolumeClaimTemplates
+	itsObjCopy.Spec.ParallelPodManagementConcurrency = itsProto.Spec.ParallelPodManagementConcurrency
 
 	if itsProto.Spec.UpdateStrategy.Type != "" || itsProto.Spec.UpdateStrategy.RollingUpdate != nil {
 		updateUpdateStrategy(itsObjCopy, itsProto)

diff --git a/deploy/helm/crds/apps.kubeblocks.io_clusters.yaml b/deploy/helm/crds/apps.kubeblocks.io_clusters.yaml
@@ -3808,6 +3808,15 @@ spec:
                       items:
                         type: string
                       type: array
+                    parallelPodManagementConcurrency:
+                      anyOf:
+                      - type: integer
+                      - type: string
+                      description: |-
+                        Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                        or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                        The default Concurrency is 100%.
+                      x-kubernetes-int-or-string: true
                     podUpdatePolicy:
                       description: |-
                         PodUpdatePolicy indicates how pods should be updated
@@ -10366,6 +10375,15 @@ spec:
                           items:
                             type: string
                           type: array
+                        parallelPodManagementConcurrency:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          description: |-
+                            Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                            or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                            The default Concurrency is 100%.
+                          x-kubernetes-int-or-string: true
                         podUpdatePolicy:
                           description: |-
                             PodUpdatePolicy indicates how pods should be updated

diff --git a/deploy/helm/crds/apps.kubeblocks.io_components.yaml b/deploy/helm/crds/apps.kubeblocks.io_components.yaml
@@ -3615,6 +3615,15 @@ spec:
                 items:
                   type: string
                 type: array
+              parallelPodManagementConcurrency:
+                anyOf:
+                - type: integer
+                - type: string
+                description: |-
+                  Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                  or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                  The default Concurrency is 100%.
+                x-kubernetes-int-or-string: true
               podUpdatePolicy:
                 description: |-
                   PodUpdatePolicy indicates how pods should be updated

diff --git a/deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml b/deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml
@@ -3899,6 +3899,15 @@ spec:
                 items:
                   type: string
                 type: array
+              parallelPodManagementConcurrency:
+                anyOf:
+                - type: integer
+                - type: string
+                description: |-
+                  Controls the concurrency of pods during initial scale up, when replacing pods on nodes,
+                  or when scaling down. It only used when `PodManagementPolicy` is set to `Parallel`.
+                  The default Concurrency is 100%.
+                x-kubernetes-int-or-string: true
               paused:
                 description: Indicates that the InstanceSet is paused, meaning the
                   reconciliation of this InstanceSet object will be paused.

diff --git a/pkg/controller/builder/builder_component.go b/pkg/controller/builder/builder_component.go
@@ -21,6 +21,7 @@ package builder
 
 import (
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 
 	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
@@ -91,6 +92,11 @@ func (builder *ComponentBuilder) SetServiceAccountName(serviceAccountName string
 	return builder
 }
 
+func (builder *ComponentBuilder) SetParallelPodManagementConcurrency(parallelPodManagementConcurrency *intstr.IntOrString) *ComponentBuilder {
+	builder.get().Spec.ParallelPodManagementConcurrency = parallelPodManagementConcurrency
+	return builder
+}
+
 func (builder *ComponentBuilder) SetPodUpdatePolicy(policy *workloads.PodUpdatePolicyType) *ComponentBuilder {
 	builder.get().Spec.PodUpdatePolicy = policy
 	return builder

diff --git a/pkg/controller/builder/builder_instance_set.go b/pkg/controller/builder/builder_instance_set.go
@@ -23,6 +23,7 @@ import (
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
 )
@@ -107,6 +108,11 @@ func (builder *InstanceSetBuilder) SetPodManagementPolicy(policy apps.PodManagem
 	return builder
 }
 
+func (builder *InstanceSetBuilder) SetParallelPodManagementConcurrency(parallelPodManagementConcurrency *intstr.IntOrString) *InstanceSetBuilder {
+	builder.get().Spec.ParallelPodManagementConcurrency = parallelPodManagementConcurrency
+	return builder
+}
+
 func (builder *InstanceSetBuilder) SetPodUpdatePolicy(policy workloads.PodUpdatePolicyType) *InstanceSetBuilder {
 	builder.get().Spec.PodUpdatePolicy = policy
 	return builder

diff --git a/pkg/controller/builder/builder_instance_set_test.go b/pkg/controller/builder/builder_instance_set_test.go
@@ -47,6 +47,7 @@ var _ = Describe("instance_set builder", func() {
 			policy                       = apps.OrderedReadyPodManagement
 			podUpdatePolicy              = workloads.PreferInPlacePodUpdatePolicyType
 		)
+		parallelPodManagementConcurrency := &intstr.IntOrString{Type: intstr.String, StrVal: "100%"}
 		selectors := map[string]string{selectorKey4: selectorValue4}
 		role := workloads.ReplicaRole{
 			Name:       "foo",
@@ -166,6 +167,7 @@ var _ = Describe("instance_set builder", func() {
 			SetVolumeClaimTemplates(vcs...).
 			AddVolumeClaimTemplates(vc).
 			SetPodManagementPolicy(policy).
+			SetParallelPodManagementConcurrency(parallelPodManagementConcurrency).
 			SetPodUpdatePolicy(podUpdatePolicy).
 			SetUpdateStrategy(strategy).
 			SetUpdateStrategyType(strategyType).
@@ -198,6 +200,7 @@ var _ = Describe("instance_set builder", func() {
 		Expect(its.Spec.VolumeClaimTemplates[0]).Should(Equal(vcs[0]))
 		Expect(its.Spec.VolumeClaimTemplates[1]).Should(Equal(vc))
 		Expect(its.Spec.PodManagementPolicy).Should(Equal(policy))
+		Expect(its.Spec.ParallelPodManagementConcurrency).Should(Equal(parallelPodManagementConcurrency))
 		Expect(its.Spec.PodUpdatePolicy).Should(Equal(podUpdatePolicy))
 		Expect(its.Spec.UpdateStrategy.Type).Should(Equal(strategyType))
 		Expect(its.Spec.UpdateStrategy.RollingUpdate).ShouldNot(BeNil())

diff --git a/pkg/controller/component/component.go b/pkg/controller/component/component.go
@@ -93,6 +93,7 @@ func BuildComponent(cluster *appsv1alpha1.Cluster, compSpec *appsv1alpha1.Cluste
 		SetReplicas(compSpec.Replicas).
 		SetResources(compSpec.Resources).
 		SetServiceAccountName(compSpec.ServiceAccountName).
+		SetParallelPodManagementConcurrency(compSpec.ParallelPodManagementConcurrency).
 		SetPodUpdatePolicy(compSpec.PodUpdatePolicy).
 		SetVolumeClaimTemplates(compSpec.VolumeClaimTemplates).
 		SetVolumes(compSpec.Volumes).

diff --git a/pkg/controller/component/rsm_convertor.go b/pkg/controller/component/rsm_convertor.go
@@ -25,6 +25,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 
 	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
@@ -40,18 +41,19 @@ func BuildWorkloadFrom(synthesizeComp *SynthesizedComponent, protoITS *workloads
 		protoITS = &workloads.InstanceSet{}
 	}
 	convertors := map[string]convertor{
-		"service":                   &itsServiceConvertor{},
-		"alternativeservices":       &itsAlternativeServicesConvertor{},
-		"roles":                     &itsRolesConvertor{},
-		"roleprobe":                 &itsRoleProbeConvertor{},
-		"credential":                &itsCredentialConvertor{},
-		"membershipreconfiguration": &itsMembershipReconfigurationConvertor{},
-		"memberupdatestrategy":      &itsMemberUpdateStrategyConvertor{},
-		"podmanagementpolicy":       &itsPodManagementPolicyConvertor{},
-		"podupdatepolicy":           &itsPodUpdatePolicyConvertor{},
-		"updatestrategy":            &itsUpdateStrategyConvertor{},
-		"instances":                 &itsInstancesConvertor{},
-		"offlineinstances":          &itsOfflineInstancesConvertor{},
+		"service":                          &itsServiceConvertor{},
+		"alternativeservices":              &itsAlternativeServicesConvertor{},
+		"roles":                            &itsRolesConvertor{},
+		"roleprobe":                        &itsRoleProbeConvertor{},
+		"credential":                       &itsCredentialConvertor{},
+		"membershipreconfiguration":        &itsMembershipReconfigurationConvertor{},
+		"memberupdatestrategy":             &itsMemberUpdateStrategyConvertor{},
+		"podmanagementpolicy":              &itsPodManagementPolicyConvertor{},
+		"parallelpodmanagementconcurrency": &itsParallelPodManagementConcurrencyConvertor{},
+		"podupdatepolicy":                  &itsPodUpdatePolicyConvertor{},
+		"updatestrategy":                   &itsUpdateStrategyConvertor{},
+		"instances":                        &itsInstancesConvertor{},
+		"offlineinstances":                 &itsOfflineInstancesConvertor{},
 	}
 	if err := covertObject(convertors, &protoITS.Spec, synthesizeComp); err != nil {
 		return nil, err
@@ -106,6 +108,20 @@ func (c *itsPodManagementPolicyConvertor) convert(args ...any) (any, error) {
 	return appsv1.ParallelPodManagement, nil
 }
 
+// itsParallelPodManagementConcurrencyConvertor is an implementation of the convertor interface, used to convert the given object into InstanceSet.Spec.ParallelPodManagementConcurrency.
+type itsParallelPodManagementConcurrencyConvertor struct{}
+
+func (c *itsParallelPodManagementConcurrencyConvertor) convert(args ...any) (any, error) {
+	synthesizedComp, err := parseITSConvertorArgs(args...)
+	if err != nil {
+		return nil, err
+	}
+	if synthesizedComp.ParallelPodManagementConcurrency != nil {
+		return synthesizedComp.ParallelPodManagementConcurrency, nil
+	}
+	return &intstr.IntOrString{Type: intstr.String, StrVal: "100%"}, nil
+}
+
 // itsPodUpdatePolicyConvertor is an implementation of the convertor interface, used to convert the given object into InstanceSet.Spec.PodUpdatePolicy.
 type itsPodUpdatePolicyConvertor struct{}