Add radosgw support

PLAN

@@ -1,17 +1,17 @@
 ObjectBucket Notifications Adapter
 ==================================
 
-The adapter will be an HTTP server and/or Kafka consumer running in a pod inside a k8s cluster, configured in noobaa as "bucketNotifications" connection (either as HTTP notifications, or Kafka notifications)
-It will distribute the notifications received from NooBaa to individual triggers, according to the configuration specified by ObjectBucketSource custom resources.
+The adapter will be an HTTP server and/or Kafka consumer running in a pod inside a k8s cluster, configured in noobaa/radosgw as "bucketNotifications" connection (either as HTTP notifications, or Kafka notifications)
+It will distribute the notifications received from NooBaa or radosgw to individual triggers, according to the configuration specified by ObjectBucketSource custom resources.
 The notifications to individual triggers will be sent as cloudevents via HTTP or Kafka
 It will be implemented in go, using k8s client libraries, the cloudevents k8s libraries, aws s3 client library, https://github.com/IBM/sarama, and the operator sdk (https://sdk.operatorframework.io)
 
 NooBaa HTTP connection setup (manual, one-time)
 -----------------------------------------------
 
 Before the adapter can receive bucket notifications via HTTP, it must be registered as an
-HTTP notification connection in NooBaa. This is a manual, one-time setup step
-performed by a cluster administrator.
+HTTP notification connection in NooBaa/radosgw. This is a manual, one-time setup step
+performed by a cluster administrator. Example for Noobaa:
 
 1. Create a Secret in the openshift-storage namespace containing the connection
    details for the adapter. The Secret holds a JSON file describing the HTTP
@@ -53,7 +53,7 @@ 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 (ADAPTER_ID and ADAPTER_TOPIC). When the operator
+identify its connection name (NOOBAA_ADAPTER_ID and NOOBAA_ADAPTER_TOPIC_ARN). 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).
@@ -181,15 +181,15 @@ 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": "$ADAPTER_ID",
+    "Id": "$NOOBAA_ADAPTER_ID",
     "Events": ["s3:ObjectCreated:*"],
-    "Topic": "$ADAPTER_TOPIC"
+    "Topic": "$NOOBAA_ADAPTER_TOPIC_ARN"
   }
 }'
 ```
 
 The Topic in the S3 notification configuration is always a reference to the
-NooBaa connection secret (ADAPTER_TOPIC, e.g. "mcg-adapter-connection/connect.json").
+NooBaa connection secret (NOOBAA_ADAPTER_TOPIC_ARN, 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
@@ -417,7 +417,7 @@ oc patch noobaa noobaa --type='merge' -n openshift-storage -p '{
 ```
 
 4. When the operator calls put-bucket-notification for a bucket, it uses
-   ADAPTER_ID and ADAPTER_TOPIC, e.g.:
+   NOOBAA_ADAPTER_ID and NOOBAA_ADAPTER_TOPIC_ARN, e.g.:
 
 ```
 aws s3api put-bucket-notification --bucket "foo-bucket" \
@@ -431,33 +431,44 @@ aws s3api put-bucket-notification --bucket "foo-bucket" \
 ```
 
 
-Kafka configuration
+Configuration
 -------------------
 
-Kafka configuration is defined via env on the adapter Deployment:
+Configuration is defined via env on the adapter Deployment:
 
-	ADAPTER_ID: identifier for bucket notification configuration
+	NOOBAA_ADAPTER_ID: identifier for NooBaa bucket notification configuration
 	    (default: "mcg-adapter")
-	ADAPTER_TOPIC: NooBaa connection secret reference used as TopicArn in
+	NOOBAA_ADAPTER_TOPIC_ARN: NooBaa connection secret reference used as TopicArn in
 	    put-bucket-notification calls
 	    (default: "mcg-adapter-connection/connect.json")
+	NOOBAA_ADAPTER_STORAGECLASS_PATTERN: A regular expression on OBC
+	    .spec.storageClassName to classify the OBC as a NooBaa-managed OBC
+	    (default: ".*noobaa\.io$")
+	RADOSGW_ADAPTER_ID: identifier for RadosGW bucket notification configuration
+	    (default: "rgw-adapter")
+	RADOSGW_ADAPTER_TOPIC_ARN: RadosGW TopicArn in put-bucket-notification calls
+	    (default: "arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications")
+	RADOSGW_ADAPTER_STORAGECLASS_PATTERN: A regular expression on OBC
+	    .spec.storageClassName to classify the OBC as a ceph/radosgw-managed OBC
+	    (default: ".*ceph-rgw$")
 	NOTIFICATIONS_MODE: "http" or "kafka" — selects whether the adapter
