Add regions to kafka

charts/qiskit-serverless/README.md

@@ -71,3 +71,29 @@ Gateway is the API that we offer to manage Qiskit Patterns. Scheduler is the par
 | secrets.servicePsql.*                     | `name` and `key` let you to specify the secret and `value` the value for the secret                                                                                                           |
 | secrets.superuser.create                  | Port number that service will be exposed externally                                                                                                                                           |
 | secrets.superuser.*                       | `name` and `key` let you to specify the secret and `value` the value for the secret                                                                                                           |
+
+**Event Streams (Multi-Region)**
+
+Event Streams is an IBM Cloud service for publishing function usage events to regional Kafka buses. The service is configured with a default region and optional additional regional buses. See [Multi-region Event Streams routing design](../../../2026-08-06-multi-region-kafka-design.md) for full details.
+
+| Name                                    | Description                                                                                                |
+|-----------------------------------------|------------------------------------------------------------------------------------------------------------|
+| application.eventStreams.enabled        | enable / disable event publishing to Kafka                                                                 |
+| application.eventStreams.environment    | deployment environment name (e.g. production, staging); passed to Kafka topic name                         |
+| application.eventStreams.defaultRegion  | default region served by unsuffixed `kafka-credentials` secret (default: `us-east`)                        |
+
+**Event Streams Secrets**
+
+The default region uses the `kafka-credentials` secret with keys `bootstrap_servers` and `api_key`. Additional regions use suffixed secrets: `kafka-credentials-eu-de`, `kafka-credentials-ap-sg`, etc. Each regional secret requires the same keys.
+
+```bash
+# Default region (required if eventStreams.enabled is true)
+kubectl create secret generic kafka-credentials \
+  --from-literal=bootstrap_servers='broker1:9093,broker2:9093' \
+  --from-literal=api_key='your-api-key'
+
+# Additional region (optional; automatically detected if present)
+kubectl create secret generic kafka-credentials-eu-de \
+  --from-literal=bootstrap_servers='broker-eu1:9093,broker-eu2:9093' \
+  --from-literal=api_key='your-eu-api-key'
+```

charts/qiskit-serverless/charts/gateway/templates/deployment.yaml

@@ -545,6 +545,8 @@ spec:
               value: {{ .Values.application.eventStreams.enabled | default false | quote }}
             - name: ENVIRONMENT
               value: {{ .Values.application.eventStreams.environment | default "" | quote }}
+            - name: EVENT_STREAMS_DEFAULT_REGION
+              value: {{ .Values.application.eventStreams.defaultRegion | default "us-east" | quote }}
             {{- if .Values.application.eventStreams.enabled }}
             - name: EVENT_STREAMS_BOOTSTRAP_SERVERS
               valueFrom:
@@ -561,6 +563,22 @@ spec:
                 secretKeyRef:
                   name: kafka-credentials
                   key: user
