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

Issue:
[mtsource] mtadapter crashes when SASL and/or TLS secret is not found 
**Describe the bug**
The mtadapter pod crashes when a secret is missing

 **Expected behavior**
A clear and concise description of what you expected to happen.
no crash

 **To Reproduce**
Steps to reproduce the behavior.
Create a KafkaSource pointing to bogus secrets

 **Knative release version**

 **Additional context**
Add any other context about the problem here such as proposed priority


Pull request:
[mtsource] Fix crash when secret is missing
Fixes #614

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Log error when one of the secret is invalid, and retry.
-
-

<!--
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.
-->


Patch:
diff --git a/pkg/source/mtadapter/adapter.go b/pkg/source/mtadapter/adapter.go
index bcd87683a..67e39c4ed 100644
--- a/pkg/source/mtadapter/adapter.go
+++ b/pkg/source/mtadapter/adapter.go
@@ -18,6 +18,7 @@ package mtadapter
 
 import (
 	"context"
+	"fmt"
 	"math"
 	"strconv"
 	"sync"
@@ -147,7 +148,7 @@ func (a *Adapter) bufferSize(ctx context.Context,
 
 // Implements MTAdapter
 
-func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
+func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) error {
 	a.sourcesMu.Lock()
 	defer a.sourcesMu.Unlock()
 
@@ -168,7 +169,37 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 	if placement == nil || placement.VReplicas == 0 {
 		// this pod does not handle this source. Skipping
 		logger.Info("no replicas assigned to this source. skipping")
-		return
+		return nil
+	}
+
+	saslUser, err := a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.User.SecretKeyRef)
+	if err != nil {
+		return err
+	}
+
+	saslPassword, err := a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Password.SecretKeyRef)
+	if err != nil {
+		return err
+	}
+
+	saslType, err := a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Type.SecretKeyRef)
+	if err != nil {
+		return err
+	}
+
+	tlsCert, err := a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Cert.SecretKeyRef)
+	if err != nil {
+		return err
+	}
+
+	tlsKey, err := a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Key.SecretKeyRef)
+	if err != nil {
+		return err
+	}
+
+	tlsCACert, err := a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.CACert.SecretKeyRef)
+	if err != nil {
+		return err
 	}
 
 	kafkaEnvConfig := client.KafkaEnvConfig{
@@ -176,15 +207,15 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 		Net: client.AdapterNet{
 			SASL: client.AdapterSASL{
 				Enable:   obj.Spec.Net.SASL.Enable,
-				User:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.User.SecretKeyRef),
-				Password: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Password.SecretKeyRef),
-				Type:     a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.SASL.Type.SecretKeyRef),
+				User:     saslUser,
+				Password: saslPassword,
+				Type:     saslType,
 			},
 			TLS: client.AdapterTLS{
 				Enable: obj.Spec.Net.TLS.Enable,
-				Cert:   a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Cert.SecretKeyRef),
-				Key:    a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.Key.SecretKeyRef),
-				CACert: a.ResolveSecret(ctx, obj.Namespace, obj.Spec.Net.TLS.CACert.SecretKeyRef),
+				Cert:   tlsCert,
+				Key:    tlsKey,
+				CACert: tlsCACert,
 			},
 		},
 	}
@@ -194,7 +225,7 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 		// TODO: periodically enforce limits as the number of partitions can dynamically change
 		bufferSizePerVReplica, err := a.bufferSize(ctx, logger, &kafkaEnvConfig, obj.Spec.Topics, scheduler.GetPodCount(obj.Status.Placement))
 		if err != nil {
-			return
+			return err
 		}
 		bufferSize := bufferSizePerVReplica * int(placement.VReplicas)
 		a.logger.Infow("setting fetch buffer size", zap.Int("size", bufferSize))
@@ -226,11 +257,13 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 	reporter, err := pkgsource.NewStatsReporter()
 	if err != nil {
 		a.logger.Error("error building statsreporter", zap.Error(err))
+		return err
 	}
 
 	httpBindingsSender, err := kncloudevents.NewHTTPMessageSenderWithTarget(obj.Status.SinkURI.String())
 	if err != nil {
-		a.logger.Fatalw("error building cloud event client", zap.Error(err))
+		a.logger.Errorw("error building cloud event client", zap.Error(err))
+		return err
 	}
 
 	adapter := a.adapterCtor(ctx, &config, httpBindingsSender, reporter)
