Add Kafka messaging backend

.github/workflows/main.yml

@@ -10,12 +10,12 @@ on:
 
 jobs:
   linting:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
       - uses: actions/checkout@v4
       - name: Set up Python 3.9
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
         with:
           python-version: "3.9"
       - uses: fedora-python/tox-github-action@v39.0
@@ -32,12 +32,12 @@ jobs:
           dnf_install: gcc krb5-devel
 
   testing:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
       - uses: actions/checkout@v4
       - name: Set up Python 3.9
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
         with:
           python-version: "3.9"
       - uses: fedora-python/tox-github-action@v39.0

README.rst

@@ -72,6 +72,19 @@ messaging broker.
 
 .. _TOML: https://github.com/toml-lang/toml
 
+Messaging Backend
+~~~~~~~~~~~~~~~~~~
+
+MTS supports three messaging backends, selected by the ``messaging_backend``
+config value:
+
+* ``fedora-messaging`` (default) -- Fedora infra RabbitMQ, configured via
+  ``conf/mts.toml``.
+* ``rhmsg`` -- Internal Unified Message Bus (UMB), configured via the
+  ``rhmsg_*`` config values.
+* ``kafka`` -- Kafka service, configured via the ``kafka_*``
+  config values.
+
 Messaging
 ---------
 
@@ -126,9 +139,11 @@ Topic Prefix
 For Fedora, messages are sent to topics with prefix ``org.fedoraproject.prod``,
 e.g. ``org.fedoraproject.prod.mts.build.tag.requested``.
 
-For internal, the prefix is ``VirtualTopic.eng.mts``, e.g.
+For internal UMB (``rhmsg``), the prefix is ``VirtualTopic.eng.mts``, e.g.
 ``VirtualTopic.eng.mts.build.tag.requested``.
 
+For Kafka, the prefix is ``eng.mts``, e.g. ``eng.mts.build.tag.requested``.
+
 Environment Variables
 ---------------------
 
@@ -151,6 +166,19 @@ Specify an alternative config file instead of reading the ``conf/config.py``
 when ``MTS_DEV`` is enabled or ``/etc/mts/config.py`` if MTS is installed in
 system. Either an absolute path or relative path works.
 
+MTS_KAFKA_USERNAME
+~~~~~~~~~~~~~~~~~~
+
+Kafka SASL username used when ``messaging_backend`` is ``kafka``. Overrides
+the ``kafka_sasl_username`` config value.
+
+MTS_KAFKA_PASSWORD
+~~~~~~~~~~~~~~~~~~
+
+Kafka SASL password used when ``messaging_backend`` is ``kafka``. Overrides
+the ``kafka_sasl_password`` config value. Sourcing the password from the
+environment avoids storing the secret in the config file.
+
 Contribution
 ------------
 

conf/config.py

@@ -51,6 +51,42 @@ class BaseConfiguration:
     # to enable durable messages.
     rhmsg_subscription_name = None
 
+    # Kafka bootstrap servers to connect to. When using SCRAM-SHA-512 over
+    # SASL_SSL, connect on port 9096, e.g.
+    # ['kafka-broker1:9096', 'kafka-broker2:9096']
+    kafka_bootstrap_servers = []
+    # Kafka topic MTS consumes MBS module state change events from.
+    kafka_consumer_topic = 'eng.mbs.module.state.change'
+    # Consumer group id. Kafka tracks committed offsets per consumer group, so
+    # keep this stable to avoid reprocessing messages across restarts.
+    kafka_consumer_group_id = 'mts'
+    # Where to start consuming when no committed offset exists for the group.
+    kafka_auto_offset_reset = 'earliest'
+    # A topic like build.tag.requested is passed to the publish function to
+    # generalize the messaging publish interface. This prefix is prepended to
+    # construct the full topic MTS publishes to.
+    kafka_topic_prefix = 'eng.mts'
+    # Timeout (seconds) to wait for a published message to be acknowledged.
+    kafka_send_timeout = 30
+
+    # Connection security for Kafka. One of: PLAINTEXT, SSL, SASL_PLAINTEXT,
+    # SASL_SSL.
+    kafka_security_protocol = 'SASL_SSL'
+    # SASL mechanism, e.g. SCRAM-SHA-512, PLAIN or OAUTHBEARER. Leave empty when
+    # not using SASL. SCRAM-SHA-512 over SASL_SSL is the recommended mechanism;
+    # avoid AWS IAM as it does not work well with non-Java clients.
+    kafka_sasl_mechanism = 'SCRAM-SHA-512'
+    # Credentials for SASL username/password mechanisms (PLAIN, SCRAM-*). The
+    # password is read from the MTS_KAFKA_PASSWORD environment variable by
+    # default so that the secret does not have to be stored in the config file.
+    kafka_sasl_username = os.environ.get('MTS_KAFKA_USERNAME', '')
+    kafka_sasl_password = os.environ.get('MTS_KAFKA_PASSWORD', '')
+    # Absolute paths to TLS material. kafka_ssl_cafile verifies the brokers;
+    # kafka_ssl_certfile/kafka_ssl_keyfile are only needed for mutual TLS auth.
+    kafka_ssl_cafile = ''
+    kafka_ssl_certfile = ''
+    kafka_ssl_keyfile = ''
+
     # Default is INFO. Refer to Python logging module to know valid values.
     log_level = 'INFO'
 

