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_AUTH_HOST}
+  port: ${NOTIF_KAFKA_AUTH_PORT}
+  destinationTopic: ${NOTIF_AUTH_DEST_TOPIC}
+  auth: plain
+  basic:
+    username: ${NOTIF_AUTH_DEST_USERNAME}
+    password: ${NOTIF_AUTH_DEST_PASSWORD}

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

@@ -11,6 +11,37 @@ KAFKA_HOST_PORT=${KAFKA_HOST_PORT:1:-1}
 export NOTIF_KAFKA_HOST=${KAFKA_HOST_PORT%:*}
 export NOTIF_KAFKA_PORT=${KAFKA_HOST_PORT#*:}
 
+export NOTIF_KAFKA_AUTH_HOST="${ZENKO_NAME}-base-queue-auth-0"
+export NOTIF_KAFKA_AUTH_HOST_PORT="$NOTIF_KAFKA_AUTH_HOST:$NOTIF_KAFKA_PORT"
+export NOTIF_KAFKA_AUTH_PORT=9094
+
+# Add an extra SASL_PLAIN Kafka listener, to support testing authenticated Kafka for bucket notifications
+kubectl get zookeepercluster "${ZENKO_NAME}-base-quorum" -o json | jq '.
+| .metadata |= {namespace, name: "\(.name)-auth" }
+| del(.spec.labels)
+| del(.spec.persistence)
+| .spec.storageType |= "ephemeral"
+| del(.spec.pod.affinity)
+| del(.spec.pod.labels)
+| del(.status)
+' | kubectl apply -f -
+kubectl wait --for=jsonpath='{.status.readyReplicas}'=1 --timeout 10m zookeepercluster "${ZENKO_NAME}-base-quorum-auth"
+
+kubectl get kafkacluster "${ZENKO_NAME}-base-queue" -o json | jq '.
+| .metadata |= {namespace, name: "\(.name)-auth" }
+| del(.status)
+| .spec.listenersConfig.internalListeners |= . + [{containerPort: 9094, name: "auth", type: "sasl_plaintext", usedForInnerBrokerCommunication: false}]
+| .spec.readOnlyConfig |= (. + "
+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"'\";
+")
+| del(.spec.brokerConfigGroups.default.storageConfigs[].pvcSpec)
+| .spec.brokerConfigGroups.default.storageConfigs[].emptyDir |= {medium: "Memory"}
+| .spec.zkAddresses |= ["'"${ZENKO_NAME}-base-quorum-auth-0.${ZENKO_NAME}-base-quorum-auth-headless:2181"'"]
+| .spec.cruiseControlConfig.capacityConfig |= ({brokerCapacities: [{brokerId: "-1", capacity: {DISK: "1000", CPU: "100", NW_IN: "1000", NW_OUT: "1000"}}]} | tostring)
+' | kubectl apply -f -
+kubectl wait --for=jsonpath='{.status.state}'=ClusterRunning --timeout 10m kafkacluster "${ZENKO_NAME}-base-queue-auth"
+
 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)
 UUID=${UUID%.*}
@@ -48,6 +79,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 $NOTIF_KAFKA_AUTH_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,7 @@ 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"
 
 kubectl run ${POD_NAME} \
   --image ${E2E_IMAGE} \

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

@@ -59,6 +59,10 @@ DR_ADMIN_SECRET_ACCESS_KEY=$(kubectl get secret end2end-pra-management-vault-adm
 KAFKA_HOST_PORT=$(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 .kafka.hosts)
 KAFKA_HOST_PORT=${KAFKA_HOST_PORT:1:-1}
+KAFKA_PORT=${KAFKA_HOST_PORT#*:}
+
+KAFKA_AUTH_HOST="end2end-base-queue-auth-0"
+KAFKA_AUTH_HOST_PORT="$KAFKA_AUTH_HOST:$KAFKA_PORT"
 
 TIME_PROGRESSION_FACTOR=$(kubectl get zenko ${ZENKO_NAME} -o jsonpath="{.metadata.annotations.zenko\.io/time-progression-factor}")
 INSTANCE_ID=$(kubectl get zenko ${ZENKO_NAME} -o jsonpath='{.status.instanceID}')
@@ -93,9 +97,14 @@ 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}",
+  "KafkaAuthHosts":"${KAFKA_AUTH_HOST_PORT}",
   "KeycloakUsername":"${KEYCLOAK_TEST_USER}",
   "KeycloakPassword":"${KEYCLOAK_TEST_PASSWORD}",
   "KeycloakHost":"${KEYCLOAK_TEST_HOST}",

.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"
@@ -400,7 +404,7 @@ jobs:
           cache-dependency-path: tests/ctst/yarn.lock
       - name: Install ctst test dependencies
         working-directory: tests/ctst
-        run: yarn install
+        run: yarn install --mutex network || (yarn cache clean && yarn install --mutex network)
       - name: Lint ctst tests
         working-directory: tests/ctst
         run: yarn lint

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/features/bucket-notifications/notifications.feature

@@ -50,7 +50,7 @@ Feature: Bucket notifications
   @PreMerge
   @Flaky
   @BucketNotification
-  Scenario Outline: Recieve notification for configured events
+  Scenario Outline: Receive notification for configured events
     Given a "<versioningConfiguration>" bucket
     And one notification destination
     When i subscribe to "<subscribedNotificationType>" notifications for destination <destination>
@@ -94,6 +94,23 @@ Feature: Bucket notifications
       |   Versioning suspended  |                  s3:ObjectTagging:Delete |                  s3:ObjectTagging:Delete | without |      filter |       receive |           0 |
       |   Versioning suspended  |                         s3:ObjectAcl:Put |                         s3:ObjectAcl:Put | without |      filter |       receive |           0 |
 
+  @2.6.0
+  @PreMerge
+  @Flaky
+  @BucketNotification
+  Scenario Outline: Receive notification for configured events in authenticated notification destinations
+    Given a "<versioningConfiguration>" bucket
+    And one authenticated notification destination
+    When i subscribe to "<subscribedNotificationType>" notifications for destination <destination>
+    And a "<notificationType>" event is triggered "<enable>" "<filterType>"
+    Then i should "<shouldReceive>" a notification for "<notificationType>" event in destination <destination>
+
+    Examples:
+      | versioningConfiguration |               subscribedNotificationType |                         notificationType |  enable | filterType  | shouldReceive | destination |
+      |           Non versioned |                       s3:ObjectCreated:* |                     s3:ObjectCreated:Put | without |      filter |       receive |           0 |
+      |               Versioned |                       s3:ObjectCreated:* |                    s3:ObjectCreated:Copy | without |      filter |       receive |           0 |
+      |   Versioning suspended  |                       s3:ObjectCreated:* |                     s3:ObjectCreated:Put | without |      filter |       receive |           0 |
+
   @2.6.0
   @PreMerge
   @BucketNotification
@@ -130,7 +147,7 @@ Feature: Bucket notifications
   @PreMerge
   @Flaky
   @BucketNotification
-  Scenario Outline: Recieve notification for configured events with correct filter
+  Scenario Outline: Receive notification for configured events with correct filter
     Given a "<versioningConfiguration>" bucket
     And one notification destination
     When i subscribe to "<notificationType>" notifications for destination <destination> with "<filterType>" filter
@@ -168,7 +185,7 @@ Feature: Bucket notifications
   @PreMerge
   @Flaky
   @BucketNotification
-  Scenario Outline: Recieve notification in multiple destinations
+  Scenario Outline: Receive notification in multiple destinations
     Given a "<versioningConfiguration>" bucket
     And two notification destinations
     When i subscribe to "<subscribedNotificationType>" notifications for destination <destination>

tests/ctst/steps/notifications.ts

@@ -103,14 +103,32 @@ async function putAcl(world: Zenko, objName: string) {
     await S3.putObjectAcl(world.getCommandParameters());
 }
 
-Given('one notification destination', function (this: Zenko) {
+function setNotificationDestination(world: Zenko, destination: string, topic: string, hosts: string) {
     const notificationDestinations = [];
     notificationDestinations.push({
-        destinationName: this.parameters.NotificationDestination,
-        topic: this.parameters.NotificationDestinationTopic,
-        hosts: this.parameters.KafkaHosts,
+        destinationName: destination,
+        topic,
+        hosts,
     });
-    this.addToSaved('notificationDestinations', notificationDestinations);
+    world.addToSaved('notificationDestinations', notificationDestinations);
+}
+
+Given('one notification destination', function (this: Zenko) {
+    setNotificationDestination(
+        this,
+        this.parameters.NotificationDestination,
+        this.parameters.NotificationDestinationTopic,
+        this.parameters.KafkaHosts,
+    );
+});
+
+Given('one authenticated notification destination', function (this: Zenko) {
+    setNotificationDestination(
+        this,
+        this.parameters.NotificationDestinationAuth,
+        this.parameters.NotificationDestinationTopicAuth,
+        this.parameters.KafkaAuthHosts,
+    );
 });
 
 Given('two notification destinations', function (this: Zenko) {

tests/ctst/world/Zenko.ts

@@ -55,8 +55,13 @@ export interface ZenkoWorldParameters extends ClientOptions {
     NotificationDestinationTopic: string;
     NotificationDestinationAlt: string;
     NotificationDestinationTopicAlt: string;
+    NotificationDestinationAuth: string;
+    NotificationDestinationTopicAuth: string;
+    NotificationDestinationAuthUsername: string;
+    NotificationDestinationAuthPassword: string;
     KafkaExternalIps: string;
     KafkaHosts: string;
+    KafkaAuthHosts: string;
     PrometheusService: string;
     KeycloakUsername: string;
     KeycloakPassword: string;