Repository: knative-sandbox/eventing-kafka
Pull request #380 (2021-02-11 14:31:03)

[mtsource] don't kill pods with placed vpods
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Limit scaling down the adapter to the last pod with placed "vpods" 
- Temporarily disable cancelling single-tenant adapters. See https://github.com/knative-sandbox/eventing-kafka/issues/382
- 

<!--
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.
-->

source e2e tests are passing ... locally. Let's see if Prow is happy.

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/380#" title="Author self-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/master/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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/380?src=pr&el=h1) Report
> Merging [#380](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/380?src=pr&el=desc) (e0b8627) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/5cab2571692609ab83d57d64cbbc5f1f5fe5484f?el=desc) (5cab257) will **increase** coverage by `0.14%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##           master     #380      +/-   ##
==========================================
+ Coverage   75.12%   75.27%   +0.14%     
==========================================
  Files         124      124              
  Lines        4792     4792              
==========================================
+ Hits         3600     3607       +7     
+ Misses        969      962       -7     
  Partials      223      223              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/380?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/380/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `71.81% <0.00%> (+6.36%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/380?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/380?src=pr&el=footer). Last update [5cab257...cdd0273](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/380?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

knative-prow-robot: @lionelvillard: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | cdd027373ff3e8ef45a4ad0d9dd08defe73cf367 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1359872594244800512) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/380). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow-robot: @lionelvillard: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 7c60a961e004521f0a19f31f4f9c8d6c1f6fd03a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1359889132796514304) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/380). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
matzew: I still see a failure on the MT source job
lionelvillard: I know :-) Not an easy fix. There are at least 2 other issues I'm currently working on.

knative-prow-robot: @lionelvillard: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 7c60a961e004521f0a19f31f4f9c8d6c1f6fd03a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1359889132796514304) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-unit-tests | b17bba523bfff1c9af12faa838eb7bff4cf9c4bc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-unit-tests/1360330539856302080) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/380). [Your PR dashboard](https://gubernator.knative.dev/pr/lionelvillard).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
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/autoscaler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360330539868884992/artifacts/line-cov.html#file0) | 55.3% | 50.0% | -5.3
[pkg/common/scheduler/statefulset/snapshot.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360330539868884992/artifacts/line-cov.html#file1) | 87.5% | 90.0% | 2.5
[pkg/source/adapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360330539868884992/artifacts/line-cov.html#file2) | 65.2% | 61.7% | -3.5
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360330539868884992/artifacts/line-cov.html#file3) | 69.6% | 68.6% | -1.0

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/autoscaler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360332550756634624/artifacts/line-cov.html#file0) | 55.3% | 50.0% | -5.3
[pkg/common/scheduler/statefulset/snapshot.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360332550756634624/artifacts/line-cov.html#file1) | 87.5% | 90.0% | 2.5
[pkg/source/adapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360332550756634624/artifacts/line-cov.html#file2) | 65.2% | 61.7% | -3.5
[pkg/source/mtadapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/380/pull-knative-sandbox-eventing-kafka-go-coverage/1360332550756634624/artifacts/line-cov.html#file3) | 69.6% | 68.6% | -1.0

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/380#" title="Author self-approved">lionelvillard</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/380#pullrequestreview-589880331" title="LGTM">steven0711dong</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/master/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/201-adapter-clusterrole.yaml b/config/source/multi/201-adapter-clusterrole.yaml
index dc3f471c6..43163dc3b 100644
--- a/config/source/multi/201-adapter-clusterrole.yaml
+++ b/config/source/multi/201-adapter-clusterrole.yaml
@@ -56,14 +56,27 @@ rules:
   - watch
 
 - apiGroups:
-    - ""
+  - apps
   resources:
-    - secrets
+  - deployments
   verbs:
-    - get
+  - get
+  - list
+  - watch
+  - create
+  - update
+  - patch
+  - delete
 
 - apiGroups:
-   - ""
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - get
+
+- apiGroups:
+  - ""
   resources:
   - events
   verbs:
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 597e9356c..47332745f 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -46,7 +46,8 @@ func (s *StatefulSetScheduler) autoscale(ctx context.Context) {
 				zap.Int32("free", s.freeCapacity(snapshot)),
 				zap.Int32("used", s.usedCapacity(snapshot)),
 				zap.Int32("pending", s.pendingVReplicas()),
-				zap.Int32("replicas", s.replicas))
+				zap.Int32("replicas", s.replicas),
+				zap.Int32("last ordinal", snapshot.lastOrdinal))
 
 			var ratio float64
 			if s.replicas == 0 {
@@ -72,13 +73,21 @@ func (s *StatefulSetScheduler) autoscale(ctx context.Context) {
 				}
 
 				// Desired ratio is 0.5 (TODO: configurable)
-				scale.Spec.Replicas = int32(math.Ceil(float64(s.usedCapacity(snapshot)+s.pendingVReplicas()) / (float64(s.capacity) * 0.5)))
+				new := int32(math.Ceil(float64(s.usedCapacity(snapshot)+s.pendingVReplicas()) / (float64(s.capacity) * 0.5)))
 
-				s.logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))
+				// Make sure not to scale down past the last pod with placed vpods
+				if new < snapshot.lastOrdinal+1 {
+					new = snapshot.lastOrdinal + 1
+				}
 
-				_, err = s.statefulSetClient.UpdateScale(ctx, s.statefulSetName, scale, metav1.UpdateOptions{})
-				if err != nil {
-					s.logger.Errorw("updating scale subresource failed", zap.Error(err))
+				if new != scale.Spec.Replicas {
+					scale.Spec.Replicas = new
+					s.logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))
+
+					_, err = s.statefulSetClient.UpdateScale(ctx, s.statefulSetName, scale, metav1.UpdateOptions{})
+					if err != nil {
+						s.logger.Errorw("updating scale subresource failed", zap.Error(err))
+					}
 				}
 			}
 		}()
diff --git a/pkg/common/scheduler/statefulset/snapshot.go b/pkg/common/scheduler/statefulset/snapshot.go
index a2891c1e3..58632ba4c 100644
--- a/pkg/common/scheduler/statefulset/snapshot.go
+++ b/pkg/common/scheduler/statefulset/snapshot.go
@@ -22,6 +22,10 @@ import "go.uber.org/zap"
 type Snapshot struct {
 	// free tracks the free capacity of each pod.
 	free map[string]int32
+
+	// lastOrdinal is the ordinal index corresponding to the last statefulset replica
+	// with placed vpods.
+	lastOrdinal int32
 }
 
 func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, error) {
@@ -36,6 +40,8 @@ func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, e
 		free[podName] = s.capacity
 	}
 
+	last := int32(0)
+
 	for _, vpod := range vpods {
 		ps := vpod.GetPlacements()
 
@@ -49,7 +55,12 @@ func (s *StatefulSetScheduler) snapshot(logger *zap.SugaredLogger) (*Snapshot, e
 				// Over committed. The autoscaler will fix it.
 				logger.Infow("pod is overcommitted", zap.String("podName", podName))
 			}
+
+			ordinal := ordinalFromPodName(podName)
+			if ordinal > last {
+				last = ordinal
+			}
 		}
 	}
-	return &Snapshot{free: free}, nil
+	return &Snapshot{free: free, lastOrdinal: last}, nil
 }
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 2dba25b4f..c9f2772fd 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -84,10 +84,6 @@ func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMe
 }
 
 func (a *Adapter) Start(ctx context.Context) error {
-	return a.start(ctx.Done())
-}
-
-func (a *Adapter) start(stopCh <-chan struct{}) error {
 	a.logger.Infow("Starting with config: ",
 		zap.String("Topics", strings.Join(a.config.Topics, ",")),
 		zap.String("ConsumerGroup", a.config.ConsumerGroup),
@@ -107,7 +103,12 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 	if err != nil {
 		panic(err)
 	}
-	defer func() { _ = group.Close() }()
+	defer func() {
+		err := group.Close()
+		if err != nil {
+			a.logger.Errorw("Failed to close consumer group", zap.Error(err))
+		}
+	}()
 
 	// Track errors
 	go func() {
@@ -116,7 +117,7 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 		}
 	}()
 
-	<-stopCh
+	<-ctx.Done()
 	a.logger.Info("Shutting down...")
 	return nil
 }
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index 6172caeb9..7ff5c824a 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -90,16 +90,19 @@ func (a *Adapter) Start(ctx context.Context) error {
 // Implements MTAdapter
 
 func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
+	a.sourcesMu.Lock()
+	defer a.sourcesMu.Unlock()
+
 	key := obj.Namespace + "/" + obj.Name
 
-	a.sourcesMu.RLock()
-	cancel, ok := a.sources[key]
-	a.sourcesMu.RUnlock()
+	_, ok := a.sources[key]
 
 	if ok {
+		// TODO: see https://github.com/knative-sandbox/eventing-kafka/issues/382
+		return
 		// TODO: do not stop if the only thing that changes is the number of vreplicas
-		a.logger.Info("stopping adapter", zap.String("key", key))
-		cancel()
+		//a.logger.Info("stopping adapter", zap.String("key", key))
+		//cancel()
 	}
 
 	placement := scheduler.GetPlacementForPod(obj.GetPlacements(), a.config.PodName)
@@ -166,17 +169,16 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 
 	}(ctx)
 
-	a.sourcesMu.Lock()
 	a.sources[key] = cancelFn
-	a.sourcesMu.Unlock()
 }
 
 func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
+	a.sourcesMu.Lock()
+	defer a.sourcesMu.Unlock()
+
 	key := obj.Namespace + "/" + obj.Name
 
-	a.sourcesMu.RLock()
 	cancel, ok := a.sources[key]
-	a.sourcesMu.RUnlock()
 
 	if !ok {
 		return
@@ -184,9 +186,7 @@ func (a *Adapter) Remove(ctx context.Context, obj *v1beta1.KafkaSource) {
 
 	cancel()
 
-	a.sourcesMu.Lock()
 	delete(a.sources, key)
-	a.sourcesMu.Unlock()
 }
 
 // ResolveSecret resolves the secret reference


Test patch:
diff --git a/pkg/common/scheduler/statefulset/snapshot_test.go b/pkg/common/scheduler/statefulset/snapshot_test.go
index f35ce2a82..73170a534 100644
--- a/pkg/common/scheduler/statefulset/snapshot_test.go
+++ b/pkg/common/scheduler/statefulset/snapshot_test.go
@@ -45,19 +45,19 @@ func TestSnapshot(t *testing.T) {
 			name:     "no vpods, no replicas",
 			vpods:    [][]duckv1alpha1.Placement{},
 			replicas: int32(0),
-			expected: Snapshot{free: map[string]int32{}},
+			expected: Snapshot{free: map[string]int32{}, lastOrdinal: 0},
 		},
 		{
 			name:     "no vpods, one replica",
 			vpods:    [][]duckv1alpha1.Placement{},
 			replicas: int32(1),
-			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(10)}},
+			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(10)}, lastOrdinal: 0},
 		},
 		{
 			name:     "one vpods, one replica",
 			vpods:    [][]duckv1alpha1.Placement{{{PodName: "statefulset-name-0", VReplicas: 1}}},
 			replicas: int32(1),
-			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(9)}},
+			expected: Snapshot{free: map[string]int32{"statefulset-name-0": int32(9)}, lastOrdinal: 0},
 		},
 		{
 			name: "many vpods, many replicas",
@@ -71,7 +71,7 @@ func TestSnapshot(t *testing.T) {
 				"statefulset-name-0": int32(8),
 				"statefulset-name-1": int32(5),
 				"statefulset-name-2": int32(5),
-				"statefulset-name-3": int32(10)}},
+				"statefulset-name-3": int32(10)}, lastOrdinal: 2},
 		},
 	}
 
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index f55a1ae35..b53337e00 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -456,7 +456,7 @@ function test_mt_source() {
   echo "Testing the multi-tenant source"
   install_mt_source || return 1
 
-  go_test_e2e -tags=source,mtsource -timeout=5m -test.parallel=${TEST_PARALLEL} ./test/...  || fail_test
+  go_test_e2e -tags=source,mtsource -timeout=20m -test.parallel=${TEST_PARALLEL} ./test/e2e/...  || fail_test
 
   uninstall_mt_source || return 1
 }