implement kafka consumer for noobaa events

Author: maschmid

Makefile

@@ -216,6 +216,11 @@ deploy: manifests kustomize ## Deploy controller to the K8s cluster specified in
 	cd config/manager && $(KUSTOMIZE) edit set image controller=${IMG}
 	$(KUSTOMIZE) build config/default | $(KUBECTL) apply -f -
 
+.PHONY: deploy-kafka
+deploy-kafka: manifests kustomize ## Deploy controller in Kafka notifications mode to the K8s cluster specified in ~/.kube/config.
+	cd config/manager && $(KUSTOMIZE) edit set image controller=${IMG}
+	$(KUSTOMIZE) build config/kafka | $(KUBECTL) apply -f -
+
 .PHONY: undeploy
 undeploy: kustomize ## Undeploy controller from the K8s cluster specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource not found errors during deletion.
 	$(KUSTOMIZE) build config/default | $(KUBECTL) delete --ignore-not-found=$(ignore-not-found) -f -

PLAN

@@ -53,10 +53,10 @@ oc patch noobaa noobaa --type='merge' -n openshift-storage -p '{
 ```
 
 The adapter's deployment should be configured with environment variables that
-identify its connection name (see NOOBAA_CONNECTION_NAME in the Kafka
-configuration section). When the operator calls put-bucket-notification for a
-bucket, it uses this connection name so that NooBaa knows where to deliver
-notifications (HTTP endpoint or Kafka topic, depending on NOTIFICATIONS_MODE).
+identify its connection name (ADAPTER_ID and ADAPTER_TOPIC). When the operator
+calls put-bucket-notification for a bucket, it uses these values so that NooBaa
+knows where to deliver notifications (HTTP endpoint or Kafka topic, depending on
+NOTIFICATIONS_MODE).
 
 Internal MCG/NooBaa trigger CRD
 -------------------------------
@@ -166,22 +166,19 @@ and use those connection details and keys to issue the request to noobaa using S
 ```
 aws s3api put-bucket-notification --bucket "foo-bucket" --notification-configuration '{
   "TopicConfiguration": {
-    "Id": "$NOOBAA_CONNECTION_NAME",
+    "Id": "$ADAPTER_ID",
     "Events": ["s3:ObjectCreated:*"],
-    "Topic": "$NOOBAA_CONNECTION_NAME/connect.json"
+    "Topic": "$ADAPTER_TOPIC"
   }
 }'
 ```
 
 The Topic in the S3 notification configuration is always a reference to the
-NooBaa connection secret: "NOOBAA_CONNECTION_NAME/connect.json". It is not the
-Kafka topic itself — for Kafka mode, the actual Kafka topic is the "topic" field
-inside the connect.json (see the NooBaa Kafka connection setup section). For HTTP
-mode, the connect.json contains the HTTP endpoint details instead.
-
-NOOBAA_CONNECTION_NAME is an env var on the adapter deployment that identifies
-which NooBaa connection was configured during the one-time setup (HTTP or Kafka,
-depending on NOTIFICATIONS_MODE).
+NooBaa connection secret (ADAPTER_TOPIC, e.g. "mcg-adapter-connection/connect.json").
+It is not the Kafka topic itself — for Kafka mode, the actual Kafka topic is the
+"topic" field inside the connect.json (see the NooBaa Kafka connection setup
+section). For HTTP mode, the connect.json contains the HTTP endpoint details
+instead.
 
 If successful, the BucketNotificationSet condition will be marked as "True"
 
@@ -404,14 +401,14 @@ oc patch noobaa noobaa --type='merge' -n openshift-storage -p '{
 }'
 ```
 