-	    receives NooBaa bucket notifications via its HTTP server or by
+	    receives NooBaa/RadosGW bucket notifications via its HTTP server or by
 	    consuming from a Kafka topic (default: "http")
 	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)
 	KAFKA_SECRET: <name of the secret from the namespace of the adapter
 	    deployment> (required in both modes, same reason as KAFKA_BROKERS)
-	KAFKA_NOTIFICATIONS_TOPIC: the Kafka topic to consume NooBaa bucket
-	    notifications from (e.g. "mcg-adapter-notifications"). Only
-	    required when NOTIFICATIONS_MODE=kafka.
+	KAFKA_NOTIFICATIONS_TOPIC: comma-separated list of the Kafka topics to
+	    consume NooBaa/RadosGW bucket notifications from
+	    (e.g. "mcg-adapter-notifications,rgw-adapter-notifications").
+	    Only required when NOTIFICATIONS_MODE=kafka.
 	KAFKA_NOTIFICATIONS_GROUP_ID: the consumer group ID for consuming
-	    NooBaa notifications (e.g. "mcg-adapter-notifications"). Only
-	    required when NOTIFICATIONS_MODE=kafka.
+	    NooBaa/RadosGW notifications (e.g. "objectbucket-notifications-adapter").
+	    Only required when NOTIFICATIONS_MODE=kafka.
 
 The same Kafka credentials (KAFKA_SECRET) are used for both consuming
-notifications from NooBaa and producing cloudevents to trigger Kafka topics.
+notifications from NooBaa/RadosGW and producing cloudevents to trigger Kafka topics.
 
 The format of the kafka secret is defined in ./kafka-secret-format.md
 

README.md

@@ -1,10 +1,12 @@
 # ObjectBucket Notifications Adapter
 
-A Kubernetes operator that receives S3 bucket notifications from NooBaa (Multicloud Object Gateway) and dispatches them as CloudEvents to configured sink endpoints.
+A Kubernetes operator that receives S3 bucket notifications from NooBaa (Multicloud Object Gateway) and/or Ceph RadosGW and dispatches them as CloudEvents to configured sink endpoints.
 
 ## Description
 
-The ObjectBucket Notifications 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 `ObjectBucketSource` 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 sources and dispatches CloudEvents via HTTP or Kafka to the matching sinks.
+The ObjectBucket Notifications Adapter runs inside a Kubernetes cluster, registered as a `bucketNotifications` connection in NooBaa and/or as an SNS topic in RadosGW. It can receive notifications via HTTP (default) or by consuming from Kafka topics. It watches `ObjectBucketSource` custom resources to determine which S3 events from which buckets should be forwarded to which endpoints. When a notification arrives, the adapter matches the event against all configured sources and dispatches CloudEvents via HTTP or Kafka to the matching sinks.
+
+The adapter determines whether an OBC is managed by NooBaa or RadosGW by matching the OBC's `spec.storageClassName` against configurable regex patterns, and uses the corresponding adapter ID and topic ARN when setting up bucket notifications.
 
 ## Custom Resource: ObjectBucketSource
 
@@ -50,26 +52,30 @@ The adapter is configured via environment variables:
 
 | Variable | Default | Description |
 |---|---|---|
-| `ADAPTER_ID` | `objectbucket-notifications-adapter` | Identifier used in the S3 bucket notification configuration |
-| `ADAPTER_TOPIC` | `objectbucket-notifications-adapter-connection/connect.json` | NooBaa connection secret reference used as TopicArn in put-bucket-notification calls |
+| `NOOBAA_ADAPTER_ID` | `mcg-adapter` | Identifier used in the S3 bucket notification configuration for NooBaa-managed OBCs |
+| `NOOBAA_ADAPTER_TOPIC_ARN` | `mcg-adapter-connection/connect.json` | NooBaa connection secret reference used as TopicArn in put-bucket-notification calls |
+| `NOOBAA_ADAPTER_STORAGECLASS_PATTERN` | `.*noobaa\.io$` | Regex matched against OBC `spec.storageClassName` to classify as NooBaa-managed |
+| `RADOSGW_ADAPTER_ID` | `rgw-adapter` | Identifier used in the S3 bucket notification configuration for RadosGW-managed OBCs |
+| `RADOSGW_ADAPTER_TOPIC_ARN` | `arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications` | RadosGW SNS TopicArn used in put-bucket-notification calls |
+| `RADOSGW_ADAPTER_STORAGECLASS_PATTERN` | `.*ceph-rgw$` | Regex matched against OBC `spec.storageClassName` to classify as RadosGW-managed |
 | `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 |
+| `NOTIFICATIONS_MODE` | `http` | `http` or `kafka` — selects how the adapter receives NooBaa/RadosGW notifications |
 | `KAFKA_BROKERS` | _(none)_ | Comma-separated list of Kafka broker addresses. Required for Kafka sinks 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`. |
