Repository: knative-sandbox/eventing-kafka (go)

Issue:
Unable to scale Kafka source with HPA
**Describe the bug**
Unable to scale single Kafka source with HPA

 **Expected behavior**
HAP should be able to scale Kafka source

 **To Reproduce**

Use HPA that is targeting Kafka source:

```
spec:
  scaleTargetRef:
    apiVersion: sources.knative.dev/v1beta1
    kind: KafkaSource
    name: kafka-src10
```

Problems:

```
 k describe hpa keda-hpa-so-96c230cf-052e-4243-a2d0-15fc80f8727f
Name:                                                      keda-hpa-so-96c230cf-052e-4243-a2d0-15fc80f8727f
Namespace:                                                 demo5
Labels:                                                    app.kubernetes.io/managed-by=keda-operator
                                                           app.kubernetes.io/name=keda-hpa-so-96c230cf-052e-4243-a2d0-15fc80f8727f
                                                           app.kubernetes.io/part-of=so-96c230cf-052e-4243-a2d0-15fc80f8727f
                                                           app.kubernetes.io/version=2.1.0
                                                           scaledObjectName=so-96c230cf-052e-4243-a2d0-15fc80f8727f
Annotations:                                               <none>
CreationTimestamp:                                         Tue, 09 Mar 2021 14:51:33 -0500
Reference:                                                 KafkaSource/kafka-src10
Metrics:                                                   ( current / target )
  "kafka-topic10-knative-group10" (target average value):  <unknown> / 2
Min replicas:                                              1
Max replicas:                                              50
KafkaSource pods:                                          2 current / 0 desired
Conditions:
  Type           Status  Reason             Message
  ----           ------  ------             -------
  AbleToScale    True    SucceededGetScale  the HPA controller was able to get the target's current scale
  ScalingActive  False   InvalidSelector    the HPA target's scale is missing a selector
Events:
  Type     Reason                        Age                  From                       Message
  ----     ------                        ----                 ----                       -------
  Warning  FailedComputeMetricsReplicas  20m (x41 over 30m)   horizontal-pod-autoscaler  selector is required
  Warning  SelectorRequired              37s (x121 over 30m)  horizontal-pod-autoscaler  selector is required
```

 **Knative release version**
0.21

 **Additional context**
Add any other context about the problem here such as proposed priority

Manual scaling of replcias works:
```
k scale kafkasources.sources.knative.dev kafka-src10 --replicas=2
kafkasource.sources.knative.dev/kafka-src10 scaled
```

Pull request:
Add label selector to enable HPA to scale Kafka source (seond attempt)
Trying to get CLA bot to work - ooriginal PR https://github.com/knative-sandbox/eventing-kafka/pull/453

Fixes #444

<!-- Please include the 'why' behind your changes if no issue exists -->
Fixes https://github.com/knative-sandbox/eventing-kafka/issues/444

## Proposed Changes

- add labelSelectorPath to CRDs
- add Status.Selector for all supported versions and conversions
- set Status.Selector with labels

<!--
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
- 🎁 HPA can be now used to scale Kafka source both single and multi-tenant. Details in https://github.com/knative-sandbox/eventing-kafka/issues/444
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
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/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)
+	}
+}