From 2bf90193e8c8503f7f5b23050580da5d7031a9b3 Mon Sep 17 00:00:00 2001 From: Marek Schmidt Date: Thu, 13 Aug 2026 15:15:15 +0200 Subject: [PATCH 1/2] configue objectbucketsource via configmap --- .../main.go | 183 +++------ ...ctbucket-notifications-adapter-config.yaml | 27 ++ ...ket-notifications-adapter-configuration.md | 192 ++++++++++ internal/objectbucketsource/config/config.go | 357 ++++++++++++++++++ .../objectbucketsource/config/config_test.go | 129 +++++++ .../objectbucketsource/config/interface.go | 9 + internal/objectbucketsource/config/mock.go | 89 +++++ .../objectbucketsource_controller.go | 29 +- .../objectbucketsource_controller_test.go | 6 +- .../notificationserver/server.go | 138 ++++++- 10 files changed, 1010 insertions(+), 149 deletions(-) create mode 100644 config/samples/objectbucket-notifications-adapter-config.yaml create mode 100644 docs/objectbucket-notifications-adapter-configuration.md create mode 100644 internal/objectbucketsource/config/config.go create mode 100644 internal/objectbucketsource/config/config_test.go create mode 100644 internal/objectbucketsource/config/interface.go create mode 100644 internal/objectbucketsource/config/mock.go diff --git a/cmd/objectbucket-notifications-adapter/main.go b/cmd/objectbucket-notifications-adapter/main.go index b547458..bace774 100644 --- a/cmd/objectbucket-notifications-adapter/main.go +++ b/cmd/objectbucket-notifications-adapter/main.go @@ -20,21 +20,16 @@ import ( "context" "crypto/tls" "flag" - "fmt" "os" "path/filepath" - "regexp" - "strconv" "strings" // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.) // to ensure that exec-entrypoint and run can make use of them. _ "k8s.io/client-go/plugin/pkg/client/auth" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/client-go/kubernetes" clientgoscheme "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/certwatcher" @@ -44,11 +39,9 @@ import ( metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" "sigs.k8s.io/controller-runtime/pkg/webhook" - "github.com/IBM/sarama" - sourcesv1alpha1 "github.com/functions-dev/func-operator/api/sources/v1alpha1" + "github.com/functions-dev/func-operator/internal/objectbucketsource/config" "github.com/functions-dev/func-operator/internal/objectbucketsource/controller" - kafkaconfig "github.com/functions-dev/func-operator/internal/objectbucketsource/kafka" "github.com/functions-dev/func-operator/internal/objectbucketsource/notificationserver" // +kubebuilder:scaffold:imports ) @@ -74,7 +67,14 @@ func main() { var probeAddr string var secureMetrics bool var enableHTTP2 bool + var configMapName string + var adapterPort int + var notificationsMode string + var kafkaBrokers string + var kafkaNotificationsTopics string + var kafkaNotificationsGroupID string var tlsOpts []func(*tls.Config) + flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") @@ -92,6 +92,23 @@ func main() { flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.") flag.BoolVar(&enableHTTP2, "enable-http2", false, "If set, HTTP/2 will be enabled for the metrics and webhook servers") + flag.StringVar(&configMapName, "config", "objectbucket-notifications-adapter-config", + "Name of the ConfigMap containing adapter configuration") + flag.IntVar(&adapterPort, "adapter-port", 8888, + "Port the notification HTTP server listens on (HTTP mode only)") + flag.StringVar(¬ificationsMode, "notifications-mode", "http", + "http or kafka - selects how the adapter receives notifications. "+ + "Default for the NOTIFICATIONS_MODE ConfigMap key, which can override it at runtime.") + flag.StringVar(&kafkaBrokers, "kafka-brokers", "", + "Comma-separated list of Kafka broker addresses (required for Kafka mode). "+ + "Default for the KAFKA_BROKERS ConfigMap key, which can override it at runtime.") + flag.StringVar(&kafkaNotificationsTopics, "kafka-notifications-topics", "", + "Comma-separated list of Kafka topics to consume notifications from (required for Kafka mode). "+ + "Default for the KAFKA_NOTIFICATIONS_TOPICS ConfigMap key, which can override it at runtime.") + flag.StringVar(&kafkaNotificationsGroupID, "kafka-notifications-group-id", "", + "Consumer group ID for consuming notifications (required for Kafka mode). "+ + "Default for the KAFKA_NOTIFICATIONS_GROUP_ID ConfigMap key, which can override it at runtime.") + opts := zap.Options{ Development: true, } @@ -178,136 +195,55 @@ func main() { os.Exit(1) } - 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, - }) + // The command-line flags provide the defaults for the notification settings. + // The actual values are resolved from the ConfigMap (falling back to these + // defaults) and can be changed at runtime. Validation happens in the config + // provider when the ConfigMap is loaded. + notificationDefaults := config.NotificationSettings{ + Mode: notificationsMode, + KafkaBrokers: splitAndTrim(kafkaBrokers), + KafkaNotificationsTopics: splitAndTrim(kafkaNotificationsTopics), + KafkaNotificationsGroupID: kafkaNotificationsGroupID, } - adapterPort := 8888 - if portStr := os.Getenv("ADAPTER_PORT"); portStr != "" { - var err error - adapterPort, err = strconv.Atoi(portStr) + // Determine namespace for ConfigMap + ns := os.Getenv("POD_NAMESPACE") + if ns == "" { + nsBytes, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace") if err != nil { - setupLog.Error(err, "invalid ADAPTER_PORT") + setupLog.Error(err, "cannot determine pod namespace") os.Exit(1) } + ns = strings.TrimSpace(string(nsBytes)) } - 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) - } - - 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 - if brokersStr := os.Getenv("KAFKA_BROKERS"); brokersStr != "" { - kafkaBrokers = strings.Split(brokersStr, ",") + // Create configuration provider + configProvider, err := config.NewProvider(context.Background(), ns, configMapName, notificationDefaults) + if err != nil { + setupLog.Error(err, "failed to create configuration provider") + os.Exit(1) } - var kafkaCfg *sarama.Config - if kafkaSecretName := os.Getenv("KAFKA_SECRET"); kafkaSecretName != "" { - ns := os.Getenv("POD_NAMESPACE") - if ns == "" { - nsBytes, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace") - if err != nil { - setupLog.Error(err, "cannot determine pod namespace for KAFKA_SECRET") - os.Exit(1) - } - ns = strings.TrimSpace(string(nsBytes)) - } - clientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie()) - if err != nil { - setupLog.Error(err, "creating kubernetes clientset for KAFKA_SECRET") - os.Exit(1) - } - secret, err := clientset.CoreV1().Secrets(ns).Get(context.Background(), kafkaSecretName, metav1.GetOptions{}) - if err != nil { - setupLog.Error(err, "reading KAFKA_SECRET", "name", kafkaSecretName, "namespace", ns) - os.Exit(1) - } - kafkaCfg, err = kafkaconfig.NewConfig(secret.Data) - if err != nil { - setupLog.Error(err, "configuring kafka from secret", "name", kafkaSecretName) - os.Exit(1) - } - setupLog.Info("kafka configured from secret", "name", kafkaSecretName, "namespace", ns) - } else { - var err error - kafkaCfg, err = kafkaconfig.NewConfig(nil) - if err != nil { - setupLog.Error(err, "creating default kafka config") - os.Exit(1) - } + // Add config provider to manager + if err := mgr.Add(configProvider); err != nil { + setupLog.Error(err, "unable to add config provider to manager") + os.Exit(1) } if err := (&controller.ObjectBucketSourceReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - AdapterConfigs: adapterConfigs, + ConfigProvider: configProvider, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "ObjectBucketSource") os.Exit(1) } // +kubebuilder:scaffold:builder - if notificationsMode == "kafka" { - if len(kafkaNotificationsTopics) == 0 { - 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 := ¬ificationserver.NotificationServer{ - Client: mgr.GetClient(), - Port: adapterPort, - KafkaBrokers: kafkaBrokers, - KafkaConfig: kafkaCfg, - NotificationsMode: notificationsMode, - KafkaNotificationsTopics: kafkaNotificationsTopics, - KafkaNotificationsGroupID: kafkaNotificationsGroupID, + Client: mgr.GetClient(), + Port: adapterPort, + ConfigProvider: configProvider, } if err := mgr.Add(notifServer); err != nil { setupLog.Error(err, "unable to add notification server") @@ -346,9 +282,14 @@ func main() { } } -func envOrDefault(key, defaultValue string) string { - if v := os.Getenv(key); v != "" { - return v +// splitAndTrim splits a comma-separated flag value, trimming whitespace and +// dropping empty entries. +func splitAndTrim(s string) []string { + var out []string + for _, part := range strings.Split(s, ",") { + if trimmed := strings.TrimSpace(part); trimmed != "" { + out = append(out, trimmed) + } } - return defaultValue + return out } diff --git a/config/samples/objectbucket-notifications-adapter-config.yaml b/config/samples/objectbucket-notifications-adapter-config.yaml new file mode 100644 index 0000000..044dcd9 --- /dev/null +++ b/config/samples/objectbucket-notifications-adapter-config.yaml @@ -0,0 +1,27 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: objectbucket-notifications-adapter-config + namespace: system +data: + # NooBaa adapter configuration + NOOBAA_ADAPTER_ID: "mcg-adapter" + NOOBAA_ADAPTER_TOPIC_ARN: "mcg-adapter-connection/connect.json" + NOOBAA_ADAPTER_STORAGECLASS_PATTERN: ".*noobaa\\.io$" + + # RadosGW adapter configuration + RADOSGW_ADAPTER_ID: "rgw-adapter" + RADOSGW_ADAPTER_TOPIC_ARN: "arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications" + RADOSGW_ADAPTER_STORAGECLASS_PATTERN: ".*ceph-rgw$" + + # Notification transport configuration (dynamic). + # These override the corresponding command-line flag defaults and are applied + # at runtime. Changing any Kafka setting gracefully restarts the consumer. + # NOTIFICATIONS_MODE: "http" # http or kafka + # KAFKA_BROKERS: "broker1:9092,broker2:9092" # required for kafka mode + # KAFKA_NOTIFICATIONS_TOPICS: "mcg-notifications,rgw-notifications" # required for kafka mode + # KAFKA_NOTIFICATIONS_GROUP_ID: "adapter-consumer-group" # required for kafka mode + + # Kafka secret for authentication (optional) + # Uncomment to use a Kafka secret for authentication + # KAFKA_SECRET: "kafka-credentials" diff --git a/docs/objectbucket-notifications-adapter-configuration.md b/docs/objectbucket-notifications-adapter-configuration.md new file mode 100644 index 0000000..a30e41c --- /dev/null +++ b/docs/objectbucket-notifications-adapter-configuration.md @@ -0,0 +1,192 @@ +# ObjectBucket Notifications Adapter Configuration + +The `objectbucket-notifications-adapter` supports runtime configuration through a Kubernetes ConfigMap. This allows you to modify adapter settings without restarting the pod. + +## Configuration Overview + +The adapter uses two types of configuration: + +### Static Configuration (Command-line flags) + +These settings are provided at pod startup and require a pod restart to change: + +| Flag | Default | Description | +|------|---------|-------------| +| `--config` | `objectbucket-notifications-adapter-config` | Name of the ConfigMap containing adapter configuration | +| `--adapter-port` | `8888` | Port the notification HTTP server listens on (HTTP mode only) | +| `--notifications-mode` | `http` | Default for the `NOTIFICATIONS_MODE` ConfigMap key (see below) | +| `--kafka-brokers` | _(none)_ | Default for the `KAFKA_BROKERS` ConfigMap key (see below) | +| `--kafka-notifications-topics` | _(none)_ | Default for the `KAFKA_NOTIFICATIONS_TOPICS` ConfigMap key (see below) | +| `--kafka-notifications-group-id` | _(none)_ | Default for the `KAFKA_NOTIFICATIONS_GROUP_ID` ConfigMap key (see below) | + +> **Note:** The notification transport settings (`--notifications-mode`, `--kafka-brokers`, +> `--kafka-notifications-topics`, `--kafka-notifications-group-id`) are now **dynamic**. The +> command-line flags only supply the *defaults*; the effective values come from the ConfigMap +> and can be changed at runtime. When any Kafka setting changes, the adapter gracefully +> restarts its Kafka consumer with the new settings. + +### Dynamic Configuration (ConfigMap) + +These settings can be changed at runtime by modifying the ConfigMap. The adapter watches for changes and reloads automatically: + +| ConfigMap Key | Default | Description | +|---------------|---------|-------------| +| `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 | +| `NOTIFICATIONS_MODE` | value of `--notifications-mode` (`http`) | `http` or `kafka` — selects how the adapter receives NooBaa/RadosGW notifications. Switching modes restarts the notification runner. | +| `KAFKA_BROKERS` | value of `--kafka-brokers` | Comma-separated list of Kafka broker addresses (required for Kafka mode). Changing it gracefully restarts the Kafka consumer. | +| `KAFKA_NOTIFICATIONS_TOPICS` | value of `--kafka-notifications-topics` | Comma-separated list of Kafka topics to consume notifications from (required for Kafka mode). Changing it gracefully restarts the Kafka consumer. | +| `KAFKA_NOTIFICATIONS_GROUP_ID` | value of `--kafka-notifications-group-id` | Consumer group ID for consuming notifications (required for Kafka mode). Changing it gracefully restarts the Kafka consumer. | +| `KAFKA_SECRET` | _(none)_ | Name of a Kubernetes Secret (in the adapter's namespace) containing Kafka credentials. See `kafka-secret-format.md`. | + +## Example ConfigMap + +```yaml +apiVersion: v1 +kind: ConfigMap +metadata: + name: objectbucket-notifications-adapter-config + namespace: your-namespace +data: + # NooBaa adapter configuration + NOOBAA_ADAPTER_ID: "mcg-adapter" + NOOBAA_ADAPTER_TOPIC_ARN: "mcg-adapter-connection/connect.json" + NOOBAA_ADAPTER_STORAGECLASS_PATTERN: ".*noobaa\\.io$" + + # RadosGW adapter configuration + RADOSGW_ADAPTER_ID: "rgw-adapter" + RADOSGW_ADAPTER_TOPIC_ARN: "arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications" + RADOSGW_ADAPTER_STORAGECLASS_PATTERN: ".*ceph-rgw$" + + # Optional: Kafka secret for authentication + KAFKA_SECRET: "kafka-credentials" +``` + +## Deployment + +### 1. Create the ConfigMap + +Create the ConfigMap in the same namespace as the adapter: + +```bash +kubectl apply -f config/samples/objectbucket-notifications-adapter-config.yaml -n your-namespace +``` + +### 2. Deploy the Adapter + +When deploying the adapter, specify the ConfigMap name and static configuration: + +```yaml +apiVersion: apps/v1 +kind: Deployment +metadata: + name: objectbucket-notifications-adapter +spec: + template: + spec: + containers: + - name: adapter + image: your-registry/objectbucket-notifications-adapter:latest + args: + - --config=objectbucket-notifications-adapter-config + - --adapter-port=8888 + - --notifications-mode=http + # For Kafka mode, add: + # - --notifications-mode=kafka + # - --kafka-brokers=broker1:9092,broker2:9092 + # - --kafka-notifications-topics=mcg-notifications,rgw-notifications + # - --kafka-notifications-group-id=adapter-consumer-group + env: + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace +``` + +## Runtime Configuration Updates + +To update the adapter configuration at runtime: + +1. Edit the ConfigMap: + ```bash + kubectl edit configmap objectbucket-notifications-adapter-config -n your-namespace + ``` + +2. The adapter will detect the change and reload the configuration automatically. You'll see log messages like: + ``` + ConfigMap changed, reloading configuration + configuration reloaded successfully + ``` + +3. The new configuration applies immediately to new reconciliation loops. Existing ObjectBucketSource resources will use the updated configuration on their next reconciliation. + +## Dynamic Notification Transport + +The notification transport settings (`NOTIFICATIONS_MODE`, `KAFKA_BROKERS`, +`KAFKA_NOTIFICATIONS_TOPICS`, `KAFKA_NOTIFICATIONS_GROUP_ID`) can be changed at runtime +via the ConfigMap. When any of them change, the adapter: + +1. Stops the current notification runner (the HTTP server or the Kafka consumer), waiting + for it to shut down gracefully. +2. Starts a new runner using the updated settings. + +This means you can, for example, switch the adapter from `http` to `kafka` mode, point the +consumer at different brokers, subscribe to different topics, or change the consumer group +ID — all without restarting the pod. + +If a ConfigMap change produces an invalid notification configuration (for example +`NOTIFICATIONS_MODE=kafka` without `KAFKA_BROKERS`), the reload is rejected: the adapter +logs an error and keeps running with the previous valid configuration. + +## Configuration Validation + +The adapter validates the configuration when loading: + +- Storage class patterns must be valid regular expressions +- All required fields have sensible defaults +- If `KAFKA_SECRET` is specified, the secret must exist in the adapter's namespace + +If validation fails, the adapter logs an error and continues using the previous valid configuration. + +## Kafka Credential Rotation + +To rotate Kafka credentials without restarting the adapter: + +1. Update the Kafka Secret with new credentials +2. Update the ConfigMap to reference the new secret (or trigger a reload by adding/removing a comment) +3. The adapter will reload and use the new credentials for new connections + +Note: Existing Kafka connections will continue using old credentials until they are recreated (on connection failure or pod restart). + +## Migration from Environment Variables + +If you're migrating from the old environment variable configuration: + +1. Create a ConfigMap with values from your current environment variables +2. Update your deployment to use command-line flags instead of environment variables for static settings +3. Remove the environment variable definitions from your deployment +4. Dynamic settings (adapter IDs, topics, patterns) can now be updated via the ConfigMap + +Example migration: + +**Old (env vars):** +```yaml +env: +- name: NOOBAA_ADAPTER_ID + value: "mcg-adapter" +- name: ADAPTER_PORT + value: "8888" +``` + +**New (ConfigMap + flags):** +```yaml +args: +- --adapter-port=8888 +- --config=objectbucket-notifications-adapter-config +``` + +And create a ConfigMap with `NOOBAA_ADAPTER_ID: "mcg-adapter"`. diff --git a/internal/objectbucketsource/config/config.go b/internal/objectbucketsource/config/config.go new file mode 100644 index 0000000..7974444 --- /dev/null +++ b/internal/objectbucketsource/config/config.go @@ -0,0 +1,357 @@ +package config + +import ( + "context" + "fmt" + "regexp" + "strings" + "sync" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes" + ctrl "sigs.k8s.io/controller-runtime" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "github.com/IBM/sarama" + kafkaconfig "github.com/functions-dev/func-operator/internal/objectbucketsource/kafka" +) + +var log = logf.Log.WithName("adapter-config") + +// AdapterBackendConfig holds configuration for a single storage backend adapter +type AdapterBackendConfig struct { + ID string + TopicARN string + StorageClassPattern *regexp.Regexp +} + +// NotificationSettings holds the transport configuration that controls how the +// adapter receives NooBaa/RadosGW notifications. These settings can be changed at +// runtime via the ConfigMap; the notification server restarts its Kafka consumer +// when any of them change. +type NotificationSettings struct { + // Mode is "http" or "kafka". + Mode string + KafkaBrokers []string + KafkaNotificationsTopics []string + KafkaNotificationsGroupID string +} + +// Config holds runtime-configurable settings for the objectbucket-notifications-adapter +type Config struct { + NoobaaAdapter AdapterBackendConfig + RadosgwAdapter AdapterBackendConfig + Notifications NotificationSettings +} + +// Provider provides access to the current configuration and watches for changes +type Provider struct { + mu sync.RWMutex + config Config + + namespace string + configMapName string + clientset *kubernetes.Clientset + cancelWatch context.CancelFunc + + // defaults holds the notification settings supplied via command-line flags. + // They are used whenever the corresponding ConfigMap keys are absent. + defaults NotificationSettings + + kafkaConfigMu sync.RWMutex + kafkaConfig *sarama.Config + kafkaSecret string + + subscribersMu sync.Mutex + subscribers []chan struct{} +} + +// NewProvider creates a new configuration provider that watches a ConfigMap. +// The defaults are used for any notification settings not present in the ConfigMap. +func NewProvider(ctx context.Context, namespace, configMapName string, defaults NotificationSettings) (*Provider, error) { + clientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie()) + if err != nil { + return nil, fmt.Errorf("creating kubernetes clientset: %w", err) + } + + p := &Provider{ + namespace: namespace, + configMapName: configMapName, + clientset: clientset, + defaults: defaults, + } + + if err := p.loadConfig(ctx); err != nil { + return nil, fmt.Errorf("loading initial config: %w", err) + } + + watchCtx, cancel := context.WithCancel(context.Background()) + p.cancelWatch = cancel + go p.watchConfigMap(watchCtx) + + return p, nil +} + +// GetConfig returns a copy of the current configuration +func (p *Provider) GetConfig() Config { + p.mu.RLock() + defer p.mu.RUnlock() + return p.config +} + +// GetKafkaConfig returns the current Kafka configuration +func (p *Provider) GetKafkaConfig() *sarama.Config { + p.kafkaConfigMu.RLock() + defer p.kafkaConfigMu.RUnlock() + return p.kafkaConfig +} + +// Subscribe returns a channel that receives a signal whenever the configuration +// is successfully reloaded. The channel is buffered (size 1) and signals are +// coalesced, so a slow subscriber never blocks the config watcher. +func (p *Provider) Subscribe() <-chan struct{} { + ch := make(chan struct{}, 1) + p.subscribersMu.Lock() + p.subscribers = append(p.subscribers, ch) + p.subscribersMu.Unlock() + return ch +} + +func (p *Provider) notifySubscribers() { + p.subscribersMu.Lock() + defer p.subscribersMu.Unlock() + for _, ch := range p.subscribers { + select { + case ch <- struct{}{}: + default: + } + } +} + +// Stop stops watching the ConfigMap +func (p *Provider) Stop() { + if p.cancelWatch != nil { + p.cancelWatch() + } +} + +// NeedLeaderElection implements the manager.Runnable interface +func (p *Provider) NeedLeaderElection() bool { + return false +} + +// Start implements the manager.Runnable interface +func (p *Provider) Start(ctx context.Context) error { + <-ctx.Done() + p.Stop() + return nil +} + +func (p *Provider) loadConfig(ctx context.Context) error { + cm, err := p.clientset.CoreV1().ConfigMaps(p.namespace).Get(ctx, p.configMapName, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("getting ConfigMap %s/%s: %w", p.namespace, p.configMapName, err) + } + + config, err := parseConfig(cm, p.defaults) + if err != nil { + return fmt.Errorf("parsing ConfigMap: %w", err) + } + + kafkaSecret := cm.Data["KAFKA_SECRET"] + var kafkaCfg *sarama.Config + if kafkaSecret != "" { + secret, err := p.clientset.CoreV1().Secrets(p.namespace).Get(ctx, kafkaSecret, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("reading KAFKA_SECRET %s/%s: %w", p.namespace, kafkaSecret, err) + } + kafkaCfg, err = kafkaconfig.NewConfig(secret.Data) + if err != nil { + return fmt.Errorf("configuring kafka from secret %s: %w", kafkaSecret, err) + } + log.Info("kafka configured from secret", "name", kafkaSecret, "namespace", p.namespace) + } else { + kafkaCfg, err = kafkaconfig.NewConfig(nil) + if err != nil { + return fmt.Errorf("creating default kafka config: %w", err) + } + } + + p.mu.Lock() + p.config = config + p.mu.Unlock() + + p.kafkaConfigMu.Lock() + p.kafkaConfig = kafkaCfg + p.kafkaSecret = kafkaSecret + p.kafkaConfigMu.Unlock() + + log.Info("configuration loaded", + "noobaa-adapter-id", config.NoobaaAdapter.ID, + "radosgw-adapter-id", config.RadosgwAdapter.ID, + "notifications-mode", config.Notifications.Mode, + "kafka-brokers", config.Notifications.KafkaBrokers, + "kafka-notifications-topics", config.Notifications.KafkaNotificationsTopics, + "kafka-notifications-group-id", config.Notifications.KafkaNotificationsGroupID) + + p.notifySubscribers() + + return nil +} + +func (p *Provider) watchConfigMap(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + default: + } + + watcher, err := p.clientset.CoreV1().ConfigMaps(p.namespace).Watch(ctx, metav1.ListOptions{ + FieldSelector: fmt.Sprintf("metadata.name=%s", p.configMapName), + }) + if err != nil { + log.Error(err, "failed to create ConfigMap watcher, retrying in 5s") + select { + case <-ctx.Done(): + return + case <-ctrl.SetupSignalHandler().Done(): + return + } + continue + } + + log.Info("watching ConfigMap for changes", "name", p.configMapName, "namespace", p.namespace) + + func() { + defer watcher.Stop() + + for { + select { + case <-ctx.Done(): + return + case event, ok := <-watcher.ResultChan(): + if !ok { + log.Info("ConfigMap watch channel closed, restarting watcher") + return + } + + if event.Type == watch.Modified || event.Type == watch.Added { + cm, ok := event.Object.(*corev1.ConfigMap) + if !ok { + log.Error(fmt.Errorf("unexpected object type"), "failed to cast to ConfigMap") + continue + } + + log.Info("ConfigMap changed, reloading configuration", "name", cm.Name) + if err := p.loadConfig(ctx); err != nil { + log.Error(err, "failed to reload configuration") + } else { + log.Info("configuration reloaded successfully") + } + } else if event.Type == watch.Deleted { + log.Error(fmt.Errorf("ConfigMap deleted"), "adapter configuration unavailable", "name", p.configMapName) + } + } + } + }() + } +} + +func parseConfig(cm *corev1.ConfigMap, defaults NotificationSettings) (Config, error) { + noobaaPattern := getOrDefault(cm.Data, "NOOBAA_ADAPTER_STORAGECLASS_PATTERN", `.*noobaa\.io$`) + radosgwPattern := getOrDefault(cm.Data, "RADOSGW_ADAPTER_STORAGECLASS_PATTERN", `.*ceph-rgw$`) + + noobaaRe, err := regexp.Compile(noobaaPattern) + if err != nil { + return Config{}, fmt.Errorf("invalid NOOBAA_ADAPTER_STORAGECLASS_PATTERN: %w", err) + } + + radosgwRe, err := regexp.Compile(radosgwPattern) + if err != nil { + return Config{}, fmt.Errorf("invalid RADOSGW_ADAPTER_STORAGECLASS_PATTERN: %w", err) + } + + notifications, err := parseNotificationSettings(cm.Data, defaults) + if err != nil { + return Config{}, err + } + + config := Config{ + NoobaaAdapter: AdapterBackendConfig{ + ID: getOrDefault(cm.Data, "NOOBAA_ADAPTER_ID", "mcg-adapter"), + TopicARN: getOrDefault(cm.Data, "NOOBAA_ADAPTER_TOPIC_ARN", "mcg-adapter-connection/connect.json"), + StorageClassPattern: noobaaRe, + }, + RadosgwAdapter: AdapterBackendConfig{ + ID: getOrDefault(cm.Data, "RADOSGW_ADAPTER_ID", "rgw-adapter"), + TopicARN: getOrDefault(cm.Data, "RADOSGW_ADAPTER_TOPIC_ARN", "arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications"), + StorageClassPattern: radosgwRe, + }, + Notifications: notifications, + } + + return config, nil +} + +// parseNotificationSettings resolves the notification transport settings from the +// ConfigMap, falling back to the provided defaults (from command-line flags) when +// a key is absent. It validates the resulting settings so that invalid ConfigMap +// changes are rejected and the previous valid configuration is retained. +func parseNotificationSettings(data map[string]string, defaults NotificationSettings) (NotificationSettings, error) { + mode := getOrDefault(data, "NOTIFICATIONS_MODE", defaults.Mode) + if mode == "" { + mode = "http" + } + if mode != "http" && mode != "kafka" { + return NotificationSettings{}, fmt.Errorf("invalid NOTIFICATIONS_MODE %q: must be \"http\" or \"kafka\"", mode) + } + + settings := NotificationSettings{ + Mode: mode, + KafkaBrokers: defaults.KafkaBrokers, + KafkaNotificationsTopics: defaults.KafkaNotificationsTopics, + KafkaNotificationsGroupID: getOrDefault(data, "KAFKA_NOTIFICATIONS_GROUP_ID", defaults.KafkaNotificationsGroupID), + } + if v, ok := data["KAFKA_BROKERS"]; ok && strings.TrimSpace(v) != "" { + settings.KafkaBrokers = splitAndTrim(v) + } + if v, ok := data["KAFKA_NOTIFICATIONS_TOPICS"]; ok && strings.TrimSpace(v) != "" { + settings.KafkaNotificationsTopics = splitAndTrim(v) + } + + if mode == "kafka" { + if len(settings.KafkaBrokers) == 0 { + return NotificationSettings{}, fmt.Errorf("KAFKA_BROKERS is required when NOTIFICATIONS_MODE=kafka") + } + if len(settings.KafkaNotificationsTopics) == 0 { + return NotificationSettings{}, fmt.Errorf("KAFKA_NOTIFICATIONS_TOPICS is required when NOTIFICATIONS_MODE=kafka") + } + if settings.KafkaNotificationsGroupID == "" { + return NotificationSettings{}, fmt.Errorf("KAFKA_NOTIFICATIONS_GROUP_ID is required when NOTIFICATIONS_MODE=kafka") + } + } + + return settings, nil +} + +func getOrDefault(data map[string]string, key, defaultValue string) string { + if v, ok := data[key]; ok && v != "" { + return v + } + return defaultValue +} + +// splitAndTrim splits a comma-separated string, trimming whitespace and dropping +// empty entries. +func splitAndTrim(s string) []string { + var out []string + for _, part := range strings.Split(s, ",") { + if trimmed := strings.TrimSpace(part); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} diff --git a/internal/objectbucketsource/config/config_test.go b/internal/objectbucketsource/config/config_test.go new file mode 100644 index 0000000..3859290 --- /dev/null +++ b/internal/objectbucketsource/config/config_test.go @@ -0,0 +1,129 @@ +package config + +import ( + "reflect" + "testing" + + corev1 "k8s.io/api/core/v1" +) + +func TestParseNotificationSettings_DefaultsWhenAbsent(t *testing.T) { + defaults := NotificationSettings{ + Mode: "http", + KafkaBrokers: []string{"b1:9092"}, + KafkaNotificationsTopics: []string{"t1"}, + KafkaNotificationsGroupID: "g1", + } + + got, err := parseNotificationSettings(map[string]string{}, defaults) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !reflect.DeepEqual(got, defaults) { + t.Fatalf("expected defaults %+v, got %+v", defaults, got) + } +} + +func TestParseNotificationSettings_ConfigMapOverrides(t *testing.T) { + defaults := NotificationSettings{Mode: "http"} + data := map[string]string{ + "NOTIFICATIONS_MODE": "kafka", + "KAFKA_BROKERS": "b1:9092, b2:9092 ", + "KAFKA_NOTIFICATIONS_TOPICS": "t1,t2", + "KAFKA_NOTIFICATIONS_GROUP_ID": "grp", + } + + got, err := parseNotificationSettings(data, defaults) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + want := NotificationSettings{ + Mode: "kafka", + KafkaBrokers: []string{"b1:9092", "b2:9092"}, + KafkaNotificationsTopics: []string{"t1", "t2"}, + KafkaNotificationsGroupID: "grp", + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("expected %+v, got %+v", want, got) + } +} + +func TestParseNotificationSettings_InvalidMode(t *testing.T) { + _, err := parseNotificationSettings(map[string]string{"NOTIFICATIONS_MODE": "bogus"}, NotificationSettings{}) + if err == nil { + t.Fatal("expected error for invalid mode, got nil") + } +} + +func TestParseNotificationSettings_KafkaRequirements(t *testing.T) { + tests := []struct { + name string + data map[string]string + }{ + { + name: "missing brokers", + data: map[string]string{ + "NOTIFICATIONS_MODE": "kafka", + "KAFKA_NOTIFICATIONS_TOPICS": "t1", + "KAFKA_NOTIFICATIONS_GROUP_ID": "g1", + }, + }, + { + name: "missing topics", + data: map[string]string{ + "NOTIFICATIONS_MODE": "kafka", + "KAFKA_BROKERS": "b1:9092", + "KAFKA_NOTIFICATIONS_GROUP_ID": "g1", + }, + }, + { + name: "missing group id", + data: map[string]string{ + "NOTIFICATIONS_MODE": "kafka", + "KAFKA_BROKERS": "b1:9092", + "KAFKA_NOTIFICATIONS_TOPICS": "t1", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := parseNotificationSettings(tt.data, NotificationSettings{}); err == nil { + t.Fatalf("expected error for %s, got nil", tt.name) + } + }) + } +} + +func TestParseNotificationSettings_EmptyModeDefaultsToHTTP(t *testing.T) { + got, err := parseNotificationSettings(map[string]string{}, NotificationSettings{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.Mode != "http" { + t.Fatalf("expected mode http, got %q", got.Mode) + } +} + +func TestParseConfig_IncludesNotifications(t *testing.T) { + cm := &corev1.ConfigMap{ + Data: map[string]string{ + "NOTIFICATIONS_MODE": "kafka", + "KAFKA_BROKERS": "b1:9092", + "KAFKA_NOTIFICATIONS_TOPICS": "t1", + "KAFKA_NOTIFICATIONS_GROUP_ID": "g1", + }, + } + + cfg, err := parseConfig(cm, NotificationSettings{Mode: "http"}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg.Notifications.Mode != "kafka" { + t.Fatalf("expected kafka mode, got %q", cfg.Notifications.Mode) + } + if cfg.NoobaaAdapter.ID != "mcg-adapter" { + t.Fatalf("expected default noobaa adapter id, got %q", cfg.NoobaaAdapter.ID) + } +} diff --git a/internal/objectbucketsource/config/interface.go b/internal/objectbucketsource/config/interface.go new file mode 100644 index 0000000..7d3238f --- /dev/null +++ b/internal/objectbucketsource/config/interface.go @@ -0,0 +1,9 @@ +package config + +import "github.com/IBM/sarama" + +// ConfigProvider provides access to adapter configuration +type ConfigProvider interface { + GetConfig() Config + GetKafkaConfig() *sarama.Config +} diff --git a/internal/objectbucketsource/config/mock.go b/internal/objectbucketsource/config/mock.go new file mode 100644 index 0000000..fbd4245 --- /dev/null +++ b/internal/objectbucketsource/config/mock.go @@ -0,0 +1,89 @@ +package config + +import ( + "regexp" + "sync" + + "github.com/IBM/sarama" +) + +// MockProvider is a simple mock implementation of the config provider for testing +type MockProvider struct { + mu sync.RWMutex + config Config + kafkaConfig *sarama.Config + subscribers []chan struct{} +} + +// NewMockProvider creates a mock provider with default test configuration +func NewMockProvider() *MockProvider { + noobaaRe := regexp.MustCompile(`.*noobaa\.io$`) + radosgwRe := regexp.MustCompile(`.*ceph-rgw$`) + + return &MockProvider{ + config: Config{ + NoobaaAdapter: AdapterBackendConfig{ + ID: "mcg-adapter", + TopicARN: "mcg-adapter-connection/connect.json", + StorageClassPattern: noobaaRe, + }, + RadosgwAdapter: AdapterBackendConfig{ + ID: "rgw-adapter", + TopicARN: "arn:aws:sns:ocs-storagecluster-cephobjectstore::rgw-adapter-notifications", + StorageClassPattern: radosgwRe, + }, + Notifications: NotificationSettings{ + Mode: "http", + }, + }, + kafkaConfig: sarama.NewConfig(), + } +} + +// GetConfig returns the mock configuration +func (m *MockProvider) GetConfig() Config { + m.mu.RLock() + defer m.mu.RUnlock() + return m.config +} + +// GetKafkaConfig returns the mock Kafka configuration +func (m *MockProvider) GetKafkaConfig() *sarama.Config { + m.mu.RLock() + defer m.mu.RUnlock() + return m.kafkaConfig +} + +// Subscribe returns a channel that is signaled whenever the mock configuration changes. +func (m *MockProvider) Subscribe() <-chan struct{} { + ch := make(chan struct{}, 1) + m.mu.Lock() + m.subscribers = append(m.subscribers, ch) + m.mu.Unlock() + return ch +} + +func (m *MockProvider) notify() { + for _, ch := range m.subscribers { + select { + case ch <- struct{}{}: + default: + } + } +} + +// SetConfig updates the mock configuration (for testing) +func (m *MockProvider) SetConfig(cfg Config) { + m.mu.Lock() + defer m.mu.Unlock() + m.config = cfg + m.notify() +} + +// SetKafkaConfig updates the mock Kafka configuration (for testing) +func (m *MockProvider) SetKafkaConfig(cfg *sarama.Config) { + m.mu.Lock() + defer m.mu.Unlock() + m.kafkaConfig = cfg + m.notify() +} diff --git a/internal/objectbucketsource/controller/objectbucketsource_controller.go b/internal/objectbucketsource/controller/objectbucketsource_controller.go index 1a1bde6..67ede48 100644 --- a/internal/objectbucketsource/controller/objectbucketsource_controller.go +++ b/internal/objectbucketsource/controller/objectbucketsource_controller.go @@ -36,6 +36,7 @@ import ( logf "sigs.k8s.io/controller-runtime/pkg/log" sourcesv1alpha1 "github.com/functions-dev/func-operator/api/sources/v1alpha1" + "github.com/functions-dev/func-operator/internal/objectbucketsource/config" "github.com/functions-dev/func-operator/internal/objectbucketsource/s3client" ) @@ -56,7 +57,7 @@ var obcGVR = schema.GroupVersionResource{ type ObjectBucketSourceReconciler struct { client.Client Scheme *runtime.Scheme - AdapterConfigs []AdapterConfig + ConfigProvider config.ConfigProvider } // +kubebuilder:rbac:groups=sources.functions.dev,resources=objectbucketsources,verbs=get;list;watch;create;update;patch;delete @@ -271,18 +272,32 @@ func (r *ObjectBucketSourceReconciler) readOBCStorageClassName(ctx context.Conte } func (r *ObjectBucketSourceReconciler) resolveAdapterConfig(ctx context.Context, namespace, obcName string) (*AdapterConfig, error) { - if len(r.AdapterConfigs) == 0 { - return nil, fmt.Errorf("no adapter configs defined") + cfg := r.ConfigProvider.GetConfig() + + adapterConfigs := []AdapterConfig{ + { + ID: cfg.NoobaaAdapter.ID, + Topic: cfg.NoobaaAdapter.TopicARN, + StorageClassPattern: cfg.NoobaaAdapter.StorageClassPattern, + }, + { + ID: cfg.RadosgwAdapter.ID, + Topic: cfg.RadosgwAdapter.TopicARN, + StorageClassPattern: cfg.RadosgwAdapter.StorageClassPattern, + }, } - if len(r.AdapterConfigs) == 1 { - return &r.AdapterConfigs[0], nil + + if len(adapterConfigs) == 1 { + return &adapterConfigs[0], nil } + storageClass, err := r.readOBCStorageClassName(ctx, namespace, obcName) if err != nil { return nil, err } - for i := range r.AdapterConfigs { - cfg := &r.AdapterConfigs[i] + + for i := range adapterConfigs { + cfg := &adapterConfigs[i] if cfg.StorageClassPattern != nil && cfg.StorageClassPattern.MatchString(storageClass) { return cfg, nil } diff --git a/internal/objectbucketsource/controller/objectbucketsource_controller_test.go b/internal/objectbucketsource/controller/objectbucketsource_controller_test.go index d0bf536..49c5d72 100644 --- a/internal/objectbucketsource/controller/objectbucketsource_controller_test.go +++ b/internal/objectbucketsource/controller/objectbucketsource_controller_test.go @@ -28,6 +28,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" sourcesv1alpha1 "github.com/functions-dev/func-operator/api/sources/v1alpha1" + "github.com/functions-dev/func-operator/internal/objectbucketsource/config" ) var _ = Describe("ObjectBucketSource Controller", func() { @@ -76,8 +77,9 @@ var _ = Describe("ObjectBucketSource Controller", func() { It("should successfully reconcile the resource", func() { By("Reconciling the created resource") controllerReconciler := &ObjectBucketSourceReconciler{ - Client: k8sClient, - Scheme: k8sClient.Scheme(), + Client: k8sClient, + Scheme: k8sClient.Scheme(), + ConfigProvider: config.NewMockProvider(), } _, err := controllerReconciler.Reconcile(ctx, reconcile.Request{ diff --git a/internal/objectbucketsource/notificationserver/server.go b/internal/objectbucketsource/notificationserver/server.go index 9ffbf51..2818a04 100644 --- a/internal/objectbucketsource/notificationserver/server.go +++ b/internal/objectbucketsource/notificationserver/server.go @@ -11,36 +11,114 @@ import ( logf "sigs.k8s.io/controller-runtime/pkg/log" ceDispatch "github.com/functions-dev/func-operator/internal/objectbucketsource/cloudevents" + "github.com/functions-dev/func-operator/internal/objectbucketsource/config" ) var log = logf.Log.WithName("notification-server") +// ConfigProvider provides access to configuration +type ConfigProvider interface { + GetConfig() config.Config + GetKafkaConfig() *sarama.Config + // Subscribe returns a channel that is signaled whenever the configuration is reloaded. + Subscribe() <-chan struct{} +} + type NotificationServer struct { - Client client.Client - Port int - KafkaBrokers []string - KafkaConfig *sarama.Config - NotificationsMode string - KafkaNotificationsTopics []string - KafkaNotificationsGroupID string + Client client.Client + Port int + ConfigProvider ConfigProvider } +// Start runs the notification runner (HTTP server or Kafka consumer) according to +// the current configuration and supervises it: when the notification-related +// settings change in the ConfigMap, it gracefully stops the current runner and +// starts a new one with the updated settings. func (s *NotificationServer) Start(ctx context.Context) error { + changes := s.ConfigProvider.Subscribe() + + for { + if ctx.Err() != nil { + return nil + } + + settings := s.ConfigProvider.GetConfig().Notifications + if shutdown := s.superviseRun(ctx, changes, settings); shutdown { + return nil + } + } +} + +// superviseRun starts the notification runner for the given settings and blocks +// until either the parent context is cancelled (returns true, indicating +// shutdown) or a restart is required because the settings changed or the runner +// exited (returns false). It always stops the runner before returning. +func (s *NotificationServer) superviseRun(ctx context.Context, changes <-chan struct{}, settings config.NotificationSettings) (shutdown bool) { + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + + errCh := make(chan error, 1) + go func() { + errCh <- s.run(runCtx, settings) + }() + + for { + select { + case <-ctx.Done(): + <-errCh + return true + case err := <-errCh: + // The runner exited on its own (fatal setup error). Restart it, + // unless we're shutting down. + if ctx.Err() != nil { + return true + } + if err != nil { + log.Error(err, "notification runner stopped unexpectedly, restarting in 5s") + select { + case <-ctx.Done(): + return true + case <-time.After(5 * time.Second): + } + } + return false + case <-changes: + newSettings := s.ConfigProvider.GetConfig().Notifications + if notificationSettingsEqual(settings, newSettings) { + // Unrelated configuration change (e.g. adapter IDs); keep running. + continue + } + log.Info("notification settings changed, restarting notification runner", + "old-mode", settings.Mode, "new-mode", newSettings.Mode, + "old-brokers", settings.KafkaBrokers, "new-brokers", newSettings.KafkaBrokers, + "old-topics", settings.KafkaNotificationsTopics, "new-topics", newSettings.KafkaNotificationsTopics, + "old-group-id", settings.KafkaNotificationsGroupID, "new-group-id", newSettings.KafkaNotificationsGroupID) + cancel() + <-errCh + return false + } + } +} + +// run starts the notification transport for the given settings and blocks until +// ctx is cancelled or a fatal error occurs. +func (s *NotificationServer) run(ctx context.Context, settings config.NotificationSettings) error { var kafkaProducer sarama.SyncProducer - if len(s.KafkaBrokers) > 0 { + if len(settings.KafkaBrokers) > 0 { var err error - kafkaProducer, err = ceDispatch.NewKafkaProducer(s.KafkaBrokers, s.KafkaConfig) + kafkaCfg := s.ConfigProvider.GetKafkaConfig() + kafkaProducer, err = ceDispatch.NewKafkaProducer(settings.KafkaBrokers, kafkaCfg) if err != nil { return fmt.Errorf("creating kafka producer: %w", err) } defer func() { _ = kafkaProducer.Close() }() - log.Info("kafka producer initialized", "brokers", s.KafkaBrokers) + log.Info("kafka producer initialized", "brokers", settings.KafkaBrokers) } handler := ¬ificationHandler{client: s.Client, kafkaProducer: kafkaProducer} - if s.NotificationsMode == "kafka" { - return s.startKafkaConsumer(ctx, handler) + if settings.Mode == "kafka" { + return s.startKafkaConsumer(ctx, handler, settings) } return s.startHTTPServer(ctx, handler) } @@ -71,21 +149,22 @@ func (s *NotificationServer) startHTTPServer(ctx context.Context, handler *notif return nil } -func (s *NotificationServer) startKafkaConsumer(ctx context.Context, handler *notificationHandler) error { - consumerConfig := *s.KafkaConfig +func (s *NotificationServer) startKafkaConsumer(ctx context.Context, handler *notificationHandler, settings config.NotificationSettings) error { + kafkaCfg := s.ConfigProvider.GetKafkaConfig() + consumerConfig := *kafkaCfg consumerConfig.Consumer.Return.Errors = true consumerConfig.Consumer.Offsets.Initial = sarama.OffsetNewest - consumerGroup, err := sarama.NewConsumerGroup(s.KafkaBrokers, s.KafkaNotificationsGroupID, &consumerConfig) + consumerGroup, err := sarama.NewConsumerGroup(settings.KafkaBrokers, settings.KafkaNotificationsGroupID, &consumerConfig) if err != nil { return fmt.Errorf("creating kafka consumer group: %w", err) } defer func() { _ = consumerGroup.Close() }() log.Info("starting kafka notification consumer", - "topics", s.KafkaNotificationsTopics, - "group", s.KafkaNotificationsGroupID, - "brokers", s.KafkaBrokers) + "topics", settings.KafkaNotificationsTopics, + "group", settings.KafkaNotificationsGroupID, + "brokers", settings.KafkaBrokers) go func() { for err := range consumerGroup.Errors() { @@ -96,7 +175,7 @@ func (s *NotificationServer) startKafkaConsumer(ctx context.Context, handler *no cgHandler := &consumerGroupHandler{handler: handler} for { - if err := consumerGroup.Consume(ctx, s.KafkaNotificationsTopics, cgHandler); err != nil { + if err := consumerGroup.Consume(ctx, settings.KafkaNotificationsTopics, cgHandler); err != nil { if ctx.Err() != nil { return nil } @@ -111,3 +190,24 @@ func (s *NotificationServer) startKafkaConsumer(ctx context.Context, handler *no func (s *NotificationServer) NeedLeaderElection() bool { return false } + +// notificationSettingsEqual reports whether the two notification settings are +// equivalent for the purposes of deciding whether the runner must be restarted. +func notificationSettingsEqual(a, b config.NotificationSettings) bool { + return a.Mode == b.Mode && + a.KafkaNotificationsGroupID == b.KafkaNotificationsGroupID && + stringSlicesEqual(a.KafkaBrokers, b.KafkaBrokers) && + stringSlicesEqual(a.KafkaNotificationsTopics, b.KafkaNotificationsTopics) +} + +func stringSlicesEqual(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} From a2589da24164dec5cde7f2fb3def82a661144a8b Mon Sep 17 00:00:00 2001 From: Marek Schmidt Date: Thu, 13 Aug 2026 15:32:01 +0200 Subject: [PATCH 2/2] vet fix --- internal/objectbucketsource/config/config.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/objectbucketsource/config/config.go b/internal/objectbucketsource/config/config.go index 7974444..cb68fa8 100644 --- a/internal/objectbucketsource/config/config.go +++ b/internal/objectbucketsource/config/config.go @@ -6,6 +6,7 @@ import ( "regexp" "strings" "sync" + "time" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -217,8 +218,7 @@ func (p *Provider) watchConfigMap(ctx context.Context) { select { case <-ctx.Done(): return - case <-ctrl.SetupSignalHandler().Done(): - return + case <-time.After(5 * time.Second): } continue }