feat: add Kafka publish for build/batch state-change events

README.md

@@ -268,6 +268,21 @@ The custom configuration options for AMQP 1.0 messaging are listed below:
 * `IIB_MESSAGING_URLS` - a list of AMQP(S) URLs to use when connecting to the AMQP 1.0 broker. This
   must be set if messaging is enabled.
 
+The custom configuration options for Kafka messaging are listed below. Kafka messaging is optional.
+When configured, IIB publishes the state-change events to configured Kafka topics :
+
+* `IIB_KAFKA_BROKERS` - a list of Kafka broker addresses (e.g. `['broker1:9096', 'broker2:9096']`).
+  If this is not set, Kafka messaging is disabled entirely.
+* `IIB_KAFKA_USERNAME` - the SASL username for Kafka authentication. Both `IIB_KAFKA_USERNAME` and
+  `IIB_KAFKA_PASSWORD` must be set together.
+* `IIB_KAFKA_PASSWORD` - the SASL password for Kafka authentication.
+* `IIB_KAFKA_SSL_CAFILE` - the path to a CA certificate file used to verify the Kafka broker's TLS
+  certificate. This defaults to `/etc/pki/tls/certs/ca-bundle.crt`.
+* `IIB_KAFKA_BUILD_STATE_TOPIC` - the Kafka topic to publish build request state-change messages to.
+  If this is not set, build state-change messages will not be published to Kafka.
+* `IIB_KAFKA_BATCH_STATE_TOPIC` - the Kafka topic to publish batch state-change messages to.
+  If this is not set, batch state-change messages will not be published to Kafka.
+
 If you wish to configure AWS S3 bucket for storing artifact files, the following **environment variables**
 must be set along with `IIB_AWS_S3_BUCKET_NAME` config variable:
 
@@ -508,10 +523,10 @@ modifications, such as registry replacement, will still be applied.
 
 ## Messaging
 
-IIB has support to send messages to an AMQP 1.0 broker. If configured to do so, IIB will send
-messages when a build request state changes and when a batch state changes. Please note that if a
-message can't be sent due to an infrastructure issue, the build request will continue as it is not
-considered a fatal error.
+IIB has support to send messages to an AMQP 1.0 broker (will be depricated in future) and to Apache Kafka topics.
+If configured to do so, IIB will send messages when a build request state changes and when a batch
+state changes. Please notethat if a message can't be sent due to an infrastructure issue, the build request will
+continue as it is not considered a fatal error.
 
 The build request state change message body is the JSON representation of the build request in
 the non-verbose format like in the `/builds` API endpoint. The message has the following keys set in

docs/module_documentation/iib.web.rst

@@ -66,6 +66,14 @@ iib.web.iib\_static\_types module
    :undoc-members:
    :show-inheritance:
 
+iib.web.kafka\_producer module
+------------------------------
+
+.. automodule:: iib.web.kafka_producer
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
 iib.web.manage module
 ---------------------
 

docs/requirements.txt

@@ -5,6 +5,7 @@ flask
 flask-login
 flask-migrate
 flask-sqlalchemy
+kafka-python
 opentelemetry-api
 opentelemetry-exporter-otlp
 opentelemetry-instrumentation

iib/web/config.py

@@ -38,6 +38,13 @@ class Config(object):
     IIB_REQUEST_RELATED_BUNDLES_DIR: Optional[str] = None
     IIB_REQUEST_RECURSIVE_RELATED_BUNDLES_DIR: Optional[str] = None
     IIB_USER_TO_QUEUE: Union[Dict[str, str], Dict[str, Dict[str, str]]] = _get_empty_dict_str_str()
+    # Kafka messaging configuration
+    IIB_KAFKA_BATCH_STATE_TOPIC: Optional[str] = None
+    IIB_KAFKA_BROKERS: Optional[List[str]] = None
+    IIB_KAFKA_BUILD_STATE_TOPIC: Optional[str] = None
+    IIB_KAFKA_SSL_CAFILE: str = '/etc/pki/tls/certs/ca-bundle.crt'
+    IIB_KAFKA_USERNAME: Optional[str] = None
+    IIB_KAFKA_PASSWORD: Optional[str] = None
     IIB_WORKER_USERNAMES: List[str] = []
     SQLALCHEMY_TRACK_MODIFICATIONS: bool = False
 

iib/web/kafka_producer.py