-4. When the operator calls put-bucket-notification for a bucket, it uses the
-   Kafka connection name as the Topic prefix, e.g.:
+4. When the operator calls put-bucket-notification for a bucket, it uses
+   ADAPTER_ID and ADAPTER_TOPIC, e.g.:
 
 ```
 aws s3api put-bucket-notification --bucket "foo-bucket" \
   --notification-configuration '{
     "TopicConfiguration": {
-      "Id": "mcg-adapter-connection",
+      "Id": "mcg-adapter",
       "Events": ["s3:ObjectCreated:*"],
       "Topic": "mcg-adapter-connection/connect.json"
     }
@@ -424,14 +421,14 @@ Kafka configuration
 
 Kafka configuration is defined via env on the adapter Deployment:
 
+	ADAPTER_ID: identifier for bucket notification configuration
+	    (default: "mcg-adapter")
+	ADAPTER_TOPIC: NooBaa connection secret reference used as TopicArn in
+	    put-bucket-notification calls
+	    (default: "mcg-adapter-connection/connect.json")
 	NOTIFICATIONS_MODE: "http" or "kafka" — selects whether the adapter
 	    receives NooBaa bucket notifications via its HTTP server or by
 	    consuming from a Kafka topic (default: "http")
-	NOOBAA_CONNECTION_NAME: the name of the NooBaa connection configured
-	    during the one-time setup (e.g. "mcg-adapter-connection"). Used as
-	    the Id and Topic prefix in put-bucket-notification calls. The same
-	    name is used regardless of notifications mode — only one mode is
-	    active at a time.
 	KAFKA_BROKERS: comma-separated list of host:port (required in both
 	    modes — even in HTTP mode the adapter needs Kafka to produce
 	    cloudevents to trigger Kafka topics)

README.md

@@ -4,7 +4,7 @@ A Kubernetes operator that receives S3 bucket notifications from NooBaa (Multicl
 
 ## Description
 
-The MCG Adapter runs as an HTTP server inside a Kubernetes cluster, registered as a `bucketNotifications` connection in NooBaa. It watches `MCGOBCTrigger` custom resources to determine which S3 events from which buckets should be forwarded to which endpoints. When NooBaa delivers a notification, the adapter matches the event against all configured triggers and dispatches CloudEvents via HTTP or Kafka to the matching endpoints.
+The MCG Adapter runs inside a Kubernetes cluster, registered as a `bucketNotifications` connection in NooBaa. It can receive notifications via HTTP (default) or by consuming from a Kafka topic. It watches `MCGOBCTrigger` custom resources to determine which S3 events from which buckets should be forwarded to which endpoints. When NooBaa delivers a notification, the adapter matches the event against all configured triggers and dispatches CloudEvents via HTTP or Kafka to the matching endpoints.
 
 ## Custom Resource: MCGOBCTrigger
 
@@ -54,13 +54,17 @@ The adapter is configured via environment variables:
 | Variable | Default | Description |
 |---|---|---|
 | `ADAPTER_ID` | `mcg-adapter` | Identifier used in the S3 bucket notification configuration |
-| `ADAPTER_TOPIC` | `mcg-adapter-connection` | Topic/connection name registered in NooBaa |
-| `ADAPTER_PORT` | `8888` | Port the notification HTTP server listens on |
-| `KAFKA_BROKERS` | _(none)_ | Comma-separated list of Kafka broker addresses (e.g. `broker1:9092,broker2:9092`). Required when using Kafka triggers. |
+| `ADAPTER_TOPIC` | `mcg-adapter-connection/connect.json` | NooBaa connection secret reference used as TopicArn in put-bucket-notification calls |
+| `ADAPTER_PORT` | `8888` | Port the notification HTTP server listens on (HTTP mode only) |
+| `NOTIFICATIONS_MODE` | `http` | `http` or `kafka` — selects how the adapter receives NooBaa notifications |
+| `KAFKA_BROKERS` | _(none)_ | Comma-separated list of Kafka broker addresses. Required for Kafka triggers and when `NOTIFICATIONS_MODE=kafka`. |
+| `KAFKA_SECRET` | _(none)_ | Name of a Kubernetes Secret (in the adapter's namespace) containing Kafka credentials. See `kafka-secret-format.md`. |
+| `KAFKA_NOTIFICATIONS_TOPIC` | _(none)_ | Kafka topic to consume NooBaa notifications from. Required when `NOTIFICATIONS_MODE=kafka`. |
+| `KAFKA_NOTIFICATIONS_GROUP_ID` | _(none)_ | Consumer group ID for consuming NooBaa notifications. Required when `NOTIFICATIONS_MODE=kafka`. |
 
-### NooBaa Connection Setup
+### NooBaa Connection Setup (HTTP mode)
 
-Before the adapter can receive notifications, register it as an HTTP connection in NooBaa (one-time setup):
+Before the adapter can receive notifications via HTTP, register it as an HTTP connection in NooBaa (one-time setup):
 
 1. Create the connection secret:
 
@@ -98,6 +102,17 @@ oc patch noobaa noobaa --type=’merge’ -n openshift-storage -p ‘{
 }’
 ```
 
