Add ShowPreciseParallelism filed (#607)

Co-authored-by: Rui Fu <[email protected]>
streamnative · Mar 15, 2023 · 405525e · 405525e
1 parent 5921b50
commit 405525e
Show file tree

Hide file tree

Showing 12 changed files with 50 additions and 4 deletions.
diff --git a/api/compute/v1alpha1/function_types.go b/api/compute/v1alpha1/function_types.go
@@ -37,6 +37,11 @@ type FunctionSpec struct {
 	ClusterName string `json:"clusterName,omitempty"`
 	// +kubebuilder:validation:Minimum=0
 	Replicas *int32 `json:"replicas,omitempty"`
+	// Whether show the precise parallelism, if true, the `Parallelism` will be equal to the `Replicas`,
+	// in such case, update the `Replicas` will cause all pods being recreated since the command of pod is updated.
+	// else, the `Parallelism` will be 1, default to false.
+	// It just affects the result of context.getNumInstances, there will be only 1 process and 1 thread in each pod in any cases.
+	ShowPreciseParallelism bool `json:"showPreciseParallelism,omitempty"`
 	// +kubebuilder:validation:Minimum=0
 	// +kubebuilder:default=1
 	MinReplicas *int32 `json:"minReplicas,omitempty"`

diff --git a/api/compute/v1alpha1/sink_types.go b/api/compute/v1alpha1/sink_types.go
@@ -38,6 +38,11 @@ type SinkSpec struct {
 	SinkType    string `json:"sinkType,omitempty"` // refer to `--sink-type` as builtin connector
 	// +kubebuilder:validation:Minimum=0
 	Replicas *int32 `json:"replicas,omitempty"`
+	// Whether show the precise parallelism, if true, the `Parallelism` will be equal to the `Replicas`,
+	// in such case, update the `Replicas` will cause all pods being recreated since the command of pod is updated.
+	// else, the `Parallelism` will be 1, default to false.
+	// It just affects the result of context.getNumInstances, there will be only 1 process and 1 thread in each pod in any cases.
+	ShowPreciseParallelism bool `json:"showPreciseParallelism,omitempty"`
 	// +kubebuilder:validation:Minimum=0
 	// +kubebuilder:default=1
 	MinReplicas *int32 `json:"minReplicas,omitempty"`

diff --git a/api/compute/v1alpha1/source_types.go b/api/compute/v1alpha1/source_types.go
@@ -45,6 +45,11 @@ type SourceSpec struct {
 	SourceType  string `json:"sourceType,omitempty"` // refer to `--source-type` as builtin connector
 	// +kubebuilder:validation:Minimum=0
 	Replicas *int32 `json:"replicas,omitempty"`
+	// Whether show the precise parallelism, if true, the `Parallelism` will be equal to the `Replicas`,
+	// in such case, update the `Replicas` will cause all pods being recreated since the command of pod is updated.
+	// else, the `Parallelism` will be 1, default to false.
+	// It just affects the result of context.getNumInstances, there will be only 1 process and 1 thread in each pod in any cases.
+	ShowPreciseParallelism bool `json:"showPreciseParallelism,omitempty"`
 	// +kubebuilder:validation:Minimum=0
 	// +kubebuilder:default=1
 	MinReplicas *int32 `json:"minReplicas,omitempty"`

diff --git a/...erator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functionmeshes.yaml b/...erator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functionmeshes.yaml
@@ -3251,6 +3251,8 @@ spec:
                               type: string
                           type: object
                         type: object
+                      showPreciseParallelism:
+                        type: boolean
                       statefulConfig:
                         properties:
                           pulsar:
@@ -6651,6 +6653,8 @@ spec:
                               type: string
                           type: object
                         type: object
+                      showPreciseParallelism:
+                        type: boolean
                       sinkConfig:
                         type: object
                         x-kubernetes-preserve-unknown-fields: true
@@ -9851,6 +9855,8 @@ spec:
                               type: string
                           type: object
                         type: object
+                      showPreciseParallelism:
+                        type: boolean
                       sourceConfig:
                         type: object
                         x-kubernetes-preserve-unknown-fields: true

diff --git a/...sh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functions.yaml b/...sh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-functions.yaml
@@ -3270,6 +3270,8 @@ spec:
                         type: string
                     type: object
                   type: object
+                showPreciseParallelism:
+                  type: boolean
                 statefulConfig:
                   properties:
                     pulsar:

diff --git a/...n-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sinks.yaml b/...n-mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sinks.yaml
@@ -3204,6 +3204,8 @@ spec:
                         type: string
                     type: object
                   type: object
+                showPreciseParallelism:
+                  type: boolean
                 sinkConfig:
                   type: object
                   x-kubernetes-preserve-unknown-fields: true

diff --git a/...mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sources.yaml b/...mesh-operator/charts/admission-webhook/templates/crd-compute.functionmesh.io-sources.yaml
@@ -3187,6 +3187,8 @@ spec:
                         type: string
                     type: object
                   type: object
+                showPreciseParallelism:
+                  type: boolean
                 sourceConfig:
                   type: object
                   x-kubernetes-preserve-unknown-fields: true

diff --git a/config/crd/bases/compute.functionmesh.io_functionmeshes.yaml b/config/crd/bases/compute.functionmesh.io_functionmeshes.yaml
@@ -3252,6 +3252,8 @@ spec:
                             type: string
                         type: object
                       type: object
+                    showPreciseParallelism:
+                      type: boolean
                     statefulConfig:
                       properties:
                         pulsar:
@@ -6652,6 +6654,8 @@ spec:
                             type: string
                         type: object
                       type: object
+                    showPreciseParallelism:
+                      type: boolean
                     sinkConfig:
                       type: object
                       x-kubernetes-preserve-unknown-fields: true
@@ -9852,6 +9856,8 @@ spec:
                             type: string
                         type: object
                       type: object
+                    showPreciseParallelism:
+                      type: boolean
                     sourceConfig:
                       type: object
                       x-kubernetes-preserve-unknown-fields: true

diff --git a/config/crd/bases/compute.functionmesh.io_functions.yaml b/config/crd/bases/compute.functionmesh.io_functions.yaml
@@ -3249,6 +3249,8 @@ spec:
                       type: string
                   type: object
                 type: object
+              showPreciseParallelism:
+                type: boolean
               statefulConfig:
                 properties:
                   pulsar:

diff --git a/config/crd/bases/compute.functionmesh.io_sinks.yaml b/config/crd/bases/compute.functionmesh.io_sinks.yaml
@@ -3183,6 +3183,8 @@ spec:
                       type: string
                   type: object
                 type: object
+              showPreciseParallelism:
+                type: boolean
               sinkConfig:
                 type: object
                 x-kubernetes-preserve-unknown-fields: true

diff --git a/config/crd/bases/compute.functionmesh.io_sources.yaml b/config/crd/bases/compute.functionmesh.io_sources.yaml
@@ -3166,6 +3166,8 @@ spec:
                       type: string
                   type: object
                 type: object
+              showPreciseParallelism:
+                type: boolean
               sourceConfig:
                 type: object
                 x-kubernetes-preserve-unknown-fields: true

diff --git a/controllers/spec/utils.go b/controllers/spec/utils.go
@@ -52,7 +52,7 @@ func convertFunctionDetails(function *v1alpha1.Function) *proto.FunctionDetails
 		UserConfig:           getUserConfig(generateFunctionConfig(function)),
 		Runtime:              runtime,
 		AutoAck:              getBoolFromPtrOrDefault(function.Spec.AutoAck, true),
-		Parallelism:          getInt32FromPtrOrDefault(function.Spec.Replicas, 1),
+		Parallelism:          getParallelism(function.Spec.Replicas, function.Spec.ShowPreciseParallelism),
 		Source:               generateFunctionInputSpec(function),
 		Sink:                 generateFunctionOutputSpec(function),
 		Resources:            generateResource(function.Spec.Resources.Requests),
@@ -110,7 +110,7 @@ func convertGoFunctionConfs(function *v1alpha1.Function) *GoFunctionConf {
 		//SecretsMap:                  marshalSecretsMap(function.Spec.SecretsMap),
 		Runtime:                     int32(proto.FunctionDetails_GO),
 		AutoACK:                     getBoolFromPtrOrDefault(function.Spec.AutoAck, true),
-		Parallelism:                 getInt32FromPtrOrDefault(function.Spec.Replicas, 1),
+		Parallelism:                 getParallelism(function.Spec.Replicas, function.Spec.ShowPreciseParallelism),
 		TimeoutMs:                   uint64(function.Spec.Timeout),
 		SubscriptionName:            function.Spec.SubscriptionName,
 		CleanupSubscription:         function.Spec.CleanupSubscription,
@@ -247,7 +247,7 @@ func convertSourceDetails(source *v1alpha1.Source) *proto.FunctionDetails {
 		UserConfig:           getUserConfig(source.Spec.SourceConfig),
 		Runtime:              proto.FunctionDetails_JAVA,
 		AutoAck:              true,
-		Parallelism:          getInt32FromPtrOrDefault(source.Spec.Replicas, 1),
+		Parallelism:          getParallelism(source.Spec.Replicas, source.Spec.ShowPreciseParallelism),
 		Source:               generateSourceInputSpec(source),
 		Sink:                 generateSourceOutputSpec(source),
 		Resources:            generateResource(source.Spec.Resources.Requests),
@@ -318,7 +318,7 @@ func convertSinkDetails(sink *v1alpha1.Sink) *proto.FunctionDetails {
 		ProcessingGuarantees: convertProcessingGuarantee(sink.Spec.ProcessingGuarantee),
 		Runtime:              proto.FunctionDetails_JAVA,
 		AutoAck:              getBoolFromPtrOrDefault(sink.Spec.AutoAck, true),
-		Parallelism:          getInt32FromPtrOrDefault(sink.Spec.Replicas, 1),
+		Parallelism:          getParallelism(sink.Spec.Replicas, sink.Spec.ShowPreciseParallelism),
 		Source:               generateSinkInputSpec(sink),
 		Sink:                 generateSinkOutputSpec(sink),
 		Resources:            generateResource(sink.Spec.Resources.Requests),
@@ -485,6 +485,13 @@ func getEnvOrDefault(key, fallback string) string {
 	return fallback
 }
 
+func getParallelism(replicas *int32, showPreciseParallelism bool) int32 {
+	if showPreciseParallelism {
+		return getInt32FromPtrOrDefault(replicas, 1)
+	}
+	return 1
+}
+
 func getDeadLetterTopicOrDefault(deadLetterTopic, subscriptionName, tenant, namespace, name string, maxMessageRetry int32) string {
 	if deadLetterTopic == "" && maxMessageRetry > 0 && (subscriptionName == "" || strings.Contains(subscriptionName, "\\")) {
 		// otherwise the auto generated DeadLetterTopic($TOPIC-$SUBNAME-DLQ) will be invalid