+| `KAFKA_NOTIFICATIONS_TOPIC` | _(none)_ | Comma-separated list of Kafka topics to consume NooBaa/RadosGW notifications from (e.g. `mcg-adapter-notifications,rgw-adapter-notifications`). Required when `NOTIFICATIONS_MODE=kafka`. |
+| `KAFKA_NOTIFICATIONS_GROUP_ID` | _(none)_ | Consumer group ID for consuming NooBaa/RadosGW notifications. Required when `NOTIFICATIONS_MODE=kafka`. |
 
 ### NooBaa Connection Setup (HTTP mode)
 
-Before the adapter can receive notifications via HTTP, 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). A helper script `noobaa-connection-setup.sh` is provided. Manual steps:
 
 1. Create the connection secret:
 
 ```sh
-oc create secret generic objectbucket-notifications-adapter-connection \
+oc create secret generic mcg-adapter-connection \
   --from-file=connect.json=/dev/stdin -n openshift-storage <<EOF
 {
-  "name": "objectbucket-notifications-adapter-connection",
+  "name": "mcg-adapter-connection",
   "notification_protocol": "http",
   "agent_request_object": {
     "host": "<adapter-service>.<adapter-namespace>.svc.cluster.local",
@@ -86,7 +92,7 @@ existing_connections=$(oc get noobaa noobaa -n openshift-storage -o json \
   | jq -c '.spec.bucketNotifications.connections // []')
 
 updated_connections=$(echo "$existing_connections" | jq -c \
-  --arg name "objectbucket-notifications-adapter-connection" \
+  --arg name "mcg-adapter-connection" \
   '[.[] | select(.name != $name)] + [{"name": $name, "namespace": "openshift-storage"}]')
 
 oc patch noobaa noobaa --type='merge' -n openshift-storage -p '{
@@ -101,15 +107,25 @@ 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.
+As an alternative to HTTP, the adapter can consume NooBaa notifications from a Kafka topic. This requires a Strimzi-managed Kafka cluster. A helper script `noobaa-kafka-connection-setup.sh` is provided. See the `PLAN` file for detailed steps.
 
 Summary of the one-time setup:
 
-1. Create a `KafkaTopic` (e.g. `objectbucket-notifications-adapter-notifications`) and `KafkaUser` resources in Strimzi
-2. Create a NooBaa connection secret with `notification_protocol: kafka` pointing to the Kafka bootstrap and topic
+1. Create a `KafkaTopic` (e.g. `mcg-adapter-notifications`) and `KafkaUser` resources in Strimzi
+2. Create a NooBaa connection secret (`mcg-adapter-connection`) 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`)
 
