Repository: knative-sandbox/eventing-kafka-broker
Pull request #2622 (2022-09-08 12:50:02)

[release-1.7] Set TerminationMessagePolicy=FallbackToLogsOnError in a test image
This is an automated cherry-pick of #2561

```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/2622#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign pierdipi for approval by writing `/assign @pierdipi` 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.7/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":["pierdipi"]} -->
aliok: /lgtm
/approve
aliok: /cherry-pick release-1.6
knative-prow-robot: @aliok: once the present PR merges, I will cherry-pick it on top of release-1.6 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2622#issuecomment-1240676128):

>/cherry-pick release-1.6


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[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2622#issuecomment-1240676022" title="Approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2622#" 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.7/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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2622?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2622](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2622?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (909c914) into [release-1.7](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/31d6268ceda3ab292d33cd0805f054de22dc64cd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (31d6268) will **not change** coverage.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##           release-1.7    #2622   +/-   ##
============================================
  Coverage        56.29%   56.29%           
============================================
  Files               74       74           
  Lines             7646     7646           
============================================
  Hits              4304     4304           
  Misses            3043     3043           
  Partials           299      299           
```



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)


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..ab26ea680 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,52 @@ 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 {
+	if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass && b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.NamespacedBrokerClass {
+		// validation for the broker of other classes is done by the other webhooks
+		return nil
 	}
-	if strings.ToLower(b.Config.Kind) != "configmap" {
-		return apis.ErrInvalidValue(b.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
+
+	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
 	}
+
+	// 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")
+	}
+
+	// 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..1b41ae662 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,122 @@ 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"),
+	}, {
+		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",
 				},
 			},
-			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)
+		want: apis.ErrInvalidValue("foo", "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec"),
+	}, {
+		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",
+				},
+			},
+		},
+	}}
+	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)
 			}
 		})
 	}
diff --git a/test/pkg/kafka/verify_num_partitions_replication.go b/test/pkg/kafka/verify_num_partitions_replication.go
index d6198482c..0e5e03ff7 100644
--- a/test/pkg/kafka/verify_num_partitions_replication.go
+++ b/test/pkg/kafka/verify_num_partitions_replication.go
@@ -66,9 +66,10 @@ func VerifyNumPartitionAndReplicationFactor(
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
 						{
-							Name:            name,
-							Image:           pkgtest.ImagePath(partitionReplicationVerifierImage),
-							ImagePullPolicy: corev1.PullIfNotPresent,
+							Name:                     name,
+							Image:                    pkgtest.ImagePath(partitionReplicationVerifierImage),
+							ImagePullPolicy:          corev1.PullIfNotPresent,
+							TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError,
 							Env: []corev1.EnvVar{
 								{
 									Name:  "BOOTSTRAP_SERVERS",