Repository: knative-sandbox/eventing-kafka
Pull request #468 (2021-03-16 13:27:48)

capacity and refreshPeriod are now configurable via env vars
Fixes https://github.com/knative-sandbox/eventing-kafka/issues/324

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add two new env variables on the mt source controller to configure pod capacity and autoscaler refresh period
- Change adapter resource request to accommodate 100 vreplicas consuming max 2MB (not enforced yet) 
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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/main/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":["lionelvillard"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/scheduler/statefulset/scheduler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/468/pull-knative-sandbox-eventing-kafka-go-coverage/1371815476820185088/artifacts/line-cov.html#file1) | 70.0% | 70.9% | 0.9

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468?src=pr&el=h1) Report
> Merging [#468](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468?src=pr&el=desc) (a1c43d4) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/5a79b3b13f13e2cbfdbda87144bf9ed6ed50b4e1?el=desc) (5a79b3b) will **decrease** coverage by `0.11%`.
> The diff coverage is `66.66%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             main     #468      +/-   ##
==========================================
- Coverage   73.78%   73.67%   -0.12%     
==========================================
  Files         138      138              
  Lines        5326     5326              
==========================================
- Hits         3930     3924       -6     
- Misses       1149     1155       +6     
  Partials      247      247              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/common/scheduler/statefulset/state.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvc3RhdGUuZ28=) | `85.29% <ø> (ø)` | |
| [pkg/common/scheduler/statefulset/scheduler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvc2NoZWR1bGVyLmdv) | `66.66% <50.00%> (+1.04%)` | :arrow_up: |
| [pkg/common/scheduler/statefulset/autoscaler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468/diff?src=pr&el=tree#diff-cGtnL2NvbW1vbi9zY2hlZHVsZXIvc3RhdGVmdWxzZXQvYXV0b3NjYWxlci5nbw==) | `73.07% <100.00%> (ø)` | |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/468/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `71.20% <0.00%> (-5.60%)` | :arrow_down: |

------

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

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/468#pullrequestreview-613547901" title="Approved">steven0711dong</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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/main/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":["lionelvillard"]} -->

Patch:
diff --git a/config/source/multi/deployments/adapter.yaml b/config/source/multi/deployments/adapter.yaml
index fd0c2d0e0..9f96e7767 100644
--- a/config/source/multi/deployments/adapter.yaml
+++ b/config/source/multi/deployments/adapter.yaml
@@ -20,7 +20,7 @@ metadata:
   labels:
     kafka.eventing.knative.dev/release: devel
 spec:
-  replicas: 1 # TODO: set to 0
+  replicas: 1
   selector:
     matchLabels: &labels
       control-plane: kafkasource-mt-adapter
@@ -56,11 +56,11 @@ spec:
 
         resources:
           requests:
-            cpu: 50m
-            memory: 1Gi
+            cpu: 100m
+            memory: 100Mi
           limits:
-            cpu: 50m
-            memory: 1Gi
+            cpu: 200m
+            memory: 150Mi
 
         ports:
         - name: metrics
diff --git a/config/source/multi/deployments/controller.yaml b/config/source/multi/deployments/controller.yaml
index bf3c60c09..8ef720d7a 100644
--- a/config/source/multi/deployments/controller.yaml
+++ b/config/source/multi/deployments/controller.yaml
@@ -44,7 +44,15 @@ spec:
           value: config-observability
         - name: CONFIG_LEADERELECTION_NAME
           value: config-leader-election
-        volumeMounts:
+
+        # how often (in seconds) the autoscaler tries to scale down the statefulset
+        - name: AUTOSCALER_REFRESH_PERIOD
+          value: '10'
+
+        # the number of virtual replicas this pod can handle
+        - name: POD_CAPACITY
+          value: '100'
+
         resources:
           requests:
             cpu: 20m
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 67ddc2304..4d6091df8 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -49,18 +49,23 @@ type autoscaler struct {
 	// capacity is the total number of virtual replicas available per pod.
 	capacity int32
 
-	// refreshPeriod is how often to try attempting scaling down the statefulset
+	// refreshPeriod is how often the autoscaler tries to scale down the statefulset
 	refreshPeriod time.Duration
 }
 
-func NewAutoscaler(ctx context.Context, namespace, name string, stateAccessor stateAccessor, refreshPeriod time.Duration) Autoscaler {
+func NewAutoscaler(ctx context.Context,
+	namespace, name string,
+	stateAccessor stateAccessor,
+	refreshPeriod time.Duration,
+	capacity int32) Autoscaler {
+
 	return &autoscaler{
 		logger:            logging.FromContext(ctx),
 		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(namespace),
 		statefulSetName:   name,
 		stateAccessor:     stateAccessor,
 		trigger:           make(chan int32, 1),
-		capacity:          int32(10),
+		capacity:          capacity,
 		refreshPeriod:     refreshPeriod,
 	}
 }
@@ -119,7 +124,7 @@ func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pen
 		// The number of replicas may be lower than the last ordinal, for instance
 		// when the statefulset is manually scaled down. In that case, replicas above
 		// scale.Spec.Replicas have not been considered when scheduling vreplicas.
-		// Adjust accordindly
+		// Adjust accordingly
 		pending -= state.freeCapacity()
 
 		// Still need more?
diff --git a/pkg/common/scheduler/statefulset/scheduler.go b/pkg/common/scheduler/statefulset/scheduler.go
index a05a22757..658e00bfe 100644
--- a/pkg/common/scheduler/statefulset/scheduler.go
+++ b/pkg/common/scheduler/statefulset/scheduler.go
@@ -38,11 +38,14 @@ import (
 )
 
 // NewScheduler creates a new scheduler with pod autoscaling enabled.
-func NewScheduler(ctx context.Context, namespace, name string, lister scheduler.VPodLister) scheduler.Scheduler {
-	capacity := int32(10) // TODO: config
+func NewScheduler(ctx context.Context,
+	namespace, name string,
+	lister scheduler.VPodLister,
+	refreshPeriod time.Duration,
+	capacity int32) scheduler.Scheduler {
 
 	stateAccessor := newStateBuilder(logging.FromContext(ctx), lister, capacity)
-	autoscaler := NewAutoscaler(ctx, namespace, name, stateAccessor, 10*time.Second) // TODO: config
+	autoscaler := NewAutoscaler(ctx, namespace, name, stateAccessor, refreshPeriod, capacity)
 
 	go autoscaler.Start(ctx)
 
@@ -67,7 +70,12 @@ type StatefulSetScheduler struct {
 	pending map[types.NamespacedName]int32
 }
 
-func NewStatefulSetScheduler(ctx context.Context, namespace, name string, lister scheduler.VPodLister, stateAccessor stateAccessor, autoscaler Autoscaler) scheduler.Scheduler {
+func NewStatefulSetScheduler(ctx context.Context,
+	namespace, name string,
+	lister scheduler.VPodLister,
+	stateAccessor stateAccessor,
+	autoscaler Autoscaler) scheduler.Scheduler {
+
 	scheduler := &StatefulSetScheduler{
 		logger:            logging.FromContext(ctx),
 		statefulSetName:   name,
diff --git a/pkg/common/scheduler/statefulset/state.go b/pkg/common/scheduler/statefulset/state.go
index 055011cc4..6343bd6a0 100644
--- a/pkg/common/scheduler/statefulset/state.go
+++ b/pkg/common/scheduler/statefulset/state.go
@@ -36,7 +36,7 @@ type state struct {
 	// with placed vpods.
 	lastOrdinal int32
 
-	// Pod capacity. Used
+	// Pod capacity.
 	capacity int32
 }
 
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
index bfabb631a..85025fccb 100644
--- a/pkg/source/reconciler/mtsource/controller.go
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -18,7 +18,9 @@ package mtsource
 
 import (
 	"context"
+	"time"
 
+	"github.com/kelseyhightower/envconfig"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -35,10 +37,22 @@ import (
 	scheduler "knative.dev/eventing-kafka/pkg/common/scheduler/statefulset"
 )
 
+type envConfig struct {
+	SchedulerRefreshPeriod int64 `envconfig:"AUTOSCALER_REFRESH_PERIOD" required:"true"`
+	PodCapacity            int32 `envconfig:"POD_CAPACITY" required:"true"`
+}
+
 func NewController(
 	ctx context.Context,
 	cmw configmap.Watcher,
 ) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	env := &envConfig{}
+	if err := envconfig.Process("", env); err != nil {
+		logger.Panicf("unable to process required environment variables: %v", err)
+	}
+
 	kafkaInformer := kafkainformer.Get(ctx)
 
 	c := &Reconciler{
@@ -55,10 +69,11 @@ func NewController(
 	// Use a different set of conditions
 	sourcesv1beta1.RegisterAlternateKafkaConditionSet(sourcesv1beta1.KafkaMTSourceCondSet)
 
-	c.scheduler = scheduler.NewScheduler(ctx, system.Namespace(), mtadapterName, c.vpodLister)
+	rp := time.Duration(env.SchedulerRefreshPeriod) * time.Second
 
-	logging.FromContext(ctx).Info("Setting up kafka event handlers")
+	c.scheduler = scheduler.NewScheduler(ctx, system.Namespace(), mtadapterName, c.vpodLister, rp, env.PodCapacity)
 
+	logging.FromContext(ctx).Info("Setting up kafka event handlers")
 	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	return impl


Test patch:
diff --git a/pkg/common/scheduler/statefulset/autoscaler_test.go b/pkg/common/scheduler/statefulset/autoscaler_test.go
index 0f1f3c9f0..1bd148681 100644
--- a/pkg/common/scheduler/statefulset/autoscaler_test.go
+++ b/pkg/common/scheduler/statefulset/autoscaler_test.go
@@ -196,7 +196,7 @@ func TestAutoscaler(t *testing.T) {
 				t.Fatal("unexpected error", err)
 			}
 
-			autoscaler := NewAutoscaler(ctx, testNs, sfsName, stateAccessor, 10*time.Second).(*autoscaler)
+			autoscaler := NewAutoscaler(ctx, testNs, sfsName, stateAccessor, 10*time.Second, int32(10)).(*autoscaler)
 
 			for _, vpod := range tc.vpods {
 				vpodClient.Append(vpod)
@@ -239,7 +239,7 @@ func TestAutoscalerScaleDownToZero(t *testing.T) {
 		t.Fatal("unexpected error", err)
 	}
 
-	autoscaler := NewAutoscaler(ctx, testNs, sfsName, stateAccessor, 2*time.Second).(*autoscaler)
+	autoscaler := NewAutoscaler(ctx, testNs, sfsName, stateAccessor, 2*time.Second, int32(10)).(*autoscaler)
 
 	done := make(chan bool)
 	go func() {