Deploy a SASL_PLAIN Kafka listener & test notifications against it

.github/scripts/end2end/configs/notification_destinations.yaml

@@ -23,3 +23,21 @@ spec:
   host: ${NOTIF_KAFKA_HOST}
   port: ${NOTIF_KAFKA_PORT}
   destinationTopic: ${NOTIF_ALT_DEST_TOPIC}
+
+---
+
+apiVersion: zenko.io/v1alpha2
+kind: ZenkoNotificationTarget
+metadata:
+  name: ${NOTIF_AUTH_DEST_NAME}
+  labels:
+    app.kubernetes.io/instance: ${ZENKO_NAME}
+spec:
+  type: kafka
+  host: ${NOTIF_KAFKA_HOST}
+  port: ${NOTIF_KAFKA_AUTH_PORT}
+  destinationTopic: ${NOTIF_AUTH_DEST_TOPIC}
+  auth: basic
+  basic:
+    username: ${NOTIF_AUTH_DEST_USERNAME}
+    password: ${NOTIF_AUTH_DEST_PASSWORD}

.github/scripts/end2end/configure-e2e-ctst.sh

@@ -10,6 +10,27 @@ KAFKA_HOST_PORT=${KAFKA_HOST_PORT:1:-1}
 # Removing the port
 export NOTIF_KAFKA_HOST=${KAFKA_HOST_PORT%:*}
 export NOTIF_KAFKA_PORT=${KAFKA_HOST_PORT#*:}
+export NOTIF_KAFKA_AUTH_PORT=9094
+
+# Add an extra SASL_PLAIN Kafka listener, to support testing authenticated Kafka for bucket notifications
+KAFKA_CLUSTER="${ZENKO_NAME}-base-queue"
+KAFKA_CONFIG=$(kubectl get kafkacluster "$KAFKA_CLUSTER" -o jsonpath='{.spec.readOnlyConfig}')
+KAFKA_CONFIG="$KAFKA_CONFIG"$(cat << EOF
+
+sasl.enabled.mechanisms=PLAIN
+listener.name.auth.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+  username="$NOTIF_AUTH_DEST_USERNAME" \
+  password="$NOTIF_AUTH_DEST_PASSWORD" \
+  user_$NOTIF_AUTH_DEST_USERNAME="$NOTIF_AUTH_DEST_PASSWORD";
+
+EOF
+)
+KAFKA_PATCH=$(jq -n --arg config "$KAFKA_CONFIG" '[
+ {"op": "add", "path": "/spec/listenersConfig/internalListeners/-", "value": {"containerPort": '"$NOTIF_KAFKA_AUTH_PORT"', "name": "auth", "type": "sasl_plaintext", "usedForInnerBrokerCommunication": false}},
+ {"op": "replace", "path": "/spec/readOnlyConfig", "value": $config}
+]')
+kubectl patch kafkacluster "$KAFKA_CLUSTER" --type='json' -p="$KAFKA_PATCH"
+kubectl wait --for=jsonpath='{.status.state}'=ClusterRunning --timeout 10m kafkacluster "$KAFKA_CLUSTER"
 
 UUID=$(kubectl get secret -l app.kubernetes.io/name=backbeat-config,app.kubernetes.io/instance=end2end \
     -o jsonpath='{.items[0].data.config\.json}' | base64 -di | jq .extensions.replication.topic)
@@ -48,6 +69,7 @@ kubectl run kafka-topics \
     --command -- bash -c \
     "kafka-topics.sh --create --topic $NOTIF_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
         kafka-topics.sh --create --topic $NOTIF_ALT_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
+        kafka-topics.sh --create --topic $NOTIF_AUTH_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
         kafka-topics.sh --create --topic $AZURE_ARCHIVE_STATUS_TOPIC --partitions 10 --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
         kafka-topics.sh --create --topic $AZURE_ARCHIVE_STATUS_TOPIC_2_NV --partitions 10 --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
         kafka-topics.sh --create --topic $AZURE_ARCHIVE_STATUS_TOPIC_2_V --partitions 10 --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \

.github/scripts/end2end/configure-e2e.sh

@@ -67,7 +67,8 @@ kubectl run kafka-topics \
     "kafka-topics.sh --create --topic $UUID.backbeat-replication-replay-0 --partitions 5 --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
     kafka-topics.sh --create --topic $UUID.backbeat-data-mover --partitions 5 --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
     kafka-topics.sh --create --topic $NOTIF_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
-    kafka-topics.sh --create --topic $NOTIF_ALT_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists "
+    kafka-topics.sh --create --topic $NOTIF_ALT_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists ; \
+    kafka-topics.sh --create --topic $NOTIF_AUTH_DEST_TOPIC --bootstrap-server $KAFKA_HOST_PORT --if-not-exists "
 
 kubectl run ${POD_NAME} \
   --image ${E2E_IMAGE} \

.github/scripts/end2end/run-e2e-ctst.sh

@@ -93,6 +93,10 @@ WORLD_PARAMETERS="$(jq -c <<EOF
   "NotificationDestinationTopic":"${NOTIF_DEST_TOPIC}",
   "NotificationDestinationAlt":"${NOTIF_ALT_DEST_NAME}",
   "NotificationDestinationTopicAlt":"${NOTIF_ALT_DEST_TOPIC}",
+  "NotificationDestinationAuth":"${NOTIF_AUTH_DEST_NAME}",
+  "NotificationDestinationTopicAuth":"${NOTIF_AUTH_DEST_TOPIC}",
+  "NotificationDestinationAuthUsername":"${NOTIF_AUTH_DEST_USERNAME}",
+  "NotificationDestinationAuthPassword":"${NOTIF_AUTH_DEST_PASSWORD}",
   "KafkaExternalIps": "${KAFKA_EXTERNAL_IP:-}",
   "PrometheusService":"${PROMETHEUS_NAME}-operated.default.svc.cluster.local",
   "KafkaHosts":"${KAFKA_HOST_PORT}",

.github/workflows/end2end.yaml

@@ -95,6 +95,10 @@ env:
   NOTIF_DEST_TOPIC: "destination-topic-1"
   NOTIF_ALT_DEST_NAME: "destination2"
   NOTIF_ALT_DEST_TOPIC: "destination-topic-2"
+  NOTIF_AUTH_DEST_NAME: "destination3"
+  NOTIF_AUTH_DEST_TOPIC: "destination-topic-3"
+  NOTIF_AUTH_DEST_USERNAME: "admin"
+  NOTIF_AUTH_DEST_PASSWORD: "admin-secret"
   SUBDOMAIN: "zenko.local"
   DR_SUBDOMAIN: "dr.zenko.local"
   SKOPEO_PATH: "/tmp"

solution/deps.yaml

@@ -2,11 +2,11 @@
 # to sort keys, use the following command
 # yq eval 'sortKeys(.)' -i deps.yaml
 backbeat:
-  sourceRegistry: ghcr.io/scality
+  sourceRegistry: ghcr.io/scality/playground/delthas
   dashboard: backbeat/backbeat-dashboards
   image: backbeat
   policy: backbeat/backbeat-policies
-  tag: 9.1.2
+  tag: 9.0.17-hcdea8d6b6a6de4e49c07067cafa847a2ae86b8a3
   envsubst: BACKBEAT_TAG
 busybox:
   image: busybox
@@ -136,7 +136,7 @@ vault:
 zenko-operator:
   sourceRegistry: ghcr.io/scality
   image: zenko-operator
-  tag: v1.8.0
+  tag: 6665b3a70724e72e0cec4180129513036fcd9452
   envsubst: ZENKO_OPERATOR_TAG
 zookeeper:
   sourceRegistry: pravega

tests/ctst/steps/notifications.ts

@@ -128,6 +128,16 @@ Given('two notification destinations', function (this: Zenko) {
     this.addToSaved('notificationDestinations', notificationDestinations);
 });
 
+Given('one authenticated notification destination', function (this: Zenko) {
+    const notificationDestinations = [];
+    notificationDestinations.push({
+        destinationName: this.parameters.NotificationDestinationAuth,
+        topic: this.parameters.NotificationDestinationTopicAuth,
+        hosts: this.parameters.KafkaHosts,
+    });
+    this.addToSaved('notificationDestinations', notificationDestinations);
+});
+
 When('i subscribe to {string} notifications for destination {int}',
     async function (this: Zenko, notificationType: string, destination: number) {
         const notificationsPerDestination : Record<string, string[]> = {};

tests/ctst/world/Zenko.ts

@@ -55,6 +55,10 @@ export interface ZenkoWorldParameters extends ClientOptions {
     NotificationDestinationTopic: string;
     NotificationDestinationAlt: string;
     NotificationDestinationTopicAlt: string;
+    NotificationDestinationAuth: string;
+    NotificationDestinationTopicAuth: string;
+    NotificationDestinationAuthUsername: string;
+    NotificationDestinationAuthPassword: string;
     KafkaExternalIps: string;
     KafkaHosts: string;
     PrometheusService: string;