Repository: knative-sandbox/eventing-kafka
Pull request #534 (2021-04-13 15:20:28)

[0.21] Backport of #525
Backport of #525 (Runaway CPU due to timed-out secret watcher)


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 assign **lberk** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @lberk` 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.21/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":["lberk"]} -->
knative-prow-robot: Hi @eric-sap. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<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.
</details>

travis-minke-sap: /ok-to-test
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/534?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.21@81cfdd5`). [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/534/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/534?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.21     #534   +/-   ##
===============================================
  Coverage                ?   73.12%           
===============================================
  Files                   ?      131           
  Lines                   ?     5164           
  Branches                ?        0           
===============================================
  Hits                    ?     3776           
  Misses                  ?     1157           
  Partials                ?      231           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/534?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/534?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [81cfdd5...25731dc](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/534?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).

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/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/534/pull-knative-sandbox-eventing-kafka-go-coverage/1381991385862770688/artifacts/line-cov.html#file1) | 77.6% | 77.3% | -0.2

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/534#pullrequestreview-634799136" title="Approved">travis-minke-sap</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.21/OWNERS)~~ [travis-minke-sap]

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/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index b5543eba2..6df7993ab 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -150,7 +150,7 @@ func main() {
 	}
 
 	// Watch The Secret For Changes
-	err = commonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, secretObserver)
+	err = commonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
 	if err != nil {
 		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
 	}
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 262bf0159..7779f68b4 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -140,7 +140,7 @@ func main() {
 	}
 
 	// Watch The Secret For Changes
-	err = commonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, secretObserver)
+	err = commonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
 	if err != nil {
 		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
 	}
diff --git a/pkg/channel/distributed/common/config/secretwatcher.go b/pkg/channel/distributed/common/config/secretwatcher.go
index 29e32af47..b03d9c93e 100644
--- a/pkg/channel/distributed/common/config/secretwatcher.go
+++ b/pkg/channel/distributed/common/config/secretwatcher.go
@@ -20,54 +20,37 @@ import (
 	"context"
 	"time"
 
-	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/informers"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
 
-	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/watch"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 	"knative.dev/eventing-kafka/pkg/common/client"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/logging"
 )
 
 type SecretObserver func(ctx context.Context, secret *corev1.Secret)
 
 //
-// Initialize The Specified Context With A Secret Watcher
+// Initialize The Specified Context With A Secret Informer
 //
-func InitializeSecretWatcher(ctx context.Context, namespace string, name string, observer SecretObserver) error {
+func InitializeSecretWatcher(ctx context.Context, namespace string, name string, resyncTime time.Duration, observer SecretObserver) error {
 
 	logger := logging.FromContext(ctx)
-	secrets := kubeclient.Get(ctx).CoreV1().Secrets(namespace)
-	watcher, err := secrets.Watch(ctx, metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("metadata.name", name).String()})
-	if err != nil {
-		logger.Error("Failed to start secret watcher", zap.Error(err))
-	}
-
-	go func() {
-		defer watcher.Stop()
-
-		for {
-			select {
-			case <-ctx.Done():
-				logger.Info("Stopped Secret Watcher")
-				return
-			case event := <-watcher.ResultChan():
-				// We only care if the secret was modified or added
-				if event.Type == watch.Added || event.Type == watch.Modified {
-					if secret, ok := event.Object.(*corev1.Secret); ok {
-						observer(ctx, secret)
-					} else {
-						logger.Error("Could not convert watched object to Secret", zap.Any("event.Object", event.Object))
-					}
-				}
-			case <-time.After(5 * time.Second):
-				// Don't hang a shutdown waiting for an event
-			}
+	secretsInformer := informers.NewSharedInformerFactoryWithOptions(
+		kubeclient.Get(ctx), resyncTime, informers.WithNamespace(namespace)).Core().V1().Secrets().Informer()
+	secretsInformer.AddEventHandler(controller.HandleAll(func(object interface{}) {
+		secret, ok := object.(*corev1.Secret)
+		if ok && secret.Name == name {
+			observer(ctx, secret)
 		}
+	}))
 
+	go func() {
+		secretsInformer.Run(ctx.Done())
+		logger.Info("Stopped Secret Watcher")
 	}()
 
 	return nil
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 6b9ffe764..5bfde842b 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -143,6 +143,8 @@ func (p *Producer) ObserveMetrics(interval time.Duration) {
 	// Fork A New Process To Run Async Metrics Collection
 	go func() {
 
+		metricsTimer := time.NewTimer(interval)
+
 		// Infinite Loop For Periodically Observing Sarama Metrics From Registry
 		for {
 
@@ -153,12 +155,15 @@ func (p *Producer) ObserveMetrics(interval time.Duration) {
 				close(p.metricsStoppedChan)
 				return
 
-			case <-time.After(interval):
+			case <-metricsTimer.C:
 				// Get All The Sarama Metrics From The Producer's Metrics Registry
 				kafkaMetrics := p.metricsRegistry.GetAll()
 
 				// Forward Metrics To Prometheus For Observation
 				p.statsReporter.Report(kafkaMetrics)
+
+				// Schedule Another Report
+				metricsTimer.Reset(interval)
 			}
 		}
 	}()


Test patch:
diff --git a/pkg/channel/distributed/common/config/secretwatcher_test.go b/pkg/channel/distributed/common/config/secretwatcher_test.go
index 9794c4e12..a936b07c3 100644
--- a/pkg/channel/distributed/common/config/secretwatcher_test.go
+++ b/pkg/channel/distributed/common/config/secretwatcher_test.go
@@ -133,7 +133,7 @@ func TestInitializeSecretWatcher(t *testing.T) {
 	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeySaslType]), commontesting.OldAuthSaslType)
 
 	// Perform The Test (Initialize The Secret Watcher)
-	err = InitializeSecretWatcher(ctx, system.Namespace(), constants.SettingsSecretName, secretWatcherHandler)
+	err = InitializeSecretWatcher(ctx, system.Namespace(), constants.SettingsSecretName, 10*time.Second, secretWatcherHandler)
 	assert.Nil(t, err)
 
 	// The secretWatcherHandler should change this back to a valid Secret after the watcher is triggered
@@ -150,10 +150,9 @@ func TestInitializeSecretWatcher(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyPassword]), commontesting.NewAuthPassword)
 
-	// Wait for the secretWatcherHandler to be called (happens pretty quickly; loop usually only runs once)
-	for try := 0; getWatchedSecret() == nil && try < 100; try++ {
-		time.Sleep(5 * time.Millisecond)
-	}
+	// Wait for the secretWatcherHandler to be called
+	assert.Eventually(t, func() bool { return getWatchedSecret() != nil }, 500*time.Millisecond, 5*time.Millisecond)
+
 	assert.NotNil(t, getWatchedSecret())
 	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyUsername]), commontesting.NewAuthUsername)
 	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyPassword]), commontesting.NewAuthPassword)