Repository: knative-sandbox/eventing-kafka
Pull request #535 (2021-04-13 15:40:05)

[0.22] 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 **travis-minke-sap** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @travis-minke-sap` 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.22/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":["travis-minke-sap"]} -->
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/535?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.22@314cc47`). [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/535/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/535?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.22     #535   +/-   ##
===============================================
  Coverage                ?   73.53%           
===============================================
  Files                   ?      147           
  Lines                   ?     5750           
  Branches                ?        0           
===============================================
  Hits                    ?     4228           
  Misses                  ?     1280           
  Partials                ?      242           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/535?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/535?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [314cc47...808d62f](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/535?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-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/535#pullrequestreview-634798670" 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.22/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 3d7666b0b..4937b5ded 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -164,7 +164,7 @@ func main() {
 	}
 
 	// Watch The Secret For Changes
-	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, secretObserver)
+	err = distributedcommonconfig.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 0db6c6e7f..d469f32fe 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -149,7 +149,7 @@ func main() {
 	}
 
 	// Watch The Secret For Changes
-	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, secretObserver)
+	err = distributedcommonconfig.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/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 1b0f54a13..172cf9b8e 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -406,6 +406,8 @@ func (d *DispatcherImpl) 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 {
 
@@ -416,12 +418,15 @@ func (d *DispatcherImpl) ObserveMetrics(interval time.Duration) {
 				close(d.MetricsStoppedChan)
 				return
 
-			case <-time.After(interval):
+			case <-metricsTimer.C:
 				// Get All The Sarama Metrics From The Producer's Metrics Registry
 				kafkaMetrics := d.MetricsRegistry.GetAll()
 
 				// Forward Metrics To Prometheus For Observation
 				d.StatsReporter.Report(kafkaMetrics)
+
+				// Schedule Another Report
+				metricsTimer.Reset(interval)
 			}
 		}
 	}()
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 1354d8f9a..71c271231 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -148,6 +148,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 {
 
@@ -158,12 +160,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 8673e5bab..01bebb3d7 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)