docker/install-dependencies.sh

@@ -10,6 +10,7 @@ dependencies=(
     python3-fedora-messaging
     python3-flask
     python3-gunicorn
+    python3-kafka
     python3-koji
     python3-qpid-proton
     python3-pip

message_tagging_service/consumer.py

@@ -59,6 +59,38 @@ def __repr__(self):
         )
 
 
+class KafkaMessage(object):
+    """Representing a message consumed from Kafka service
+
+    :param msg: the record consumed from Kafka.
+    :type msg: ``kafka.consumer.fetcher.ConsumerRecord``
+    """
+
+    def __init__(self, msg):
+        self._orig_msg = msg
+        self._body = json.loads(msg.value)
+
+    @property
+    def id(self):
+        # Kafka records do not carry a message id, so derive a stable one from
+        # the record's coordinates within the topic.
+        return '{}-{}-{}'.format(
+            self._orig_msg.topic, self._orig_msg.partition, self._orig_msg.offset)
+
+    @property
+    def topic(self):
+        return self._orig_msg.topic
+
+    @property
+    def body(self):
+        return self._body
+
+    def __repr__(self):
+        return "{}(id={}, topic={}, body={})".format(
+            self.__class__.__name__, repr(self.id), repr(self.topic), repr(self.body)
+        )
+
+
 def consume(msg):
     """Do the work to tag build if it matches a rule
 
@@ -150,6 +182,48 @@ def _consumer_wrapper(msg, data=None):
         subscription_name=conf.rhmsg_subscription_name)
 
 
+def kafka_backend():
+    """Launch consumer backend based on kafka-python to consume from Kafka
+
+    Offsets are committed manually only after ``consume`` returns, giving an
+    at-least-once delivery guarantee: if the process crashes while handling a
+    message, the uncommitted message is re-delivered on restart.
+
+    Because an uncommitted offset is re-delivered, letting an exception escape
+    this loop would stop the consumer on the same message forever. Failures are
+    logged and the offset committed instead, matching how ``consume`` already
+    keeps going after a failed tagging attempt.
+    """
+    from kafka import KafkaConsumer
+    from message_tagging_service.messaging import kafka_client_config
+
+    consumer = KafkaConsumer(
+        conf.kafka_consumer_topic,
+        group_id=conf.kafka_consumer_group_id,
+        auto_offset_reset=conf.kafka_auto_offset_reset,
+        enable_auto_commit=False,
+        **kafka_client_config())
+
+    try:
+        for msg in consumer:
+            logger.debug('Received message: %r', msg)
+            try:
+                consume(KafkaMessage(msg))
+            except json.JSONDecodeError as e:
+                # A message with an undecodable body is a poison message; log
+                # it and commit so the consumer does not get stuck on it.
+                logger.error(f'Cannot decode message body: {msg.value}')
+                logger.error(f'Reason: {str(e)}')
+            except Exception:  # noqa
+                # Anything else, e.g. a message missing a field ``consume``
+                # expects, would otherwise be re-delivered forever.
+                logger.exception(f'Failed to handle message {msg.value}')
+                logger.info('Continue to handle next MBS message ...')
+            consumer.commit()
+    finally:
+        consumer.close()
+
+
 def run():
     """The entrypoint of MTS to run specific consumer backend
 
@@ -159,5 +233,7 @@ def run():
         rhmsg_backend()
     elif conf.messaging_backend == 'fedora-messaging':
         fedora_messaging_backend()
+    elif conf.messaging_backend == 'kafka':
+        kafka_backend()
     else:
         raise ValueError(f'Unknown messaging backend: {conf.messaging_backend}')

