Repository: knative-sandbox/eventing-kafka
Pull request #456 (2021-03-11 14:27:58)

reduce apiserver requests for dispatcher
Signed-off-by: XinYang <xinydev@gmail.com>

The dispatcher now only gets information for the channel it cares about. reducing the GET requests to apiserver in a cluster with lots of channels.

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-  return before GET requests for channels dispatcher doesn’t need to care about

<!--
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 assign **slinkydeveloper** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @slinkydeveloper` 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/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":["slinkydeveloper"]} -->
knative-prow-robot: Hi @xinydev. 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>

knative-prow-robot: Welcome @xinydev! It looks like this is your first PR to knative-sandbox/eventing-kafka 🎉
lionelvillard: /ok-to-test
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/dispatcher/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/456/pull-knative-sandbox-eventing-kafka-go-coverage/1370022439764889600/artifacts/line-cov.html#file0) | 78.9% | 71.8% | -7.0

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/456?src=pr&el=h1) Report
> Merging [#456](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/456?src=pr&el=desc) (ee04c1d) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/9c479e092235c16fdc3dff45a890a9999ab00cee?el=desc) (9c479e0) will **decrease** coverage by `0.13%`.
> The diff coverage is `100.00%`.

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

```diff
@@            Coverage Diff             @@
##             main     #456      +/-   ##
==========================================
- Coverage   73.94%   73.80%   -0.14%     
==========================================
  Files         138      138              
  Lines        5315     5315              
==========================================
- Hits         3930     3923       -7     
- Misses       1140     1145       +5     
- Partials      245      247       +2     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/456?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [.../distributed/dispatcher/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/456/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9jb250cm9sbGVyL2thZmthY2hhbm5lbC5nbw==) | `64.36% <100.00%> (-8.05%)` | :arrow_down: |

------

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


Patch:
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index 877f48501..c5770a4b0 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -147,7 +147,7 @@ jobs:
           git check-attr --stdin linguist-generated | grep -Ev ': (set|true)$' | cut -d: -f1 |
           git check-attr --stdin linguist-vendored | grep -Ev ': (set|true)$' | cut -d: -f1 |
           grep -Ev '^(vendor/|third_party/|.git)' |
-          xargs misspell -error |
+          xargs misspell -i importas -error |
           reviewdog -efm="%f:%l:%c: %m" \
                 -name="github.com/client9/misspell" \
                 -reporter="github-pr-check" \
diff --git a/config/source/multi/resources/kafkasource.yaml b/config/source/multi/resources/kafkasource.yaml
index a0667e64b..8ed4c08fa 100644
--- a/config/source/multi/resources/kafkasource.yaml
+++ b/config/source/multi/resources/kafkasource.yaml
@@ -48,6 +48,8 @@ spec:
           specReplicasPath: .spec.consumers
           # statusReplicasPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Replicas.
           statusReplicasPath: .status.consumers
+          # labelSelectorPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Selector
+          labelSelectorPath: .status.selector
       additionalPrinterColumns:
         - name: Topics
           type: string
diff --git a/config/source/single/resources/kafkasource.yaml b/config/source/single/resources/kafkasource.yaml
index a0667e64b..8ed4c08fa 100644
--- a/config/source/single/resources/kafkasource.yaml
+++ b/config/source/single/resources/kafkasource.yaml
@@ -48,6 +48,8 @@ spec:
           specReplicasPath: .spec.consumers
           # statusReplicasPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Replicas.
           statusReplicasPath: .status.consumers
+          # labelSelectorPath defines the JSONPath inside of a custom resource that corresponds to Scale.Status.Selector
+          labelSelectorPath: .status.selector
       additionalPrinterColumns:
         - name: Topics
           type: string
diff --git a/pkg/apis/sources/v1alpha1/kafka_conversion.go b/pkg/apis/sources/v1alpha1/kafka_conversion.go
index 37787e770..72778a281 100644
--- a/pkg/apis/sources/v1alpha1/kafka_conversion.go
+++ b/pkg/apis/sources/v1alpha1/kafka_conversion.go
@@ -47,6 +47,7 @@ func (source *KafkaSource) ConvertTo(ctx context.Context, obj apis.Convertible)
 		}
 		source.Status.Status.DeepCopyInto(&sink.Status.Status)
 		sink.Status.Consumers = source.Status.Consumers
+		sink.Status.Selector = source.Status.Selector
 		source.Status.Placeable.DeepCopyInto(&sink.Status.Placeable)
 
 		// Optionals
@@ -95,6 +96,7 @@ func (sink *KafkaSource) ConvertFrom(ctx context.Context, obj apis.Convertible)
 
 		source.Status.Status.DeepCopyInto(&sink.Status.Status)
 		sink.Status.Consumers = source.Status.Consumers
+		sink.Status.Selector = source.Status.Selector
 		source.Status.Placeable.DeepCopyInto(&sink.Status.Placeable)
 
 		// Optionals
