feat: malformatted EventMesh Secret causes warning state instead of Error state

internal/controller/operator/eventing/controller.go

@@ -669,6 +669,12 @@ func (r *Reconciler) reconcileEventMeshBackend(ctx context.Context, eventing *op
 	// Start the EventMesh subscription controller
 	err = r.reconcileEventMeshSubManager(ctx, eventing, eventMeshSecret)
 	if err != nil {
+		// In case the the error is caused by a malformatted secret, we want to set the status to warning,
+		// to indicate the requirement of user interaction.
+		if IsMalformattedSecretErr(err) {
+			return kctrl.Result{}, r.syncSubManagerStatusWithNATSState(ctx, operatorv1alpha1.StateWarning, eventing,
+				ErrEventMeshSecretMalformatted, log)
+		}
 		return kctrl.Result{}, r.syncStatusWithSubscriptionManagerErr(ctx, eventing, err, log)
 	}
 	eventing.Status.SetSubscriptionManagerReadyConditionToTrue()

internal/controller/operator/eventing/controller_test.go

@@ -452,8 +452,8 @@ func Test_stopNatsCRWatch(t *testing.T) {
 			testEnv.Reconciler.stopNATSCRWatch(eventing)
 
 			// Check the results
-			require.Equal(t, nil, testcase.watchNatsWatcher)
-			require.False(t, testcase.natsCRWatchStarted)
+			require.Nil(t, tc.watchNatsWatcher)
+			require.False(t, tc.natsCRWatchStarted)
 		})
 	}
 }

internal/controller/operator/eventing/eventmesh.go