message_tagging_service/messaging.py

@@ -90,11 +90,76 @@ def _rhmsg_publish(topic, msg):
             producer.send(outgoing_msg)
 
 
+def kafka_client_config():
+    """Build common connection kwargs for KafkaProducer and KafkaConsumer.
+
+    Only the options that are configured are included so that unused auth
+    material (e.g. SASL credentials when using plain SSL) is left at the
+    kafka-python defaults.
+
+    :return: keyword arguments accepted by both ``KafkaProducer`` and
+        ``KafkaConsumer``.
+    :rtype: dict
+    """
+    config = {
+        'bootstrap_servers': conf.kafka_bootstrap_servers,
+        'security_protocol': conf.kafka_security_protocol,
+    }
+    if conf.kafka_sasl_mechanism:
+        config['sasl_mechanism'] = conf.kafka_sasl_mechanism
+    if conf.kafka_sasl_username:
+        config['sasl_plain_username'] = conf.kafka_sasl_username
+    if conf.kafka_sasl_password:
+        config['sasl_plain_password'] = conf.kafka_sasl_password
+    if conf.kafka_ssl_cafile:
+        config['ssl_cafile'] = conf.kafka_ssl_cafile
+    if conf.kafka_ssl_certfile:
+        config['ssl_certfile'] = conf.kafka_ssl_certfile
+    if conf.kafka_ssl_keyfile:
+        config['ssl_keyfile'] = conf.kafka_ssl_keyfile
+    return config
+
+
+def _kafka_publish(topic, msg):
+    """Send message to the Kafka service
+
+    :param str topic: the topic where message will be sent to, e.g.
+        ``build.tagged``. The configured ``kafka_topic_prefix`` is prepended to
+        build the full topic name.
+    :param dict msg: the message that will be sent.
+    """
+    from kafka import KafkaProducer
+
+    full_topic = f'{conf.kafka_topic_prefix.rstrip(".")}.{topic}'
+
+    if conf.dry_run:
+        logger.info(
+            'DRY-RUN: send message to kafka, topic: %s, msg: %s',
+            full_topic, msg)
+        return
+
+    producer = KafkaProducer(
+        value_serializer=lambda v: json.dumps(v).encode('utf-8'),
+        **kafka_client_config())
+    try:
+        logger.debug('Send message to topic %s: %s', full_topic, msg)
+        future = producer.send(full_topic, value=msg)
+        # Block until the broker acknowledges the message so that any delivery
+        # error is raised here and counted by the caller's failure handler.
+        future.get(timeout=conf.kafka_send_timeout)
+    finally:
+        producer.flush()
+        producer.close()
+
+
 _messaging_backends = {
     'fedora-messaging': {
         'publish': _fedora_messaging_publish
     },
     'rhmsg': {
         'publish': _rhmsg_publish
+    },
+    'kafka': {
+        'publish': _kafka_publish
     }
 }

requirements-no-hashes.txt

@@ -68,6 +68,8 @@ jsonschema==4.23.0
     # via fedora-messaging
 jsonschema-specifications==2023.12.1
     # via jsonschema
+kafka-python==2.0.2
+    # via -r requirements.in
 koji==1.34.2
     # via -r requirements.in
 markupsafe==2.1.5

requirements.in

@@ -1,5 +1,6 @@
 Flask
 PyYAML
+kafka-python
 koji
 psutil
 python-qpid-proton

requirements.txt

@@ -304,6 +304,10 @@ jsonschema-specifications==2023.12.1 \
     --hash=sha256:48a76787b3e70f5ed53f1160d2b81f586e4ca6d1548c5de7085d1682674764cc \
     --hash=sha256:87e4fdf3a94858b8a2ba2778d9ba57d8a9cafca7c7489c46ba0d30a8bc6a9c3c
     # via jsonschema
+kafka-python==2.0.2 \
+    --hash=sha256:04dfe7fea2b63726cd6f3e79a2d86e709d608d74406638c5da33a01d45a9d7e3 \
+    --hash=sha256:2d92418c7cb1c298fa6c7f0fb3519b520d0d7526ac6cb7ae2a4fc65a51a94b6e
+    # via -r requirements.in
 koji==1.34.2 \
     --hash=sha256:2965b0a6a680779cdbc8dff5b19b2bdbd0e925bd4f848b40bee18078b00c4c59 \
     --hash=sha256:4c4a2840b5209f81d656d269f81a582899a0d2978c47ef35575082f7e72afae7 \