@@ -250,6 +283,7 @@ func (a *Adapter) Update(ctx context.Context, obj *v1beta1.KafkaSource) {
 
 	a.sources[key] = cancelFn
 	a.logger.Infow("source added", "name", obj.Name)
+	return nil
 }
 
 func (a *Adapter) Remove(obj *v1beta1.KafkaSource) {
@@ -273,15 +307,20 @@ func (a *Adapter) Remove(obj *v1beta1.KafkaSource) {
 }
 
 // ResolveSecret resolves the secret reference
-func (a *Adapter) ResolveSecret(ctx context.Context, ns string, ref *corev1.SecretKeySelector) string {
+func (a *Adapter) ResolveSecret(ctx context.Context, ns string, ref *corev1.SecretKeySelector) (string, error) {
 	if ref == nil {
-		return ""
+		return "", nil
 	}
 	secret, err := a.kubeClient.CoreV1().Secrets(ns).Get(ctx, ref.Name, metav1.GetOptions{})
 	if err != nil {
-		a.logger.Fatalw("failed to read secret", zap.String("secretname", ref.Name), zap.Error(err))
-		return ""
+		a.logger.Errorw("failed to read secret", zap.String("secretname", ref.Name), zap.Error(err))
+		return "", err
+	}
+
+	if value, ok := secret.Data[ref.Key]; ok && len(value) > 0 {
+		return string(value), nil
 	}
 
-	return string(secret.Data[ref.Key])
+	a.logger.Errorw("missing secret key or empty secret value", zap.String("secretname", ref.Name), zap.String("secretkey", ref.Key))
+	return "", fmt.Errorf("missing secret key or empty secret value (%s/%s)", ref.Name, ref.Key)
 }
diff --git a/pkg/source/mtadapter/controller.go b/pkg/source/mtadapter/controller.go
index 2f7654087..061b6eedf 100644
--- a/pkg/source/mtadapter/controller.go
+++ b/pkg/source/mtadapter/controller.go
@@ -34,7 +34,7 @@ import (
 // MTAdapter is the interface the multi-tenant KafkaSource adapter must implement
 type MTAdapter interface {
 	// Update is called when the source is ready and when the specification and/or status has changed.
-	Update(ctx context.Context, source *v1beta1.KafkaSource)
+	Update(ctx context.Context, source *v1beta1.KafkaSource) error
 
 	// Remove is called when the source has been deleted.
 	Remove(source *v1beta1.KafkaSource)
diff --git a/pkg/source/mtadapter/kafkasource.go b/pkg/source/mtadapter/kafkasource.go
index 336420823..d97e1fb12 100644
--- a/pkg/source/mtadapter/kafkasource.go
+++ b/pkg/source/mtadapter/kafkasource.go
@@ -41,9 +41,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1beta1.KafkaSou
 	}
 
 	// Update the adapter state
-	r.mtadapter.Update(ctx, source)
-
-	return nil
+	return r.mtadapter.Update(ctx, source)
 }
 
 func (r *Reconciler) deleteFunc(obj interface{}) {


Test patch:
diff --git a/pkg/source/mtadapter/adapter_test.go b/pkg/source/mtadapter/adapter_test.go
index f7bf2231b..676950a2d 100644
--- a/pkg/source/mtadapter/adapter_test.go
+++ b/pkg/source/mtadapter/adapter_test.go
@@ -21,9 +21,11 @@ import (
 	"testing"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 
-	_ "knative.dev/pkg/client/injection/kube/client/fake"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	pkgtesting "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/source"
 
@@ -63,7 +65,7 @@ func TestUpdateRemoveSources(t *testing.T) {
 		adapterStopped <- true
 	}()
 
-	adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
+	err := adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name",
 			Namespace: "test-ns",
@@ -77,11 +79,15 @@ func TestUpdateRemoveSources(t *testing.T) {
 		},
 	})
 
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
 	if _, ok := adapter.sources["test-ns/test-name"]; !ok {
 		t.Error(`Expected adapter to contain "test-ns/test-name"`)
 	}
 
-	adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
+	err = adapter.Update(ctx, &sourcesv1beta1.KafkaSource{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-name-badBroker",
 			Namespace: "test-ns",
@@ -99,6 +105,10 @@ func TestUpdateRemoveSources(t *testing.T) {
 		},
 	})
 
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
 	if _, ok := adapter.sources["test-ns/test-name-badBroker"]; !ok {
 		t.Error(`Expected adapter to contain "test-ns/test-name-badBroker"`)
 	}
@@ -163,6 +173,201 @@ func TestUpdateRemoveSources(t *testing.T) {
 	}
 }
 
+func TestSourceMTAdapter(t *testing.T) {
+	testCases := map[string]struct {
+		objects []runtime.Object
+		wantErr bool
+		source  sourcesv1beta1.KafkaSource
+	}{
+		"with sasl secret": {
+			wantErr: false,
+			objects: []runtime.Object{
+				&corev1.Secret{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "mysecret",
+						Namespace: "test-ns",
+					},
+					Data: map[string][]byte{
+						"user1": []byte("fjwoia"),
+					},
+				},
+			},
+			source: sourcesv1beta1.KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-missing-sasl-secret",
+					Namespace: "test-ns",
+				},
+				Spec: sourcesv1beta1.KafkaSourceSpec{
+					KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+						Net: bindingsv1beta1.KafkaNetSpec{
+							SASL: bindingsv1beta1.KafkaSASLSpec{
+								User: bindingsv1beta1.SecretValueFromSource{
+									SecretKeyRef: &corev1.SecretKeySelector{
+										LocalObjectReference: corev1.LocalObjectReference{
+											Name: "mysecret",
+										},
+										Key: "user1",
+									},
+								},
+							},
+						},
+					},
+				},
+				Status: sourcesv1beta1.KafkaSourceStatus{
+					Placeable: duckv1alpha1.Placeable{
+						Placement: []duckv1alpha1.Placement{
+							{PodName: podName, VReplicas: int32(1)},
+						}},
+				},
+			},
+		},
+		"missing sasl secret key": {
+			wantErr: true,
+			objects: []runtime.Object{
+				&corev1.Secret{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "mysecret",
+						Namespace: "test-ns",
+					},
+					Data: map[string][]byte{
+						"user": []byte("fjwoia"),
+					},
+				},
+			},
+			source: sourcesv1beta1.KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-missing-sasl-secret",
+					Namespace: "test-ns",
+				},
+				Spec: sourcesv1beta1.KafkaSourceSpec{
+					KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+						Net: bindingsv1beta1.KafkaNetSpec{
+							SASL: bindingsv1beta1.KafkaSASLSpec{
+								User: bindingsv1beta1.SecretValueFromSource{
+									SecretKeyRef: &corev1.SecretKeySelector{
+										LocalObjectReference: corev1.LocalObjectReference{
+											Name: "mysecret",
+										},
+										Key: "user1",
+									},
+								},
+							},
+						},
+					},
+				},
+				Status: sourcesv1beta1.KafkaSourceStatus{
+					Placeable: duckv1alpha1.Placeable{
+						Placement: []duckv1alpha1.Placement{
+							{PodName: podName, VReplicas: int32(1)},
+						}},
+				},
+			},
+		},
+		"missing sasl secret": {
+			wantErr: true,
+			source: sourcesv1beta1.KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-missing-sasl-secret",
+					Namespace: "test-ns",
+				},
+				Spec: sourcesv1beta1.KafkaSourceSpec{
+					KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+						Net: bindingsv1beta1.KafkaNetSpec{
+							SASL: bindingsv1beta1.KafkaSASLSpec{
+								User: bindingsv1beta1.SecretValueFromSource{
+									SecretKeyRef: &corev1.SecretKeySelector{
+										LocalObjectReference: corev1.LocalObjectReference{
+											Name: "bogus",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				Status: sourcesv1beta1.KafkaSourceStatus{
+					Placeable: duckv1alpha1.Placeable{
+						Placement: []duckv1alpha1.Placement{
+							{PodName: podName, VReplicas: int32(1)},
+						}},
+				},
+			},
+		},
+		"missing tls secret": {
+			wantErr: true,
+			source: sourcesv1beta1.KafkaSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-missing-tls-secret",
+					Namespace: "test-ns",
+				},
+				Spec: sourcesv1beta1.KafkaSourceSpec{
+					KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+						Net: bindingsv1beta1.KafkaNetSpec{
+							TLS: bindingsv1beta1.KafkaTLSSpec{
+								Cert: bindingsv1beta1.SecretValueFromSource{
+									SecretKeyRef: &corev1.SecretKeySelector{
+										LocalObjectReference: corev1.LocalObjectReference{
+											Name: "bogus",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				Status: sourcesv1beta1.KafkaSourceStatus{
+					Placeable: duckv1alpha1.Placeable{
+						Placement: []duckv1alpha1.Placement{
+							{PodName: podName, VReplicas: int32(1)},
+						}},
+				},
+			},
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			ctx, _ := pkgtesting.SetupFakeContext(t)
+
+			if tc.objects != nil {
+				ctx, _ = fakekubeclient.With(ctx, tc.objects...)
+			}
+
+			ctx, cancelAdapter := context.WithCancel(ctx)
+
+			env := &AdapterConfig{PodName: podName, MemoryLimit: "0"}
+			ceClient := adaptertest.NewTestClient()
+
+			adapter := newAdapter(ctx, env, ceClient, newSampleAdapter).(*Adapter)
+
+			adapterStopped := make(chan bool)
+			go func() {
+				err := adapter.Start(ctx)
+				if err != nil {
+					t.Error("Unexpected error ", err)
+				}
+				adapterStopped <- true
+			}()
+
+			err := adapter.Update(ctx, &tc.source)
+			if tc.wantErr && err == nil {
+				t.Error("Error expected, got none")
+			}
+
+			if !tc.wantErr && err != nil {
+				t.Errorf("No error expected, got %v", err)
+			}
+
+			cancelAdapter()
+
+			select {
+			case <-adapterStopped:
+			case <-time.After(2 * time.Second):
+				t.Error("adapter failed to stop after 2 seconds")
+			}
+		})
+	}
+}
+
 type sampleAdapter struct {
 	running bool
 }