+### NooBaa Connection Setup (Kafka mode)
+
+As an alternative to HTTP, the adapter can consume NooBaa notifications from a Kafka topic. This requires a Strimzi-managed Kafka cluster. See the `PLAN` file for detailed setup steps, including creating the KafkaTopic, KafkaUsers, and the NooBaa Kafka connection secret.
+
+Summary of the one-time setup:
+
+1. Create a `KafkaTopic` (e.g. `mcg-adapter-notifications`) and `KafkaUser` resources in Strimzi
+2. Create a NooBaa connection secret with `notification_protocol: kafka` pointing to the Kafka bootstrap and topic
+3. Patch the NooBaa CR to register the Kafka connection
+4. Create a Kubernetes Secret in the adapter’s namespace with the adapter’s Kafka credentials (see `kafka-secret-format.md`)
+
 ### Deploy
 
 ```sh
@@ -106,6 +121,14 @@ make install
 make deploy IMG=<some-registry>/mcg-adapter:tag
 ```
 
+To deploy in Kafka notifications mode (using default values from the PLAN):
+
+```sh
+make deploy-kafka IMG=<some-registry>/mcg-adapter:tag
+```
+
+The `deploy-kafka` target uses the kustomize overlay in `config/kafka/` which sets `NOTIFICATIONS_MODE=kafka` along with default Kafka broker, topic, and secret values. Edit `config/kafka/kafka_env_patch.yaml` to customize these for your cluster.
+
 ### Run Locally (development)
 
 ```sh

cmd/main.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/tls"
 	"flag"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -228,6 +229,18 @@ func main() {
 			os.Exit(1)
 		}
 	}
+	notificationsMode := os.Getenv("NOTIFICATIONS_MODE")
+	if notificationsMode == "" {
+		notificationsMode = "http"
+	}
+	if notificationsMode != "http" && notificationsMode != "kafka" {
+		setupLog.Error(fmt.Errorf("invalid NOTIFICATIONS_MODE %q", notificationsMode), "must be \"http\" or \"kafka\"")
+		os.Exit(1)
+	}
+
+	kafkaNotificationsTopic := os.Getenv("KAFKA_NOTIFICATIONS_TOPIC")
+	kafkaNotificationsGroupID := os.Getenv("KAFKA_NOTIFICATIONS_GROUP_ID")
+
 	var kafkaBrokers []string
 	if brokersStr := os.Getenv("KAFKA_BROKERS"); brokersStr != "" {
 		kafkaBrokers = strings.Split(brokersStr, ",")
@@ -280,11 +293,29 @@ func main() {
 	}
 	// +kubebuilder:scaffold:builder
 
+	if notificationsMode == "kafka" {
+		if kafkaNotificationsTopic == "" {
+			setupLog.Error(fmt.Errorf("KAFKA_NOTIFICATIONS_TOPIC is required when NOTIFICATIONS_MODE=kafka"), "missing env")
+			os.Exit(1)
+		}
+		if kafkaNotificationsGroupID == "" {
+			setupLog.Error(fmt.Errorf("KAFKA_NOTIFICATIONS_GROUP_ID is required when NOTIFICATIONS_MODE=kafka"), "missing env")
+			os.Exit(1)
+		}
+		if len(kafkaBrokers) == 0 {
+			setupLog.Error(fmt.Errorf("KAFKA_BROKERS is required when NOTIFICATIONS_MODE=kafka"), "missing env")
+			os.Exit(1)
+		}
+	}
+
 	notifServer := &notificationserver.NotificationServer{
-		Client:       mgr.GetClient(),
-		Port:         adapterPort,
-		KafkaBrokers: kafkaBrokers,
-		KafkaConfig:  kafkaCfg,
+		Client:                    mgr.GetClient(),
+		Port:                      adapterPort,
+		KafkaBrokers:              kafkaBrokers,
+		KafkaConfig:               kafkaCfg,
+		NotificationsMode:         notificationsMode,
+		KafkaNotificationsTopic:   kafkaNotificationsTopic,
+		KafkaNotificationsGroupID: kafkaNotificationsGroupID,
 	}
 	if err := mgr.Add(notifServer); err != nil {
 		setupLog.Error(err, "unable to add notification server")

config/kafka/kafka_env_patch.yaml

@@ -0,0 +1,20 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: controller-manager
+spec:
+  template:
+    spec:
+      containers:
+      - name: manager
+        env:
+        - name: NOTIFICATIONS_MODE
+          value: "kafka"
+        - name: KAFKA_BROKERS
+          value: "my-cluster-kafka-bootstrap.kafka.svc:9095"
+        - name: KAFKA_SECRET
+          value: "mcg-adapter-user"
+        - name: KAFKA_NOTIFICATIONS_TOPIC
+          value: "mcg-adapter-notifications"
+        - name: KAFKA_NOTIFICATIONS_GROUP_ID
+          value: "mcg-adapter-notifications"

config/kafka/kustomization.yaml

@@ -0,0 +1,7 @@
+resources:
+- ../default
+
+patches:
+- path: kafka_env_patch.yaml
+  target:
+    kind: Deployment

config/manager/kustomization.yaml

@@ -5,5 +5,5 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 images:
 - name: controller
-  newName: example.com/mcg-adapter
-  newTag: v0.0.1
+  newName: quay.io/maschmid/mcg-adapter
+  newTag: latest

config/manager/manager.yaml

@@ -72,6 +72,8 @@ spec:
           value: "mcg-adapter-connection/connect.json"
         - name: ADAPTER_PORT
           value: "8888"
+        - name: NOTIFICATIONS_MODE
+          value: "http"
         ports:
         - containerPort: 8888
           name: notifications

internal/notificationserver/consumer.go

@@ -0,0 +1,38 @@
+package notificationserver
+
+import (
+	"github.com/IBM/sarama"
+)
+
+type consumerGroupHandler struct {
+	handler *notificationHandler
+}
+
+func (h *consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error {
+	log.Info("kafka consumer group session setup")
+	return nil
+}
+
+func (h *consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error {
+	log.Info("kafka consumer group session cleanup")
+	return nil
+}
+
+func (h *consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		log.Info("received kafka notification",
+			"topic", msg.Topic,
+			"partition", msg.Partition,
+			"offset", msg.Offset)
+
+		if err := h.handler.processNotification(session.Context(), msg.Value); err != nil {
+			log.Error(err, "processing kafka notification",
+				"topic", msg.Topic,
+				"partition", msg.Partition,
+				"offset", msg.Offset)
+		}
+
+		session.MarkMessage(msg, "")
+	}
+	return nil
+}

internal/notificationserver/handler.go

@@ -61,14 +61,20 @@ func (h *notificationHandler) handleNotification(w http.ResponseWriter, r *http.
 	}
 	defer func() { _ = r.Body.Close() }()
 
-	var payload notificationPayload
-	if err := json.Unmarshal(body, &payload); err != nil {
-		log.Error(err, "parsing notification payload")
+	if err := h.processNotification(r.Context(), body); err != nil {
+		log.Error(err, "processing notification")
 		http.Error(w, "bad request", http.StatusBadRequest)
 		return
 	}
 
-	ctx := r.Context()
+	w.WriteHeader(http.StatusOK)
+}
+
+func (h *notificationHandler) processNotification(ctx context.Context, body []byte) error {
+	var payload notificationPayload
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return fmt.Errorf("parsing notification payload: %w", err)
+	}
 
 	var dataRecords []parsedRecord
 	for _, rawRecord := range payload.Records {
@@ -89,7 +95,7 @@ func (h *notificationHandler) handleNotification(w http.ResponseWriter, r *http.
 		h.dispatchDataEvents(ctx, dataRecords)
 	}
 
-	w.WriteHeader(http.StatusOK)
+	return nil
 }
 
 func (h *notificationHandler) handleTestEvent(ctx context.Context, bucketName string) {