@@ -0,0 +1,95 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+import json
+import threading
+from typing import Any, Dict, Optional, Union
+
+from flask import current_app
+from kafka import KafkaProducer
+
+from iib.web.iib_static_types import (
+    BaseClassRequestResponse,
+    BatchRequestResponseList,
+)
+
+_kafka_producer: Optional[KafkaProducer] = None
+_producer_lock = threading.Lock()
+
+
+def get_kafka_producer() -> Optional[KafkaProducer]:
+    """Return a cached, thread-safe KafkaProducer."""
+    global _kafka_producer
+
+    # First check without lock for performance
+    if _kafka_producer is not None:
+        return _kafka_producer
+
+    with _producer_lock:
+        # Second check inside lock to prevent race conditions
+        if _kafka_producer is not None:
+            return _kafka_producer
+
+        conf = current_app.config
+        brokers = conf.get('IIB_KAFKA_BROKERS')
+        if not brokers:
+            return None
+
+        try:
+            username = conf.get('IIB_KAFKA_USERNAME')
+            password = conf.get('IIB_KAFKA_PASSWORD')
+
+            if not username or not password:
+                current_app.logger.error('Kafka credentials missing, cannot initialize producer')
+                return None
+
+            producer_kwargs = {
+                'bootstrap_servers': brokers,
+                'security_protocol': 'SASL_SSL',
+                'sasl_mechanism': 'SCRAM-SHA-512',
+                'sasl_plain_username': username,
+                'sasl_plain_password': password,
+                'value_serializer': lambda v: json.dumps(v).encode('utf-8'),
+                'key_serializer': lambda k: str(k).encode('utf-8') if k is not None else None,
+                'linger_ms': 5,
+            }
+
+            ca_file = conf.get('IIB_KAFKA_SSL_CAFILE')
+            if ca_file:
+                producer_kwargs['ssl_cafile'] = ca_file
+
+            _kafka_producer = KafkaProducer(**producer_kwargs)
+
+            current_app.logger.info('Kafka producer initialised')
+            return _kafka_producer
+        except Exception:
+            current_app.logger.exception('Failed to initialise KafkaProducer')
+            return None
+
+
+def on_send_error(excp):
+    """Log failed Kafka message delivery in a background-thread-safe way."""
+    import logging
+
+    logging.getLogger(__name__).error('Failed to deliver message to Kafka', exc_info=excp)
+
+
+def send_kafka_message(
+    producer: KafkaProducer,
+    topic: str,
+    content: Union[BaseClassRequestResponse, BatchRequestResponseList],
+    properties: Optional[Dict[str, Any]] = None,
+) -> None:
+    """Send a single message to a Kafka topic asynchronously."""
+    properties = properties or {}
+    headers = [(k, str(v).encode('utf-8')) for k, v in properties.items() if v is not None]
+    message_key = properties.get('id', properties.get('batch'))
+
+    try:
+        current_app.logger.info(f'Queuing message for Kafka topic {topic}')
+
+        future = producer.send(topic, key=message_key, value=content, headers=headers)
+
+        # Attach callbacks for async failure handling
+        future.add_errback(on_send_error)
+
+    except Exception:
+        current_app.logger.exception(f'Failed to queue message for Kafka topic {topic}')

iib/web/messaging.py

@@ -16,13 +16,82 @@
     BatchRequestResponseList,
 )
 from iib.web.models import Batch, Request, RequestStateMapping
+from iib.web.kafka_producer import get_kafka_producer, send_kafka_message
 
 __all__ = ['Envelope', 'json_to_envelope', 'send_messages', 'send_message_for_state_change']
 
 
 Envelope = namedtuple('Envelope', 'address message')
 
 