diff --git a/pkg/apis/sources/v1alpha1/kafka_types.go b/pkg/apis/sources/v1alpha1/kafka_types.go
index 3fdbc71f6..06bc3e33e 100644
--- a/pkg/apis/sources/v1alpha1/kafka_types.go
+++ b/pkg/apis/sources/v1alpha1/kafka_types.go
@@ -137,6 +137,10 @@ type KafkaSourceStatus struct {
 	// For round-tripping only.
 	Consumers int32 `json:"consumers,omitempty"`
 
+	// Use for labelSelectorPath when scaling Kafka source
+	// +optional
+	Selector string `json:"selector,omitempty"`
+
 	// Implement Placeable.
 	// For round-tripping only.
 	// +optional
diff --git a/pkg/apis/sources/v1beta1/kafka_types.go b/pkg/apis/sources/v1beta1/kafka_types.go
index 7d4329d50..70e9ce3b9 100644
--- a/pkg/apis/sources/v1beta1/kafka_types.go
+++ b/pkg/apis/sources/v1beta1/kafka_types.go
@@ -109,6 +109,10 @@ type KafkaSourceStatus struct {
 	// +optional
 	Consumers int32 `json:"consumers,omitempty"`
 
+	// Use for labelSelectorPath when scaling Kafka source
+	// +optional
+	Selector string `json:"selector,omitempty"`
+
 	// Implement Placeable.
 	// +optional
 	v1alpha1.Placeable `json:",inline"`
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
index 7bf77c9f4..93f7e8431 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -114,6 +114,11 @@ func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 
 	r.logger.Info("Reconcile", zap.String("key", key))
 
+	// Only Reconcile KafkaChannel Associated With This Dispatcher
+	if r.channelKey != key {
+		return nil
+	}
+
 	// Convert the namespace/name string into a distinct namespace and name.
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
@@ -132,11 +137,6 @@ func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 		return err
 	}
 
-	// Only Reconcile KafkaChannel Associated With This Dispatcher
-	if r.channelKey != key {
-		return nil
-	}
-
 	if !original.Status.IsReady() {
 		return fmt.Errorf("channel is not ready - cannot configure and update subscriber status")
 	}
diff --git a/pkg/source/reconciler/mtsource/kafkasource.go b/pkg/source/reconciler/mtsource/kafkasource.go
index 2f45715d7..172a181d6 100644
--- a/pkg/source/reconciler/mtsource/kafkasource.go
+++ b/pkg/source/reconciler/mtsource/kafkasource.go
@@ -79,6 +79,8 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource
 	}
 	src.Status.MarkSink(sinkURI)
 
+	src.Status.Selector = "control-plane=kafkasource-mt-adapter"
+
 	if val, ok := src.GetLabels()[v1beta1.KafkaKeyTypeLabel]; ok {
 		found := false
 		for _, allowed := range v1beta1.KafkaKeyTypeAllowed {
diff --git a/pkg/source/reconciler/source/kafkasource.go b/pkg/source/reconciler/source/kafkasource.go
index 2585caafc..04c79cac0 100644
--- a/pkg/source/reconciler/source/kafkasource.go
+++ b/pkg/source/reconciler/source/kafkasource.go
@@ -131,6 +131,12 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource
 	}
 	src.Status.MarkSink(sinkURI)
 
+	selector, err := resources.GetLabelsAsSelector(src.Name)
+	if err != nil {
+		return fmt.Errorf("getting labels as selector: %v", err)
+	}
+	src.Status.Selector = selector.String()
+
 	if val, ok := src.GetLabels()[v1beta1.KafkaKeyTypeLabel]; ok {
 		found := false
 		for _, allowed := range v1beta1.KafkaKeyTypeAllowed {
diff --git a/pkg/source/reconciler/source/resources/labels.go b/pkg/source/reconciler/source/resources/labels.go
index 048b91850..097b5a717 100644
--- a/pkg/source/reconciler/source/resources/labels.go
+++ b/pkg/source/reconciler/source/resources/labels.go
@@ -16,6 +16,11 @@ limitations under the License.
 
 package resources
 
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
 const (
 	// controllerAgentName is the string used by this controller to identify
 	// itself when creating events.
@@ -28,3 +33,12 @@ func GetLabels(name string) map[string]string {
 		"eventing.knative.dev/sourceName": name,
 	}
 }
+
+// GetLabelsAsSelector returns labels wrapped as Selector object
+func GetLabelsAsSelector(name string) (labels.Selector, error) {
+	labels := GetLabels(name)
+	var labelSelector metav1.LabelSelector
+	labelSelector.MatchLabels = labels
+	selector, err := metav1.LabelSelectorAsSelector(&labelSelector)
+	return selector, err
+}


Test patch:
diff --git a/pkg/source/reconciler/source/resources/labels_test.go b/pkg/source/reconciler/source/resources/labels_test.go
index d2d302878..c3dfe4b56 100644
--- a/pkg/source/reconciler/source/resources/labels_test.go
+++ b/pkg/source/reconciler/source/resources/labels_test.go
@@ -36,3 +36,18 @@ func TestGetLabels(t *testing.T) {
 		t.Fatalf("%v is not equal to %v", testLabels, wantLabels)
 	}
 }
+
+func TestGetLabelsAsSelector(t *testing.T) {
+
+	testLabels, err := GetLabelsAsSelector("testSourceName")
+	if err != nil {
+		t.Fatalf("Unable to get labels as selector")
+	}
+	testLabelsString := testLabels.String()
+
+	wantLabels := "eventing.knative.dev/source=kafka-source-controller,eventing.knative.dev/sourceName=testSourceName"
+
+	if testLabelsString != wantLabels {
+		t.Fatalf("%v is not equal to %v", testLabels, wantLabels)
+	}
+}