Repository: knative-sandbox/eventing-kafka-broker
Pull request #2632 (2022-09-09 08:45:02)

[release-1.5] Manual backport of proper webhook implementation
This is an automated cherry-pick of #2626

```release-note

```

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2632#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign gab-satchi for approval by writing `/assign @gab-satchi` in a comment. For more information see:[The Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.5/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":["gab-satchi"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2632?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2632](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2632?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0725b2c) into [release-1.5](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/b09bd3655ad853544cd9ce6d4544c937e1f3c9fc?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b09bd36) will **decrease** coverage by `0.09%`.
> The diff coverage is `55.17%`.

```diff
@@               Coverage Diff               @@
##           release-1.5    #2632      +/-   ##
===============================================
- Coverage        60.13%   60.03%   -0.10%     
===============================================
  Files               71       71              
  Lines             7171     7186      +15     
===============================================
+ Hits              4312     4314       +2     
- Misses            2543     2556      +13     
  Partials           316      316              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2632?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ol-plane/pkg/apis/eventing/v1/broker\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2632/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MS9icm9rZXJfdmFsaWRhdGlvbi5nbw==) | `55.88% <55.17%> (-17.81%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2632/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `70.91% <0.00%> (-1.20%)` | :arrow_down: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

aliok: /lgtm
/approve
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2632#issuecomment-1241691923" title="Approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2632#" title="Author self-approved">knative-prow-robot</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-broker).

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-broker/blob/release-1.5/OWNERS)~~ [aliok]

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":[]} -->
knative-prow[bot]: @knative-prow-robot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.5 | 0725b2c619c2ff0035c370eeef0209f7830647b0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2632/reconciler-tests_eventing-kafka-broker_release-1.5/1568158535345246208) | true | `/test reconciler-tests_eventing-kafka-broker_release-1.5`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
aliok: /test reconciler-tests_eventing-kafka-broker_release-1.5

Patch:
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 817f85797..fe73316c1 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -32,7 +32,7 @@ import (
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
-	eventingcorev1beta1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingcorev1 "knative.dev/eventing/pkg/apis/eventing/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 
@@ -49,10 +49,7 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	eventingv1alpha1.SchemeGroupVersion.WithKind("KafkaSink"):    &eventingv1alpha1.KafkaSink{},
 	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"):    &sourcesv1beta1.KafkaSource{},
 	messagingv1beta1.SchemeGroupVersion.WithKind("KafkaChannel"): &messagingv1beta1.KafkaChannel{},
-}
-
-var validationCallbacks = map[schema.GroupVersionKind]validation.Callback{
-	eventingcorev1beta1.SchemeGroupVersion.WithKind("Broker"): eventingv1.BrokerValidationCallback(),
+	eventingcorev1.SchemeGroupVersion.WithKind("Broker"):         &eventingv1.BrokerStub{},
 }
 
 var defaultingCallbacks = map[schema.GroupVersionKind]defaulting.Callback{
@@ -131,9 +128,6 @@ func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher)
 
 		// Whether to disallow unknown fields.
 		true,
-
-		// Extra validating callbacks to be applied to resources.
-		validationCallbacks,
 	)
 }
 
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation.go b/control-plane/pkg/apis/eventing/v1/broker_validation.go
index fcfd551b8..88465108a 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation.go
@@ -21,25 +21,39 @@ import (
 	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/pkg/webhook"
-	"knative.dev/pkg/webhook/resourcesemantics/validation"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/pkg/webhook/resourcesemantics"
 )
 