@@ -28,13 +28,17 @@ const (
 	secretKeyTokenURL     = "token_url"
 	secretKeyCertsURL     = "certs_url"
 
+	EMSecretMessagingMissing      = "messaging is missing from EM secret"
+	EMSecretNamespaceMissing      = "namespace is missing from EM secret"
 	EventMeshSecretMissingMessage = "The specified EventMesh secret is not found. Please provide an existing secret."
+	EventMeshSecretMalformatted   = "The EventMesh secret data is not formatted properly."
 )
 
 var (
-	ErrEMSecretMessagingMissing = errors.New("messaging is missing from EM secret")
-	ErrEMSecretNamespaceMissing = errors.New("namespace is missing from EM secret")
-	ErrEventMeshSecretMissing   = errors.New(EventMeshSecretMissingMessage)
+	ErrEMSecretMessagingMissing    = errors.New(EMSecretMessagingMissing)
+	ErrEMSecretNamespaceMissing    = errors.New(EMSecretNamespaceMissing)
+	ErrEventMeshSecretMissing      = errors.New(EventMeshSecretMissingMessage)
+	ErrEventMeshSecretMalformatted = errors.New(EventMeshSecretMalformatted)
 )
 
 type oauth2Credentials struct {
@@ -44,6 +48,15 @@ type oauth2Credentials struct {
 	certsURL     []byte
 }
 
+func newMalformattedSecretErr(e string) error {
+	// The new error is handled as a string (%s), the malformation error is wrapped as an Err.
+	return fmt.Errorf("%s: %w", e, ErrEventMeshSecretMalformatted)
+}
+
+func IsMalformattedSecretErr(err error) bool {
+	return errors.Is(err, ErrEventMeshSecretMalformatted)
+}
+
 func (r *Reconciler) reconcileEventMeshSubManager(ctx context.Context, eventing *v1alpha1.Eventing, eventMeshSecret *kcorev1.Secret) error {
 	// gets oauth2ClientID and secret and stops the EventMesh subscription manager if changed
 	err := r.syncOauth2ClientIDAndSecret(ctx, eventing)
@@ -314,38 +327,38 @@ func getSecretForPublisher(eventMeshSecret *kcorev1.Secret) (*kcorev1.Secret, er
 		label.KeyName: label.ValueEventingPublisherProxy,
 	}
 
-	if _, ok := eventMeshSecret.Data["messaging"]; !ok {
-		return nil, ErrEMSecretMessagingMissing
-	}
-	messagingBytes := eventMeshSecret.Data["messaging"]
-
 	if _, ok := eventMeshSecret.Data["namespace"]; !ok {
-		return nil, ErrEMSecretNamespaceMissing
+		return nil, newMalformattedSecretErr(EMSecretNamespaceMissing)
 	}
 	namespaceBytes := eventMeshSecret.Data["namespace"]
 
+	if _, ok := eventMeshSecret.Data["messaging"]; !ok {
+		return nil, newMalformattedSecretErr(EMSecretMessagingMissing)
+	}
+	messagingBytes := eventMeshSecret.Data["messaging"]
+
 	var messages []Message
 	err := json.Unmarshal(messagingBytes, &messages)
 	if err != nil {
-		return nil, err
+		return nil, newMalformattedSecretErr(err.Error())
 	}
 
 	for _, msg := range messages {
 		if msg.Broker.BrokerType == "saprestmgw" {
 			if len(msg.OA2.ClientID) == 0 {
-				return nil, errors.New("client ID is missing")
+				return nil, newMalformattedSecretErr("client ID is missing")
 			}
 			if len(msg.OA2.ClientSecret) == 0 {
-				return nil, errors.New("client secret is missing")
+				return nil, newMalformattedSecretErr("client secret is missing")
 			}
 			if len(msg.OA2.TokenEndpoint) == 0 {
-				return nil, errors.New("tokenendpoint is missing")
+				return nil, newMalformattedSecretErr("tokenendpoint is missing")
 			}
 			if len(msg.OA2.GrantType) == 0 {
-				return nil, errors.New("granttype is missing")
+				return nil, newMalformattedSecretErr("granttype is missing")
 			}
 			if len(msg.URI) == 0 {
-				return nil, errors.New("publish URL is missing")
+				return nil, newMalformattedSecretErr("publish URL is missing")
 			}
 
 			secret.StringData = getSecretStringData(msg.OA2.ClientID, msg.OA2.ClientSecret, msg.OA2.TokenEndpoint, msg.OA2.GrantType, msg.URI, string(namespaceBytes))

internal/controller/operator/eventing/eventmesh_test.go

@@ -702,7 +702,8 @@ func Test_GetSecretForPublisher(t *testing.T) {
 			gotPublisherSecret, err := getSecretForPublisher(publisherSecret)
 			if testcase.expectedError != nil {
 				require.Error(t, err)
-				require.ErrorContains(t, err, testcase.expectedError.Error())
+				require.ErrorIs(t, err, ErrEventMeshSecretMalformatted)
+				require.ErrorContains(t, err, tc.expectedError.Error())
 				return
 			}
 			require.NoError(t, err)
@@ -1117,3 +1118,37 @@ func Test_syncOauth2ClientIDAndSecret(t *testing.T) {
 		})
 	}
 }
+
+// Test_IsMalfromattedSecret verifies that the function IsMalformattedSecretErr asses correctly
+// if a given error is a malformatted secret error or not.
+func Test_IsMalfromattedSecret(t *testing.T) {
+	testCases := []struct {
+		name       string
+		givenErr   error
+		wantResult bool
+	}{
+		{
+			name:       "should return true when an error is an ErrMalformedSecret",
+			givenErr:   ErrEventMeshSecretMalformatted,
+			wantResult: true,
+		}, {
+			name:       "should return true when an error is a wrapped ErrMalformedSecret",
+			givenErr:   newMalformattedSecretErr("this error will wrap ErrMalformedSecret"),
+			wantResult: true,
+		}, {
+			name:       "should return false when an error is not an ErrMalformedSecret",
+			givenErr:   fmt.Errorf("this is not a malformed secret error"),
+			wantResult: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// when
+			result := IsMalformattedSecretErr(tc.givenErr)
+
+			// then
+			require.Equal(t, tc.wantResult, result)
+		})
+	}
+}

internal/controller/operator/eventing/integrationtests/controller/integration_test.go

@@ -577,6 +577,123 @@ func Test_WatcherEventingCRK8sObjects(t *testing.T) {
 	}
 }
 
