diff --git a/charts/qiskit-serverless/README.md b/charts/qiskit-serverless/README.md index 0fe236b8d..98a89362b 100644 --- a/charts/qiskit-serverless/README.md +++ b/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' +``` diff --git a/charts/qiskit-serverless/charts/gateway/templates/deployment.yaml b/charts/qiskit-serverless/charts/gateway/templates/deployment.yaml index 535fd5688..08dc70b17 100644 --- a/charts/qiskit-serverless/charts/gateway/templates/deployment.yaml +++ b/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: diff --git a/charts/qiskit-serverless/charts/gateway/values.yaml b/charts/qiskit-serverless/charts/gateway/values.yaml index c98e008fc..e520e8814 100644 --- a/charts/qiskit-serverless/charts/gateway/values.yaml +++ b/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: diff --git a/gateway/core/ibm_cloud/event_streams/kafka_event_streams_client.py b/gateway/core/ibm_cloud/event_streams/kafka_event_streams_client.py index a28b90d36..718ff1175 100644 --- a/gateway/core/ibm_cloud/event_streams/kafka_event_streams_client.py +++ b/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_ — broker list for additional regions + EVENT_STREAMS_API_KEY_ — API key for additional regions + EVENT_STREAMS_USER_ — 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::...). + 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") diff --git a/gateway/main/settings.py b/gateway/main/settings.py index 726761b38..949a6a2df 100644 --- a/gateway/main/settings.py +++ b/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")) diff --git a/gateway/tests/core/services/ibm_cloud/event_streams/test_event_streams_client.py b/gateway/tests/core/services/ibm_cloud/event_streams/test_event_streams_client.py index e06ba21d9..21971bfd3 100644 --- a/gateway/tests/core/services/ibm_cloud/event_streams/test_event_streams_client.py +++ b/gateway/tests/core/services/ibm_cloud/event_streams/test_event_streams_client.py @@ -15,6 +15,7 @@ from __future__ import annotations import json +import logging import os import uuid as uuid_module from datetime import datetime, timezone @@ -94,6 +95,57 @@ def test_topic_constructed_from_environment(self): assert client.topic == "quantum.staging.function-usage.v1" + def test_custom_user_in_default_region(self): + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker1:9093", + "EVENT_STREAMS_API_KEY": "my-key", + "EVENT_STREAMS_USER": "custom-user", + "ENVIRONMENT": "staging", + }, + ): + KafkaEventStreamsClient() + + mock_producer_cls.assert_called_once_with( + { + "bootstrap.servers": "broker1:9093", + "security.protocol": "SASL_SSL", + "sasl.mechanisms": "PLAIN", + "sasl.username": "custom-user", + "sasl.password": "my-key", + "enable.idempotence": True, + "acks": "all", + } + ) + + def test_custom_user_in_regional_producer(self): + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker-us:9093", + "EVENT_STREAMS_API_KEY": "us-key", + "EVENT_STREAMS_USER": "default-user", + "EVENT_STREAMS_BOOTSTRAP_SERVERS_EU_DE": "broker-eu:9093", + "EVENT_STREAMS_API_KEY_EU_DE": "eu-key", + "EVENT_STREAMS_USER_EU_DE": "custom-eu-user", + "ENVIRONMENT": "production", + }, + clear=True, + ): + KafkaEventStreamsClient() + + calls = mock_producer_cls.call_args_list + assert len(calls) == 2 + + default_call = [c for c in calls if "broker-us" in str(c)][0] + eu_call = [c for c in calls if "broker-eu" in str(c)][0] + + assert default_call[0][0]["sasl.username"] == "default-user" + assert eu_call[0][0]["sasl.username"] == "custom-eu-user" + def test_emit_job_started_publishes_correct_payload(self): job = _make_job() @@ -352,6 +404,218 @@ def test_business_model_absent_from_non_license_events(self): assert "business_model" not in published["data"] assert published["data"]["job_started_at"] == job.running_started_at.isoformat() + def test_default_region_producer_from_unsuffixed_vars(self): + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker1:9093", + "EVENT_STREAMS_API_KEY": "default-key", + "ENVIRONMENT": "production", + }, + clear=True, + ): + client = KafkaEventStreamsClient() + + assert "us-east" in client._producers + mock_producer_cls.assert_called_once() + + def test_suffixed_vars_discovered_by_scan(self): + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker-us:9093", + "EVENT_STREAMS_API_KEY": "us-key", + "EVENT_STREAMS_BOOTSTRAP_SERVERS_EU_DE": "broker-eu:9093", + "EVENT_STREAMS_API_KEY_EU_DE": "eu-key", + "ENVIRONMENT": "production", + }, + clear=True, + ): + client = KafkaEventStreamsClient() + + assert "us-east" in client._producers + assert "eu-de" in client._producers + assert mock_producer_cls.call_count == 2 + + def test_event_streams_default_region_respected(self): + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker1:9093", + "EVENT_STREAMS_API_KEY": "key", + "EVENT_STREAMS_DEFAULT_REGION": "eu-gb", + "ENVIRONMENT": "production", + }, + clear=True, + ): + client = KafkaEventStreamsClient() + + assert client._default_region == "eu-gb" + assert "eu-gb" in client._producers + + def test_routing_selects_right_producer(self): + job_us = _make_job(instance_crn="crn:v1:bluemix:public:quantum-computing:us-east:a/abc:def::") + job_eu = _make_job(instance_crn="crn:v1:bluemix:public:quantum-computing:eu-de:a/abc:def::") + + mock_producer_us = MagicMock() + mock_producer_eu = MagicMock() + mock_producer_us.flush.return_value = 0 + mock_producer_eu.flush.return_value = 0 + + def create_producer_side_effect(config): + if "broker-us" in config.get("bootstrap.servers", ""): + return mock_producer_us + elif "broker-eu" in config.get("bootstrap.servers", ""): + return mock_producer_eu + return MagicMock() + + with patch(f"{_CLIENT_MOD}.Producer", side_effect=create_producer_side_effect): + with patch(f"{_CLIENT_MOD}.uuid"): + with patch(f"{_CLIENT_MOD}.datetime") as mock_dt: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker-us:9093", + "EVENT_STREAMS_API_KEY": "us-key", + "EVENT_STREAMS_BOOTSTRAP_SERVERS_EU_DE": "broker-eu:9093", + "EVENT_STREAMS_API_KEY_EU_DE": "eu-key", + "ENVIRONMENT": "production", + }, + clear=True, + ): + mock_dt.now.return_value = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + client = KafkaEventStreamsClient() + + client.emit_job_started(job_us, "classical_24x120") + client.emit_job_started(job_eu, "classical_24x120") + + assert mock_producer_us.produce.called + assert mock_producer_eu.produce.called + + def test_unconfigured_region_raises(self): + job = _make_job(instance_crn="crn:v1:bluemix:public:quantum-computing:au-syd:a/abc:def::") + + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch(f"{_CLIENT_MOD}.uuid"): + with patch(f"{_CLIENT_MOD}.datetime") as mock_dt: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "b:9093", + "EVENT_STREAMS_API_KEY": "k", + "ENVIRONMENT": "production", + }, + clear=True, + ): + mock_dt.now.return_value = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + client = KafkaEventStreamsClient() + mock_producer = mock_producer_cls.return_value + mock_producer.flush.return_value = 0 + + with pytest.raises(RuntimeError, match="No producer configured for region au-syd"): + client.emit_job_started(job, "classical_24x120") + + def test_null_crn_uses_default_region(self): + job = _make_job(instance_crn=None) + + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch(f"{_CLIENT_MOD}.uuid"): + with patch(f"{_CLIENT_MOD}.datetime") as mock_dt: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "b:9093", + "EVENT_STREAMS_API_KEY": "k", + "ENVIRONMENT": "production", + }, + clear=True, + ): + mock_dt.now.return_value = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + client = KafkaEventStreamsClient() + mock_producer = mock_producer_cls.return_value + mock_producer.flush.return_value = 0 + + client.emit_job_started(job, "classical_24x120") + + mock_producer.produce.assert_called_once() + + def test_malformed_crn_uses_default_region(self): + job = _make_job(instance_crn="not:a:valid:crn") + + with patch(f"{_CLIENT_MOD}.Producer") as mock_producer_cls: + with patch(f"{_CLIENT_MOD}.uuid"): + with patch(f"{_CLIENT_MOD}.datetime") as mock_dt: + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "b:9093", + "EVENT_STREAMS_API_KEY": "k", + "ENVIRONMENT": "production", + }, + clear=True, + ): + mock_dt.now.return_value = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + client = KafkaEventStreamsClient() + mock_producer = mock_producer_cls.return_value + mock_producer.flush.return_value = 0 + + client.emit_job_started(job, "classical_24x120") + + mock_producer.produce.assert_called_once() + + def test_broker_list_without_matching_api_key_raises_at_init(self): + with patch(f"{_CLIENT_MOD}.Producer"): + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "b:9093", + "EVENT_STREAMS_API_KEY": "k", + "EVENT_STREAMS_BOOTSTRAP_SERVERS_EU_DE": "broker-eu:9093", + "ENVIRONMENT": "production", + }, + clear=True, + ): + with pytest.raises(ValueError, match="missing EVENT_STREAMS_API_KEY_EU_DE"): + KafkaEventStreamsClient() + + def test_startup_log_line(self, caplog): + with patch(f"{_CLIENT_MOD}.Producer"): + with patch.dict( + os.environ, + { + "EVENT_STREAMS_BOOTSTRAP_SERVERS": "broker-us:9093", + "EVENT_STREAMS_API_KEY": "us-key", + "EVENT_STREAMS_BOOTSTRAP_SERVERS_EU_DE": "broker-eu:9093", + "EVENT_STREAMS_API_KEY_EU_DE": "eu-key", + "ENVIRONMENT": "production", + }, + clear=True, + ): + with caplog.at_level(logging.INFO): + KafkaEventStreamsClient() + + assert "Event Streams producers initialized" in caplog.text + assert "regions=" in caplog.text + assert "default=us-east" in caplog.text + + def test_region_from_crn_extracts_correctly(self): + assert ( + KafkaEventStreamsClient._region_from_crn("crn:v1:bluemix:public:quantum-computing:us-east:a/abc:def::") + == "us-east" + ) + assert ( + KafkaEventStreamsClient._region_from_crn("crn:v1:bluemix:public:quantum-computing:eu-de:a/abc:def::") + == "eu-de" + ) + + def test_region_from_crn_returns_none_for_invalid_crn(self): + assert KafkaEventStreamsClient._region_from_crn(None) is None + assert KafkaEventStreamsClient._region_from_crn("") is None + assert KafkaEventStreamsClient._region_from_crn("not:a:valid:crn") is None + def test_filler_job_publishes_nothing(self): """A filler job generates no usage events: the base class short-circuits all four emits.""" job = _make_job() diff --git a/gateway/tox.ini b/gateway/tox.ini index 0ed60c8ec..341af41f7 100644 --- a/gateway/tox.ini +++ b/gateway/tox.ini @@ -25,6 +25,9 @@ setenv = # Local-only dev secret so static analysis tools (pylint-django, import-linter) # can import Django settings, which now fail closed with DEBUG off and no key. DJANGO_SECRET_KEY=django-insecure-tox-only-change-me + # Event Streams test configuration for multi-region support + EVENT_STREAMS_DEFAULT_REGION=us-east + ENVIRONMENT=production deps = -rrequirements.txt -rrequirements-dev.txt