+### RadosGW Connection Setup (Kafka mode)
+
+For Ceph RadosGW, bucket notifications are delivered via Kafka using SNS topics. A helper script `rook-kafka-connection-setup.sh` is provided. See the `PLAN` file for detailed steps.
+
+Summary of the one-time setup:
+
+1. Create a `KafkaTopic` (e.g. `rgw-adapter-notifications`) and `KafkaUser` resources in Strimzi
+2. Create an SNS topic in RadosGW pointing to the Kafka broker and topic
+3. Create a Kubernetes Secret in the adapter's namespace with the adapter's Kafka credentials (see `kafka-secret-format.md`)
+
 ### Deploy
 
 ```sh

cmd/main.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -212,14 +213,34 @@ func main() {
 		os.Exit(1)
 	}
 
-	adapterID := os.Getenv("ADAPTER_ID")
-	if adapterID == "" {
-		adapterID = "objectbucket-notifications-adapter"
-	}
-	adapterTopic := os.Getenv("ADAPTER_TOPIC")
-	if adapterTopic == "" {
-		adapterTopic = "objectbucket-notifications-adapter-connection/connect.json"
+	noobaaAdapterID := envOrDefault("NOOBAA_ADAPTER_ID", "mcg-adapter")
+	noobaaAdapterTopic := envOrDefault("NOOBAA_ADAPTER_TOPIC_ARN", "mcg-adapter-connection/connect.json")
+	noobaaStorageClassPattern := envOrDefault("NOOBAA_ADAPTER_STORAGECLASS_PATTERN", `.*noobaa\.io$`)
+
+	radosgwAdapterID := envOrDefault("RADOSGW_ADAPTER_ID", "rgw-adapter")
+	radosgwAdapterTopic := envOrDefault("RADOSGW_ADAPTER_TOPIC_ARN",
+		"arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications")
+	radosgwStorageClassPattern := envOrDefault("RADOSGW_ADAPTER_STORAGECLASS_PATTERN", `.*ceph-rgw$`)
+
+	adapterConfigs := make([]controller.AdapterConfig, 0, 2)
+	for _, cfg := range []struct {
+		id, topic, pattern string
+	}{
+		{noobaaAdapterID, noobaaAdapterTopic, noobaaStorageClassPattern},
+		{radosgwAdapterID, radosgwAdapterTopic, radosgwStorageClassPattern},
+	} {
+		re, err := regexp.Compile(cfg.pattern)
+		if err != nil {
+			setupLog.Error(err, "invalid storageclass pattern", "pattern", cfg.pattern)
+			os.Exit(1)
+		}
+		adapterConfigs = append(adapterConfigs, controller.AdapterConfig{
+			ID:                  cfg.id,
+			Topic:               cfg.topic,
+			StorageClassPattern: re,
+		})
 	}
+
 	adapterPort := 8888
 	if portStr := os.Getenv("ADAPTER_PORT"); portStr != "" {
 		var err error
@@ -238,7 +259,14 @@ func main() {
 		os.Exit(1)
 	}
 
-	kafkaNotificationsTopic := os.Getenv("KAFKA_NOTIFICATIONS_TOPIC")
+	var kafkaNotificationsTopics []string
+	if topicsStr := os.Getenv("KAFKA_NOTIFICATIONS_TOPIC"); topicsStr != "" {
+		for _, t := range strings.Split(topicsStr, ",") {
+			if trimmed := strings.TrimSpace(t); trimmed != "" {
+				kafkaNotificationsTopics = append(kafkaNotificationsTopics, trimmed)
+			}
+		}
+	}
 	kafkaNotificationsGroupID := os.Getenv("KAFKA_NOTIFICATIONS_GROUP_ID")
 
 	var kafkaBrokers []string
@@ -283,18 +311,17 @@ func main() {
 	}
 
 	if err := (&controller.ObjectBucketSourceReconciler{
-		Client:       mgr.GetClient(),
-		Scheme:       mgr.GetScheme(),
-		AdapterID:    adapterID,
-		AdapterTopic: adapterTopic,
+		Client:         mgr.GetClient(),
+		Scheme:         mgr.GetScheme(),
+		AdapterConfigs: adapterConfigs,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ObjectBucketSource")
 		os.Exit(1)
 	}
 	// +kubebuilder:scaffold:builder
 
 	if notificationsMode == "kafka" {
-		if kafkaNotificationsTopic == "" {
+		if len(kafkaNotificationsTopics) == 0 {
 			setupLog.Error(fmt.Errorf("KAFKA_NOTIFICATIONS_TOPIC is required when NOTIFICATIONS_MODE=kafka"), "missing env")
 			os.Exit(1)
 		}
@@ -314,7 +341,7 @@ func main() {
 		KafkaBrokers:              kafkaBrokers,
 		KafkaConfig:               kafkaCfg,
 		NotificationsMode:         notificationsMode,
-		KafkaNotificationsTopic:   kafkaNotificationsTopic,
+		KafkaNotificationsTopics:  kafkaNotificationsTopics,
 		KafkaNotificationsGroupID: kafkaNotificationsGroupID,
 	}
 	if err := mgr.Add(notifServer); err != nil {
@@ -353,3 +380,10 @@ func main() {
 		os.Exit(1)
 	}
 }
+
+func envOrDefault(key, defaultValue string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return defaultValue
+}

config/kafka/kafka_env_patch.yaml

@@ -15,6 +15,6 @@ spec:
         - name: KAFKA_SECRET
           value: "objectbucket-notifications-adapter-user"
         - name: KAFKA_NOTIFICATIONS_TOPIC
-          value: "objectbucket-notifications-adapter-notifications"
+          value: "mcg-adapter-notifications,rgw-adapter-notifications"
         - name: KAFKA_NOTIFICATIONS_GROUP_ID
-          value: "objectbucket-notifications-adapter-notifications"
+          value: "objectbucket-notifications-adapter"

config/manager/manager.yaml

@@ -66,10 +66,18 @@ spec:
         image: controller:latest
         name: manager
         env:
-        - name: ADAPTER_ID
-          value: "objectbucket-notifications-adapter"
-        - name: ADAPTER_TOPIC
-          value: "objectbucket-notifications-adapter-connection/connect.json"
+        - name: NOOBAA_ADAPTER_ID
+          value: "mcg-adapter"
+        - name: NOOBAA_ADAPTER_TOPIC_ARN
+          value: "mcg-adapter-connection/connect.json"
+        - name: NOOBAA_ADAPTER_STORAGECLASS_PATTERN
+          value: ".*noobaa\\.io$"
+        - name: RADOSGW_ADAPTER_ID
+          value: "rgw-adapter"
+        - name: RADOSGW_ADAPTER_TOPIC_ARN
+          value: "arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications"
+        - name: RADOSGW_ADAPTER_STORAGECLASS_PATTERN
+          value: ".*ceph-rgw$"
         - name: ADAPTER_PORT
           value: "8888"
         - name: NOTIFICATIONS_MODE