+func Test_EventMesh_MalformattedSecret(t *testing.T) {
+	testCases := []struct {
+		name            string
+		givenSecretData map[string][]byte
+		wantMatcher     gomegatypes.GomegaMatcher
+	}{
+		{
+			name: "should have ready state when EventMesh secret is malformatted",
+			givenSecretData: map[string][]byte{
+				"management": []byte("foo"),
+				"messaging": []byte(`[
+			  {
+				"broker": {
+				  "type": "bar"
+				},
+				"oa2": {
+				  "clientid": "foo",
+				  "clientsecret": "foo",
+				  "granttype": "client_credentials",
+				  "tokenendpoint": "bar"
+				},
+				"protocol": [
+				  "amqp10ws"
+				],
+				"uri": "foo"
+			  },
+			  {
+				"broker": {
+				  "type": "foo"
+				},
+				"oa2": {
+				  "clientid": "bar",
+				  "clientsecret": "bar",
+				  "granttype": "client_credentials",
+				  "tokenendpoint": "foo"
+				},
+				"protocol": [
+				  "bar"
+				],
+				"uri": "bar"
+			  },
+			  {
+				"broker": {
+				  "type": "foo"
+				},
+				"oa2": {
+				  "clientid": "foo",
+				  "clientsecret": "bar",
+				  "granttype": "client_credentials",
+				  "tokenendpoint": "foo"
+				},
+				"protocol": [
+				  "httprest"
+				],
+				"uri": "bar"
+			  }
+			]`),
+				"namespace":         []byte("bar"),
+				"serviceinstanceid": []byte("foo"),
+				"xsappname":         []byte("bar"),
+			},
+			wantMatcher: gomega.And(
+				matchers.HaveStatusReady(),
+			),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			g := gomega.NewWithT(t)
+
+			// Given:
+			// Make the Deployment always beeing ready.
+			eventingcontroller.IsDeploymentReady = func(deployment *kappsv1.Deployment) bool {
+				return true
+			}
+
+			// Create an Eventing CR. We do not care about the details in this test.
+			givenEventingCR := utils.NewEventingCR(
+				utils.WithEventMeshBackend("test-secret-name"),
+				utils.WithEventingPublisherData(1, 1, "199m", "99Mi", "399m", "199Mi"),
+				utils.WithEventingEventTypePrefix("test-prefix"),
+			)
+
+			// Create an unique Namespace for this test run.
+			givenNamespace := givenEventingCR.Namespace
+			testEnvironment.EnsureNamespaceCreation(t, givenNamespace)
+
+			// // Create an EventMesh Secret. This is the crucial part of the test;
+			// // if the Secret is malformatted, Eventing should have the warning status.
+			// // Frist we need to extract the Secret name and namespace from the given Eventing CR.
+			// subArray := strings.Split(givenEventingCR.Spec.Backend.Config.EventMeshSecret, "/")
+			// secretName, secretNameSpace := subArray[1], subArray[0]
+			// // Now we can assemble the Secret.
+			// secret := kcorev1.Secret{
+			// 	ObjectMeta: kmetav1.ObjectMeta{
+			// 		Name:      secretName,
+			// 		Namespace: secretNameSpace,
+			// 	},
+			// 	Data: tc.givenSecretData,
+			// 	Type: "Opaque",
+			// }
+			// // Finally, we create the Secret in the K8s cluster.
+			// testEnvironment.EnsureEventMeshSecretCreated(t, &secret)
+			testEnvironment.EnsureDefaultEventMeshSecretCreated(t, givenEventingCR)
+
+			// When:
+			// Create the Eventing CR with the given EventMesh Secret.
+			testEnvironment.EnsureK8sResourceCreated(t, givenEventingCR)
+
+			// Then:
+			// Check the Eventing CR status against the expected status.
+			testEnvironment.GetEventingAssert(g, givenEventingCR).Should(tc.wantMatcher)
+		})
+	}
+}
+
 func Test_CreateEventingCR_EventMesh(t *testing.T) {
 	testCases := []struct {
 		name                          string
@@ -662,16 +779,15 @@ func Test_CreateEventingCR_EventMesh(t *testing.T) {
 			}
 
 			// create unique namespace for this test run.
-			givenNamespace := testcase.givenEventing.Namespace
-
+			givenNamespace := tc.givenEventing.Namespace
 			testEnvironment.EnsureNamespaceCreation(t, givenNamespace)
 
 			// create eventing-webhook-auth secret.
 			testEnvironment.EnsureOAuthSecretCreated(t, testcase.givenEventing)
 
 			if !testcase.shouldEventMeshSecretNotFound {
 				// create EventMesh secret.
-				testEnvironment.EnsureEventMeshSecretCreated(t, testcase.givenEventing)
+				testEnvironment.EnsureDefaultEventMeshSecretCreated(t, tc.givenEventing)
 			}
 
 			originalKubeClient := testEnvironment.KubeClient
@@ -804,8 +920,8 @@ func Test_DeleteEventingCR(t *testing.T) {
 				testEnvironment.EnsureNATSResourceStateReady(t, nats)
 			} else {
 				// create eventing-webhook-auth secret.
-				testEnvironment.EnsureOAuthSecretCreated(t, testcase.givenEventing)
-				testEnvironment.EnsureEventMeshSecretCreated(t, testcase.givenEventing)
+				testEnvironment.EnsureOAuthSecretCreated(t, tc.givenEventing)
+				testEnvironment.EnsureDefaultEventMeshSecretCreated(t, tc.givenEventing)
 			}
 			testEnvironment.EnsureK8sResourceCreated(t, testcase.givenEventing)
 
@@ -936,7 +1052,7 @@ func Test_WatcherNATSResource(t *testing.T) {
 				)
 				// create necessary EventMesh secrets
 				testEnvironment.EnsureOAuthSecretCreated(t, eventingResource)
-				testEnvironment.EnsureEventMeshSecretCreated(t, eventingResource)
+				testEnvironment.EnsureDefaultEventMeshSecretCreated(t, eventingResource)
 			} else {
 				eventingResource = utils.NewEventingCR(
 					utils.WithEventingCRNamespace(givenNamespace),

internal/controller/operator/eventing/integrationtests/controllerswitching/integration_test.go

@@ -166,10 +166,10 @@ func Test_Switching(t *testing.T) {
 			// create eventing-webhook-auth secret.
 			testEnvironment.EnsureOAuthSecretCreated(t, testcase.givenEventing)
 			// create EventMesh secret.
-			if testcase.givenEventing.Spec.Backend.Type == operatorv1alpha1.EventMeshBackendType {
-				testEnvironment.EnsureEventMeshSecretCreated(t, testcase.givenEventing)
-			} else if testcase.givenSwitchedEventing.Spec.Backend.Type == operatorv1alpha1.EventMeshBackendType {
-				testEnvironment.EnsureEventMeshSecretCreated(t, testcase.givenSwitchedEventing)
+			if tc.givenEventing.Spec.Backend.Type == operatorv1alpha1.EventMeshBackendType {
+				testEnvironment.EnsureDefaultEventMeshSecretCreated(t, tc.givenEventing)
+			} else if tc.givenSwitchedEventing.Spec.Backend.Type == operatorv1alpha1.EventMeshBackendType {
+				testEnvironment.EnsureDefaultEventMeshSecretCreated(t, tc.givenSwitchedEventing)
 			}
 
 			// create Eventing CR.

internal/controller/operator/eventing/integrationtests/without_apirule_crd/integration_test.go

@@ -91,7 +91,7 @@ func Test_EventMesh_APIRule_Dependency_Check(t *testing.T) {
 			testEnvironment.EnsureOAuthSecretCreated(t, testcase.givenEventing)
 
 			// create EventMesh secret.
-			testEnvironment.EnsureEventMeshSecretCreated(t, testcase.givenEventing)
+			testEnvironment.EnsureDefaultEventMeshSecretCreated(t, tc.givenEventing)
 
 			// when
 			// create Eventing CR.

test/utils/integration/integration.go

@@ -915,10 +915,15 @@ func (env TestEnvironment) EnsureCABundleInjectedIntoWebhooks(t *testing.T) {
 	}, SmallTimeOut, SmallPollingInterval, "failed to ensure correctness of CABundle in Webhooks")
 }
 
-func (env TestEnvironment) EnsureEventMeshSecretCreated(t *testing.T, eventing *v1alpha1.Eventing) {
+func (env TestEnvironment) EnsureDefaultEventMeshSecretCreated(t *testing.T, eventing *v1alpha1.Eventing) {
 	t.Helper()
 	subarr := strings.Split(eventing.Spec.Backend.Config.EventMeshSecret, "/")
 	secret := testutils.NewEventMeshSecret(subarr[1], subarr[0])
+	env.EnsureEventMeshSecretCreated(t, secret)
+}
+
+func (env TestEnvironment) EnsureEventMeshSecretCreated(t *testing.T, secret *kcorev1.Secret) {
+	t.Helper()
 	env.EnsureK8sResourceCreated(t, secret)
 }
 

test/utils/utils.go

@@ -109,6 +109,8 @@ func NewAPIRuleCRD() *kapiextensionsv1.CustomResourceDefinition {
 	return result
 }
 
+// NewEventingCR creates a new Eventing CR with the given options.
+// If no options are provided, the default (e. g. random name and namespace) will be used.
 func NewEventingCR(opts ...EventingOption) *v1alpha1.Eventing {
 	name := fmt.Sprintf(NameFormat, GetRandString(randomNameLen))
 	namespace := fmt.Sprintf(NamespaceFormat, GetRandString(randomNameLen))
@@ -266,6 +268,17 @@ func NewDeployment(name, namespace string, annotations map[string]string) *kapps
 	}
 }
 
+func NewEventMeshSecretWithData(name, namespace string, data map[string][]byte) *kcorev1.Secret {
+	return &kcorev1.Secret{
+		ObjectMeta: kmetav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+		Data: data,
+		Type: "Opaque",
+	}
+}
+
 func NewEventMeshSecret(name, namespace string) *kcorev1.Secret {
 	return &kcorev1.Secret{
 		ObjectMeta: kmetav1.ObjectMeta{