+            # Multi-region Event Streams support: optional suffixed regional secrets
+            # (kafka-credentials-eu-de, kafka-credentials-ap-sg, etc) are projected here if present
+            {{- range $region := (list "eu-de") }}
+            {{- if (lookup "v1" "Secret" $.Release.Namespace (print "kafka-credentials-" $region)) }}
+            - name: EVENT_STREAMS_BOOTSTRAP_SERVERS_{{ upper (replace $region "-" "_") }}
+              valueFrom:
+                secretKeyRef:
+                  name: kafka-credentials-{{ $region }}
+                  key: bootstrap_servers
+            - name: EVENT_STREAMS_API_KEY_{{ upper (replace $region "-" "_") }}
+              valueFrom:
+                secretKeyRef:
+                  name: kafka-credentials-{{ $region }}
+                  key: api_key
+            {{- end }}
+            {{- end }}
             {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:

charts/qiskit-serverless/charts/gateway/values.yaml

@@ -85,6 +85,10 @@ application:
   eventStreams:
     enabled: false
     environment: ""
+    # Default region served by unsuffixed broker/API key secrets. Additional regions
+    # are configured via suffixed secrets (kafka-credentials-eu-de, etc).
+    # See: Multi-region Event Streams routing design (2026-08-06)
+    defaultRegion: "us-east"
   # Optional IBM w3id SSO backoffice login. The provider base url is not
   # sensitive and lives here; credentials go in secrets.w3idSso (see below).
   w3idSso:

gateway/core/ibm_cloud/event_streams/kafka_event_streams_client.py

@@ -43,38 +43,90 @@ class KafkaEventStreamsClient(EventStreamsClient):
     boundaries without interpreting the metric type. License fee events also
     carry `business_model`.
 
-    Configured from environment variables:
-      EVENT_STREAMS_BOOTSTRAP_SERVERS — comma-separated broker list
-      EVENT_STREAMS_API_KEY           — SASL/PLAIN password
-      ENVIRONMENT                     — deployment environment (e.g. production, staging)
+    Configured from environment variables per region:
+      EVENT_STREAMS_BOOTSTRAP_SERVERS         — comma-separated broker list (default region)
+      EVENT_STREAMS_API_KEY                   — SASL/PLAIN password (default region)
+      EVENT_STREAMS_USER                      — SASL/PLAIN username (default: 'token')
+      EVENT_STREAMS_BOOTSTRAP_SERVERS_<REGION> — broker list for additional regions
+      EVENT_STREAMS_API_KEY_<REGION>          — API key for additional regions
+      EVENT_STREAMS_USER_<REGION>             — SASL/PLAIN username for additional regions
+      EVENT_STREAMS_DEFAULT_REGION            — default region (default: us-east)
+      ENVIRONMENT                             — deployment environment (e.g. production, staging)
     """
 
     def __init__(self) -> None:
-        bootstrap_servers = os.environ["EVENT_STREAMS_BOOTSTRAP_SERVERS"]
-        api_key = os.environ["EVENT_STREAMS_API_KEY"]
         environment = os.environ["ENVIRONMENT"]
+        default_region = os.environ.get("EVENT_STREAMS_DEFAULT_REGION", "us-east")
 
-        # LOG: Initialization
+        # Initialize producers from environment variables
+        self._producers: dict[str, Producer] = {}
+        self._default_region = default_region
+
+        # Register default region from unsuffixed variables
+        default_bootstrap_servers = os.environ.get("EVENT_STREAMS_BOOTSTRAP_SERVERS")
+        default_api_key = os.environ.get("EVENT_STREAMS_API_KEY")
+        default_user = os.environ.get("EVENT_STREAMS_USER", "token")
+        if default_bootstrap_servers and default_api_key:
+            logger.debug("Registering default region producer: region=%s", default_region)
+            self._producers[default_region] = self._create_producer(
+                default_bootstrap_servers, default_api_key, default_user
+            )
+
+        # Discover regional producers by scanning for suffixed env vars
+        for env_key in os.environ:
+            if env_key.startswith("EVENT_STREAMS_BOOTSTRAP_SERVERS_"):
+                suffix = env_key[len("EVENT_STREAMS_BOOTSTRAP_SERVERS_") :]
+                region = suffix.lower().replace("_", "-")
+                bootstrap_servers = os.environ[env_key]
+                api_key_env = f"EVENT_STREAMS_API_KEY_{suffix}"
+                user_env = f"EVENT_STREAMS_USER_{suffix}"
+                api_key = os.environ.get(api_key_env)
+                user = os.environ.get(user_env, "token")
+
+                if api_key is None:
+                    raise ValueError(f"Region {region}: found {env_key} but missing {api_key_env}")
+
+                logger.debug("Registering regional producer: region=%s", region)
+                self._producers[region] = self._create_producer(bootstrap_servers, api_key, user)
+
+        self.topic = f"quantum.{environment}.function-usage.v1"
+
+        # Log initialized regions
+        regions = sorted(self._producers.keys())
         logger.info(
-            "Initializing KafkaEventStreamsClient bootstrap_servers=%s environment=%s", bootstrap_servers, environment
+            "Event Streams producers initialized: regions=%s (default=%s)",
+            regions,
+            default_region,
         )
 
-        self._producer = Producer(
+    def _create_producer(self, bootstrap_servers: str, api_key: str, user: str = "token") -> Producer:
+        """Create and return a Kafka producer with the given credentials."""
+        return Producer(
             {
                 "bootstrap.servers": bootstrap_servers,
                 "security.protocol": "SASL_SSL",
                 "sasl.mechanisms": "PLAIN",
-                "sasl.username": "token",
+                "sasl.username": user,
                 "sasl.password": api_key,
-                # Add delivery callback configuration
                 "enable.idempotence": True,
                 "acks": "all",
             }
         )
-        self.topic = f"quantum.{environment}.function-usage.v1"
 
-        # LOG: Client created successfully
-        logger.info("KafkaEventStreamsClient initialized successfully topic=%s", self.topic)
+    @staticmethod
+    def _region_from_crn(instance_crn: str | None) -> str | None:
+        """Extract the region from an instance CRN.
+
+        The region is the 6th colon-delimited segment of the CRN
+        (crn:v1:bluemix:public:quantum-computing:<region>:...).
+        Returns None if the CRN is absent or has too few segments.
+        """
+        if not instance_crn:
+            return None
+        parts = instance_crn.split(":")
+        if len(parts) > 5:
+            return parts[5]
+        return None
 
     def _emit_job_started(self, job, metric_type: str | None = None) -> None:
         """Publish a job-started event for the given metric (metric_value=0)."""
@@ -182,15 +234,26 @@ def _publish(
             "data": data,
         }
 
+        # Route to the appropriate regional producer
+        region = self._region_from_crn(job.instance_crn)
+        if region is None:
+            region = self._default_region
+        producer = self._producers.get(region)
+        if producer is None:
+            raise RuntimeError(
+                f"KafkaEventStreamsClient: No producer configured for region {region} "
+                f"(job_id={job.id}, event_id={event_id})"
+            )
+
         try:
-            self._producer.produce(
+            producer.produce(
                 topic=self.topic,
                 key=str(job.id).encode("utf-8"),
                 value=json.dumps(event).encode("utf-8"),
                 callback=self._delivery_callback,
             )
 
-            remaining = self._producer.flush(timeout=5)
+            remaining = producer.flush(timeout=5)
 
             if remaining > 0:
                 raise RuntimeError(f"KafkaEventStreamsClient: {remaining} message(s) not delivered after flush timeout")

gateway/main/settings.py

@@ -337,6 +337,10 @@
 LIMITS_GPU_CLUSTERS = int(os.environ.get("LIMITS_MAX_GPU_CLUSTERS", "1"))
 LIMITS_MAX_FLEETS = int(os.environ.get("LIMITS_MAX_FLEETS", "1000"))  # Fleets Project limit
 EVENT_STREAMS_ENABLED = os.environ.get("EVENT_STREAMS_ENABLED", "false").lower() == "true"
+# The Event Streams service is global; usage events are regional. EVENT_STREAMS_DEFAULT_REGION
+# specifies which regional Kafka bus receives events from unsuffixed broker/API key environment
+# variables. Additional regions are configured via suffixed variables (e.g. EVENT_STREAMS_BOOTSTRAP_SERVERS_EU_DE).
+EVENT_STREAMS_DEFAULT_REGION = os.environ.get("EVENT_STREAMS_DEFAULT_REGION", "us-east")
 LIMITS_CPU_PER_TASK = int(os.environ.get("LIMITS_CPU_PER_TASK", "4"))
 LIMITS_GPU_PER_TASK = int(os.environ.get("LIMITS_GPU_PER_TASK", "1"))
 LIMITS_MEMORY_PER_TASK = int(os.environ.get("LIMITS_MEMORY_PER_TASK", "8"))