Repository: knative-sandbox/eventing-kafka
Pull request #860 (2021-09-08 18:00:29)

[release-0.25] control maximum allowed vreplica count
This is an automated cherry-pick of #822

```release-note
🎁 [Multi-tenant KafkaSource] Calculate an optimal maximum allowed vreplica number based on Kafka source partitions count so that scheduler does not schedule more vreplicas than the calculated number
```

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/860#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **rhuss** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @rhuss` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-0.25/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["rhuss"]} -->
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka%2Fpull%2F860) for more info**.

<!-- need_author_consent -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/860?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-0.25@ba92355`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/860/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/860?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@               Coverage Diff               @@
##             release-0.25     #860   +/-   ##
===============================================
  Coverage                ?   74.61%           
===============================================
  Files                   ?      139           
  Lines                   ?     6361           
  Branches                ?        0           
===============================================
  Hits                    ?     4746           
  Misses                  ?     1395           
  Partials                ?      220           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/860?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/860?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [ba92355...ce59e92](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/860?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

steven0711dong: @googlebot I consent.
steven0711dong: /lgtm
lionelvillard: /approve

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/860#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/860#issuecomment-915472341" title="Approved">lionelvillard</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-0.25/OWNERS)~~ [lionelvillard]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/config/source/multi/deployments/controller.yaml b/config/source/multi/deployments/controller.yaml
index e4bf9a5db..5540616fb 100644
--- a/config/source/multi/deployments/controller.yaml
+++ b/config/source/multi/deployments/controller.yaml
@@ -51,8 +51,13 @@ spec:
 
         # The number of virtual replicas each adapter pod can handle.
         - name: POD_CAPACITY
-          value: '100'
+          value: '20'
+
+        - name: MAX_MPS_PER_PARTITION
+          value: '15'
 
+        - name: VREPLICA_LIMITS_MPS
+          value: '250'
         # The scheduling policy type for placing vreplicas on pods (see type SchedulerPolicyType for enum list)
         - name: SCHEDULER_POLICY_TYPE
           value: 'MAXFILLUP'
diff --git a/pkg/apis/duck/v1alpha1/placement_types.go b/pkg/apis/duck/v1alpha1/placement_types.go
index b108211e2..8f8fe8fd3 100644
--- a/pkg/apis/duck/v1alpha1/placement_types.go
+++ b/pkg/apis/duck/v1alpha1/placement_types.go
@@ -27,7 +27,8 @@ import (
 // Placeable is a list of podName and virtual replicas pairs.
 // Each pair represents the assignment of virtual replicas to a pod
 type Placeable struct {
-	Placement []Placement `json:"placements,omitempty"`
+	MaxAllowedVReplicas *int32      `json:"maxAllowedVReplicas,omitempty"`
+	Placement           []Placement `json:"placements,omitempty"`
 }
 
 // PlaceableType is a skeleton type wrapping Placeable in the manner we expect
diff --git a/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
index a5b89327d..4a972bfc0 100644
--- a/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/duck/v1alpha1/zz_generated.deepcopy.go
@@ -27,6 +27,11 @@ import (
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Placeable) DeepCopyInto(out *Placeable) {
 	*out = *in
+	if in.MaxAllowedVReplicas != nil {
+		in, out := &in.MaxAllowedVReplicas, &out.MaxAllowedVReplicas
+		*out = new(int32)
+		**out = **in
+	}
 	if in.Placement != nil {
 		in, out := &in.Placement, &out.Placement
 		*out = make([]Placement, len(*in))
diff --git a/pkg/apis/sources/v1beta1/kafka_scheduling.go b/pkg/apis/sources/v1beta1/kafka_scheduling.go
index 293ecda6f..34696ff23 100644
--- a/pkg/apis/sources/v1beta1/kafka_scheduling.go
+++ b/pkg/apis/sources/v1beta1/kafka_scheduling.go
@@ -32,6 +32,11 @@ func (k *KafkaSource) GetVReplicas() int32 {
 	if k.Spec.Consumers == nil {
 		return 1
 	}
+	if k.Status.MaxAllowedVReplicas != nil {
+		if *k.Spec.Consumers > *k.Status.MaxAllowedVReplicas {
+			return *k.Status.MaxAllowedVReplicas
+		}
+	}
 	return *k.Spec.Consumers
 }
 
diff --git a/pkg/source/client/offsets.go b/pkg/source/client/offsets.go
index dac1574a0..93fc0972e 100644
--- a/pkg/source/client/offsets.go
+++ b/pkg/source/client/offsets.go
@@ -32,15 +32,16 @@ import (
 // is closed before at least one message is consumed from ALL partitions.
 // Without InitOffsets, an event sent to a partition with an uninitialized offset
 // will not be forwarded when the session is closed (or a rebalancing is in progress).
-func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string, consumerGroup string) error {
+func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string, consumerGroup string) (int32, error) {
+	totalPartitions := 0
 	offsetManager, err := sarama.NewOffsetManagerFromClient(consumerGroup, kafkaClient)
 	if err != nil {
-		return err
+		return -1, err
 	}
 
 	kafkaAdminClient, err := sarama.NewClusterAdminFromClient(kafkaClient)
 	if err != nil {
-		return fmt.Errorf("failed to create a Kafka admin client: %w", err)
+		return -1, fmt.Errorf("failed to create a Kafka admin client: %w", err)
 	}
 	defer kafkaAdminClient.Close()
 
@@ -48,9 +49,9 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 	topicPartitions := make(map[string][]int32)
 	for _, topic := range topics {
 		partitions, err := kafkaClient.Partitions(topic)
-
+		totalPartitions += len(partitions)
 		if err != nil {
-			return fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
+			return -1, fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
 		}
 
 		topicPartitions[topic] = partitions
@@ -59,13 +60,13 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 	// Fetch topic offsets
 	topicOffsets, err := knsarama.GetOffsets(kafkaClient, topicPartitions, sarama.OffsetNewest)
 	if err != nil {
-		return fmt.Errorf("failed to get the topic offsets: %w", err)
+		return -1, fmt.Errorf("failed to get the topic offsets: %w", err)
 	}
 
 	// Look for uninitialized offset (-1)
 	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
 	if err != nil {
-		return err
+		return -1, err
 	}
 
 	dirty := false
@@ -88,7 +89,7 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 
 				pm, err := offsetManager.ManagePartition(topic, partitionID)
 				if err != nil {
-					return fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
+					return -1, fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
 				}
 
 				pm.MarkOffset(offset, "")
@@ -103,6 +104,6 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 	}
 
 	// At this stage the KafkaSource instance is considered Ready
-	return nil
+	return int32(totalPartitions), nil
 
 }
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
index 68f170565..aebb817db 100644
--- a/pkg/source/reconciler/mtsource/controller.go
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -47,9 +47,11 @@ import (
 )
 
 type envConfig struct {
-	SchedulerRefreshPeriod int64                            `envconfig:"AUTOSCALER_REFRESH_PERIOD" required:"true"`
-	PodCapacity            int32                            `envconfig:"POD_CAPACITY" required:"true"`
-	SchedulerPolicy        stsscheduler.SchedulerPolicyType `envconfig:"SCHEDULER_POLICY_TYPE" required:"true"`
+	SchedulerRefreshPeriod        int64                            `envconfig:"AUTOSCALER_REFRESH_PERIOD" required:"true"`
+	PodCapacity                   int32                            `envconfig:"POD_CAPACITY" required:"true"`
+	VReplicaMPS                   int32                            `envconfig:"VREPLICA_LIMITS_MPS" required:"false" default:"-1"`
+	MaxEventPerSecondPerPartition int32                            `envconfig:"MAX_MPS_PER_PARTITION" required:"false" default:"-1"`
+	SchedulerPolicy               stsscheduler.SchedulerPolicyType `envconfig:"SCHEDULER_POLICY_TYPE" required:"true"`
 }
 
 func NewController(
@@ -67,10 +69,12 @@ func NewController(
 	nodeInformer := nodeinformer.Get(ctx)
 
 	c := &Reconciler{
-		KubeClientSet:  kubeclient.Get(ctx),
-		kafkaClientSet: kafkaclient.Get(ctx),
-		kafkaLister:    kafkaInformer.Lister(),
-		configs:        source.WatchConfigurations(ctx, component, cmw),
+		KubeClientSet:                 kubeclient.Get(ctx),
+		kafkaClientSet:                kafkaclient.Get(ctx),
+		kafkaLister:                   kafkaInformer.Lister(),
+		configs:                       source.WatchConfigurations(ctx, component, cmw),
+		VReplicaMPS:                   env.VReplicaMPS,
+		MaxEventPerSecondPerPartition: env.MaxEventPerSecondPerPartition,
 	}
 
 	impl := kafkasource.NewImpl(ctx, c)
diff --git a/pkg/source/reconciler/mtsource/kafkasource.go b/pkg/source/reconciler/mtsource/kafkasource.go
index f80ff386c..6b8b8072e 100644
--- a/pkg/source/reconciler/mtsource/kafkasource.go
+++ b/pkg/source/reconciler/mtsource/kafkasource.go
@@ -53,6 +53,9 @@ type Reconciler struct {
 	sinkResolver *resolver.URIResolver
 	configs      source.ConfigAccessor
 	scheduler    scheduler.Scheduler
+
+	VReplicaMPS                   int32
+	MaxEventPerSecondPerPartition int32
 }
 
 // Check that our Reconciler implements Interface
@@ -120,13 +123,17 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource
 	defer c.Close()
 	src.Status.MarkConnectionEstablished()
 
-	err = client.InitOffsets(ctx, c, src.Spec.Topics, src.Spec.ConsumerGroup)
+	totalPartitions, err := client.InitOffsets(ctx, c, src.Spec.Topics, src.Spec.ConsumerGroup)
 	if err != nil {
 		logging.FromContext(ctx).Errorw("unable to initialize consumergroup offsets", zap.Error(err))
 		src.Status.MarkInitialOffsetNotCommitted("OffsetsNotCommitted", "Unable to initialize consumergroup offsets: %v", err)
 		return err
 	}
 	src.Status.MarkInitialOffsetCommitted()
+	if r.MaxEventPerSecondPerPartition != -1 && r.VReplicaMPS != -1 {
+		maxVReplicas := totalPartitions*r.MaxEventPerSecondPerPartition/r.VReplicaMPS + 1
+		src.Status.MaxAllowedVReplicas = &maxVReplicas
+	}
 
 	// Finally, schedule the source
 	if err := r.reconcileMTReceiveAdapter(src); err != nil {
diff --git a/pkg/source/reconciler/source/kafkasource.go b/pkg/source/reconciler/source/kafkasource.go
index b31f2eb67..b7f0b25dc 100644
--- a/pkg/source/reconciler/source/kafkasource.go
+++ b/pkg/source/reconciler/source/kafkasource.go
@@ -186,7 +186,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource
 	defer c.Close()
 	src.Status.MarkConnectionEstablished()
 
-	err = client.InitOffsets(ctx, c, src.Spec.Topics, src.Spec.ConsumerGroup)
+	_, err = client.InitOffsets(ctx, c, src.Spec.Topics, src.Spec.ConsumerGroup)
 	if err != nil {
 		logging.FromContext(ctx).Errorw("unable to initialize consumergroup offsets", zap.Error(err))
 		src.Status.MarkInitialOffsetNotCommitted("OffsetsNotCommitted", "Unable to initialize consumergroup offsets: %v", err)


Test patch:
diff --git a/pkg/source/client/offsets_test.go b/pkg/source/client/offsets_test.go
index 269d8f9dc..77d5f4ec0 100644
--- a/pkg/source/client/offsets_test.go
+++ b/pkg/source/client/offsets_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
 
 	logtesting "knative.dev/pkg/logging/testing"
 )
@@ -138,8 +139,12 @@ func TestInitOffsets(t *testing.T) {
 			}
 			defer sc.Close()
 			ctx := logtesting.TestContextWithLogger(t)
-			err = InitOffsets(ctx, sc, tc.topics, group)
-
+			partitionCt, err := InitOffsets(ctx, sc, tc.topics, group)
+			total := 0
+			for _, partitions := range tc.topicOffsets {
+				total += len(partitions)
+			}
+			assert.Equal(t, int(partitionCt), total)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}