+def _build_request_state_change_data(
+    request: Request,
+) -> tuple:
+    """
+    Build the content and properties for a request state change message.
+
+    Used by both the legacy AMQP and Kafka paths.
+
+    :param iib.web.models.Request request: the request that changed states
+    :return: a tuple of (content, properties)
+    :rtype: tuple
+    """
+    request_json = cast(BaseClassRequestResponse, request.to_json(verbose=False))
+    properties = {
+        'batch': request_json['batch'],
+        'id': request_json['id'],
+        'state': request_json['state'],
+        'user': request_json['user'],
+    }
+    return request_json, properties
+
+
+def _build_batch_state_change_data(
+    batch: Batch,
+    new_batch: Optional[bool] = False,
+) -> Optional[tuple]:
+    """
+    Build the content and properties for a batch state change message.
+
+    Returns ``None`` when no message should be sent.
+    Used by both the legacy AMQP and Kafka paths.
+
+    :param iib.web.models.Batch batch: the batch that changed states
+    :param bool new_batch: if ``True``, a new batch message will be generated
+    :return: a tuple of (content, properties) or None
+    :rtype: tuple or None
+    """
+    if new_batch:
+        batch_state = 'in_progress'
+    else:
+        batch_state = batch.state
+
+    if not (new_batch or batch_state in RequestStateMapping.get_final_states()):
+        return None
+
+    batch_username = getattr(batch.user, 'username', None)
+    content: BatchRequestResponseList = {
+        'batch': batch.id,
+        'annotations': batch.annotations,
+        'requests': [
+            {
+                'id': r.id,
+                'organization': getattr(r, 'organization', None),
+                'request_type': r.type_name,
+            }
+            for r in batch.requests
+        ],
+        'state': batch_state,
+        'user': batch_username,
+    }
+    properties = {
+        'batch': batch.id,
+        'state': batch_state,
+        'user': batch_username,
+    }
+    return content, properties
+
+
 def _get_batch_state_change_envelope(
     batch: Batch,
     new_batch: Optional[bool] = False,
@@ -47,34 +116,10 @@ def _get_batch_state_change_envelope(
         )
         return None
 
-    if new_batch:
-        # Avoid querying the database for the batch state since we know it's a new batch
-        batch_state = 'in_progress'
-    else:
-        batch_state = batch.state
-
-    if new_batch or batch_state in RequestStateMapping.get_final_states():
+    data = _build_batch_state_change_data(batch, new_batch)
+    if data:
         current_app.logger.debug('Preparing to send a state change message for batch %d', batch.id)
-        batch_username = getattr(batch.user, 'username', None)
-        content: BatchRequestResponseList = {
-            'batch': batch.id,
-            'annotations': batch.annotations,
-            'requests': [
-                {
-                    'id': request.id,
-                    'organization': getattr(request, 'organization', None),
-                    'request_type': request.type_name,
-                }
-                for request in batch.requests
-            ],
-            'state': batch_state,
-            'user': batch_username,
-        }
-        properties = {
-            'batch': batch.id,
-            'state': batch_state,
-            'user': batch_username,
-        }
+        content, properties = data
         return json_to_envelope(batch_address, content, properties)
     return None
 
@@ -98,15 +143,8 @@ def _get_request_state_change_envelope(request: Request) -> Optional[Envelope]:
         return None
 
     current_app.logger.debug('Preparing to send a state change message for request %d', request.id)
-    # cast from Union - see Request.to_json
-    request_json = cast(BaseClassRequestResponse, request.to_json(verbose=False))
-    properties = {
-        'batch': request_json['batch'],
-        'id': request_json['id'],
-        'state': request_json['state'],
-        'user': request_json['user'],
-    }
-    return json_to_envelope(request_address, request_json, properties)
+    content, properties = _build_request_state_change_data(request)
+    return json_to_envelope(request_address, content, properties)
 
 
 def _get_ssl_domain() -> Optional[proton.SSLDomain]:
@@ -199,6 +237,41 @@ def send_messages(envelopes: List[Envelope]) -> None:
             connection.close()
 
 
+def _send_kafka_messages(
+    requests: List[Request],
+    batch: Batch,
+    new_batch: Optional[bool] = False,
+) -> None:
+    """
+    Send request and batch state-change messages to Kafka.
+
+    :param list requests: one or more requests whose state changed
+    :param iib.web.models.Batch batch: the batch associated with the requests
+    :param bool new_batch: if ``True``, a batch-creation message is sent
+    """
+    try:
+        producer = get_kafka_producer()
+        if not producer:
+            return
+
+        conf = current_app.config
+
+        build_topic = conf.get('IIB_KAFKA_BUILD_STATE_TOPIC')
+        if build_topic:
+            for request in requests:
+                content, properties = _build_request_state_change_data(request)
+                send_kafka_message(producer, build_topic, content, properties)
+
+        batch_topic = conf.get('IIB_KAFKA_BATCH_STATE_TOPIC')
+        if batch_topic:
+            batch_data = _build_batch_state_change_data(batch, new_batch)
+            if batch_data:
+                content, properties = batch_data
+                send_kafka_message(producer, batch_topic, content, properties)
+    except:  # noqa: E722
+        current_app.logger.exception('Failed to send one or more messages')
+
+
 def send_message_for_state_change(request: Request, new_batch_msg: Optional[bool] = False) -> None:
     """
     Send the appropriate message(s) based on a build request state change.
@@ -223,6 +296,8 @@ def send_message_for_state_change(request: Request, new_batch_msg: Optional[bool
     if envelopes:
         send_messages(envelopes)
 
+    _send_kafka_messages([request], request.batch, new_batch=new_batch_msg)
+
 
 def send_messages_for_new_batch_of_requests(requests: List[Request]) -> None:
     """
@@ -250,3 +325,5 @@ def send_messages_for_new_batch_of_requests(requests: List[Request]) -> None:
 
     if envelopes:
         send_messages(envelopes)
+
+    _send_kafka_messages(requests, batch, new_batch=True)

requirements.txt

@@ -538,6 +538,10 @@ jmespath==1.1.0 \
     # via
     #   boto3
     #   botocore
+kafka-python==3.0.11 \
+    --hash=sha256:9d10cab4e11e02545d82c7e5af5702da5aa46dd4eccd11ad92a50bf6dbbecd14 \
+    --hash=sha256:a003d927e79c801d6cfd1e59ceaaf78807351e75cdb5b8ee9ce4262586f9780f
+    # via iib (setup.py)
 kombu==5.6.2 \
     --hash=sha256:8060497058066c6f5aed7c26d7cd0d3b574990b09de842a8c5aaed0b92cc5a55 \
     --hash=sha256:efcfc559da324d41d61ca311b0c64965ea35b4c55cc04ee36e55386145dace93

setup.py

@@ -17,6 +17,7 @@
         'flask-migrate',
         'flask-sqlalchemy',
         'importlib-resources',
+        'kafka-python',
         'kubernetes',
         'operator-manifest==0.0.5',
         'psycopg2-binary',