-type BrokerPartial struct {
+type BrokerStub struct {
+	metav1.TypeMeta `json:",inline"`
+
 	// +optional
 	metav1.ObjectMeta `json:"metadata,omitempty"`
-	// Spec defines the desired state of the Broker.
-	Spec BrokerPartialSpec `json:"spec,omitempty"`
+
+	Spec eventing.BrokerSpec `json:"spec,omitempty"`
+
+	// +optional
+	Status eventing.BrokerStatus `json:"status,omitempty"`
 }
 
-type BrokerPartialSpec struct {
+var (
+	// Check that Broker is resourcesemantics.GenericCRD.
+	// We do this to make sure the validating/defaulting webhooks can operate on our stub.
+	// If we don't do this stubbing, we will end up doing the defaulting/validating using the upstream Broker type.
+	// However, we want to let the upstream webhook to do its job and not us. We only want to default/validate
+	// the parts relevant to us.
+	_ resourcesemantics.GenericCRD = (*BrokerStub)(nil)
+)
+
+type BrokerStubSpec struct {
 	// Config is a KReference to the configuration that specifies
 	// configuration options for this Broker. For example, this could be
 	// a pointer to a ConfigMap.
@@ -47,27 +61,55 @@ type BrokerPartialSpec struct {
 	Config *duckv1.KReference `json:"config,omitempty"`
 }
 
-func (b BrokerPartialSpec) Validate(context.Context) error {
-	if b.Config == nil {
-		return apis.ErrMissingField("config").ViaField("spec")
+func (b *BrokerStub) Validate(context.Context) *apis.FieldError {
+	// TODO: uncomment the following check when we backport namespaced broker into 1.6
+	// if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass && b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.NamespacedBrokerClass {
+	if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass {
+		// validation for the broker of other classes is done by the other webhooks
+		return nil
+	}
+
+	if b.Spec.Config == nil {
+		// validation of the Config requiredness is done by the upstream webhook and we want to let it do its job.
+		// otherwise API server will say that our Kafka admission controller refused it, but it is a generic validation.
+		return nil
 	}
-	if strings.ToLower(b.Config.Kind) != "configmap" {
-		return apis.ErrInvalidValue(b.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
+
+	// we specifically expect our broker class to use ConfigMap as the config type
+	if strings.ToLower(b.Spec.Config.Kind) != "configmap" {
+		return apis.ErrInvalidValue(b.Spec.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
 	}
+
+	// we specifically expect for our broker classes, that there's a namespace in the config
+	if b.Spec.Config.Namespace == "" {
+		return apis.ErrMissingField(b.Spec.Config.Namespace, "namespace").ViaField("config").ViaField("spec")
+	}
+
+	// TODO: uncomment the following check when we backport namespaced broker into 1.6
+	// for the namespaced broker, we expect the config to be in the same namespace as the broker
+	//if b.Annotations[eventing.BrokerClassAnnotationKey] == kafka.NamespacedBrokerClass && b.Spec.Config.Namespace != b.Namespace {
+	//	return apis.ErrInvalidValue(b.Spec.Config.Namespace, "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec")
+	//}
+
 	return nil
 }
 
-func validateBrokerFromUnstructured(ctx context.Context, unstructured *unstructured.Unstructured) error {
-	broker := BrokerPartial{}
-	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructured.UnstructuredContent(), &broker); err != nil {
-		return err
-	}
-	if class, ok := broker.Annotations[eventing.BrokerClassAnnotationKey]; !ok || class != kafka.BrokerClass {
+func (b *BrokerStub) SetDefaults(ctx context.Context) {
+	// do nothing, we let the upstream webhook do its job
+}
+
+func (b *BrokerStub) DeepCopyObject() runtime.Object {
+	if b == nil {
 		return nil
 	}
-	return broker.Spec.Validate(ctx)
-}
 
-func BrokerValidationCallback() validation.Callback {
-	return validation.NewCallback(validateBrokerFromUnstructured, webhook.Create, webhook.Update)
+	out := &BrokerStub{
+		TypeMeta: b.TypeMeta, //simple struct
+	}
+
+	b.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	b.Spec.DeepCopyInto(&out.Spec)
+	b.Status.DeepCopyInto(&out.Status)
+
+	return out
 }


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
index 0c08ba763..207dabf26 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
@@ -20,88 +20,124 @@ import (
 	"context"
 	"testing"
 
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
-func TestValidateBrokerFromUnstructured(t *testing.T) {
+func TestValidate(t *testing.T) {
 	tests := []struct {
-		name         string
-		ctx          context.Context
-		unstructured *unstructured.Unstructured
-		wantErr      bool
-	}{
-		{
-			name: "no kafka broker",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{"spec": map[string]interface{}{}},
+		name string
+		b    BrokerStub
+		want *apis.FieldError
+	}{{
+		name: "missing annotation",
+		b:    BrokerStub{},
+	}, {
+		name: "empty annotation",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": ""},
 			},
-			wantErr: false,
 		},
-		{
-			name: "missing config",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{},
+	}, {
+		name: "other broker class",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Foo"},
+			},
+		},
+	}, {
+		name: "no spec.config",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+		},
+	}, {
+		name: "no spec.config.namespace",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Kind: "ConfigMap",
 				},
 			},
-			wantErr: true,
 		},
-		{
-			name: "unknown kind",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{
-						"config": map[string]string{
-							"apiVersion": "eventing.knative.dev/v1",
-							"kind":       "Broker",
-							"namespace":  "ns",
-							"name":       "name",
-						},
-					},
+		want: apis.ErrMissingField("", "namespace").ViaField("config").ViaField("spec"),
+	}, {
+		name: "spec.config is not configmap",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Kind:      "Service",
+					Namespace: "foo",
 				},
 			},
-			wantErr: true,
 		},
-		{
-			name: "ConfigMap in config",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{
-						"config": map[string]string{
-							"apiVersion": "v1",
-							"kind":       "ConfigMap",
-							"namespace":  "ns",
-							"name":       "name",
-						},
-					},
+		want: apis.ErrInvalidValue("Service", "kind", "Expected ConfigMap").ViaField("config").ViaField("spec"),
+		// TODO: uncomment the following check when we backport namespaced broker into 1.6
+		//}, {
+		//	name: "spec.config.namespace is different",
+		//	b: BrokerStub{
+		//		ObjectMeta: metav1.ObjectMeta{
+		//			Namespace:   "my-namespace",
+		//			Annotations: map[string]string{"eventing.knative.dev/broker.class": "KafkaNamespaced"},
+		//		},
+		//		Spec: eventingv1.BrokerSpec{
+		//			Config: &duckv1.KReference{
+		//				Kind:      "ConfigMap",
+		//				Namespace: "foo",
+		//			},
+		//		},
+		//	},
+		//	want: apis.ErrInvalidValue("foo", "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec"),
+		// TODO: uncomment the following check when we backport namespaced broker into 1.6
+		//}, {
+		//	name: "valid config - namespaced broker",
+		//	b: BrokerStub{
+		//		ObjectMeta: metav1.ObjectMeta{
+		//			Namespace:   "my-namespace",
+		//			Annotations: map[string]string{"eventing.knative.dev/broker.class": "KafkaNamespaced"},
+		//		},
+		//		Spec: eventingv1.BrokerSpec{
+		//			Config: &duckv1.KReference{
+		//				Namespace:  "my-namespace",
+		//				Name:       "name",
+		//				Kind:       "ConfigMap",
+		//				APIVersion: "v1",
+		//			},
+		//		},
+		//	},
+	}, {
+		name: "valid config - regular broker",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:   "my-namespace",
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Name:       "name",
+					Namespace:  "my-namespace",
+					Kind:       "ConfigMap",
+					APIVersion: "v1",
 				},
 			},
-			wantErr: false,
 		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := validateBrokerFromUnstructured(tt.ctx, tt.unstructured); (err != nil) != tt.wantErr {
-				t.Errorf("validateBrokerFromUnstructured() error = %v, wantErr %v", err, tt.wantErr)
+	}}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := test.b.Validate(context.Background())
+			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+				t.Error("Broker.Validate (-want, +got) =", diff)
 			}
 		})
 	}