Merge pull request #308 from ymorin-orange/nyma/gl292-retain-expiry

python: handle retain in IoT3 SDK, allow filtering in IQM

Author: ymorin-orange

python/iot3/src/iot3/core/mqtt.py

@@ -81,19 +81,26 @@ def __init__(
         and thus receiving messages from specific topics. If no msg_cb
         is provided, it is not possible to subscribe, and only emitting
         is permitted. msg_cb must be a callable that accepts at least
-        those keyword arguments: data, topic, payload. Ideally, to be
-        future-proof, it should be declared with:
+        those keyword arguments: data, topic, payload, retain. Ideally,
+        to be future-proof, it should be declared with:
             def my_cb(
                 *_args,
-                *,
                 data: Any,
                 topic: str,
                 payload: bytes,
+                retain: bool | int,
                 **_kwargs,
             ) -> None
 
         If msg_cb_data is specified, it is passed as the data keyword
         argument of msg_cb, otherwise None is passed.
+
+        Note: the retain flag will be set to False if the message was
+        not published in retain; it will be set to True if the message
+        was published in retain without a Message Expiry Interval; it
+        will be set to a non-zero integer representing the remaining
+        Message Expiry Interval if the message was published in retain
+        with a non-zero Message Expiry Interval.
         """
 
         self.msg_cb = msg_cb
@@ -171,27 +178,49 @@ def stop(self):
         self.client.disconnect()
         self.client.loop_stop()
 
-    def publish(self, *, topic: str, payload: bytes | str):
+    def publish(
+        self,
+        *,
+        topic: str,
+        payload: bytes | str,
+        retain: bool | int = False,
+    ):
         """Publish an MQTT message
 
         :param topic: The MQTT topic to post on.
         :param payload: The payload to post.
+        :param retain: Whether the message should be set to retain; can be
+                       either a bool(), in which case the message will not
+                       be retained (False) or retained indefinitely (True),
+                       or a strictly positive int(), in which case the message
+                       will be sent in retain with a Message Expiry Interval
+                       set to the specified value (min: 1, max: 2^32-1).
         """
         with self.span_ctxmgr_cb(
             name="IoT3 Core MQTT Message",
             kind=otel.SpanKind.PRODUCER,
         ) as span:
-            new_traceparent = span.to_traceparent()
-            span.set_attribute(key="iot3.core.mqtt.topic", value=topic)
-            span.set_attribute(key="iot3.core.mqtt.payload_size", value=len(payload))
             properties = paho.mqtt.properties.Properties(
                 paho.mqtt.packettypes.PacketTypes.PUBLISH,
             )
+            new_traceparent = span.to_traceparent()
+            span.set_attribute(key="iot3.core.mqtt.topic", value=topic)
+            span.set_attribute(key="iot3.core.mqtt.payload_size", value=len(payload))
+            if retain:
+                span.set_attribute(key="iot3.core.mqtt.retain", value=True)
+                if isinstance(retain, int):
+                    if retain < 1:
+                        raise ValueError(
+                            f"retain must be strictly positive (not: {retain})"
+                        )
+                    properties.MessageExpiryInterval = retain
+                    retain = True
             if new_traceparent:
                 properties.UserProperty = ("traceparent", new_traceparent)
             msg_info = self.client.publish(
                 topic=topic,
                 payload=payload,
+                retain=retain,
                 properties=properties,
             )
             if msg_info.rc:
@@ -214,7 +243,7 @@ def subscribe(self, *, topics: list[str]):
         with self.subscriptions_lock:
             sub = topics.difference(self.subscriptions)
             if sub and self.client.is_connected():
-                self.client.subscribe(list(map(lambda t: (t, 0), sub)))
+                self._do_subscribe(topics=sub)
             self.subscriptions.update(topics)
 
     def subscribe_replace(self, *, topics: list[str]):
@@ -242,7 +271,7 @@ def subscribe_replace(self, *, topics: list[str]):
                 if unsub:
                     self.client.unsubscribe(list(unsub))
                 if sub:
-                    self.client.subscribe(list(map(lambda t: (t, 0), sub)))
+                    self._do_subscribe(topics=sub)
             self.subscriptions.clear()
             self.subscriptions.update(topics)
 
@@ -270,6 +299,19 @@ def unsubscribe_all(self):
         with self.subscriptions_lock:
             self.unsubscribe(topics=self.subscriptions)
 
+    def _do_subscribe(
+        self,
+        *,
+        topics: list[str],
+    ):
+        opts = paho.mqtt.client.SubscribeOptions(
+            qos=2,
+            retainAsPublished=True,
+        )
+        self.client.subscribe(
+            list(map(lambda t: (t, opts), self.subscriptions)),
+        )
+
     # In theory, we would not need this method, as we could very well
     # have set   self.client.on_message = msg_cb   and be done with
     # that. Having this intermediate __on_message() allows us to do
@@ -285,9 +327,13 @@ def __on_message(
             "kind": otel.SpanKind.CONSUMER,
         }
         try:
-            properties = dict(message.properties.UserProperty)
-            span_kwargs["span_links"] = [properties["traceparent"]]
-        except Exception:
+            properties = message.properties
+        except TypeError:
+            properties = None
+        try:
+            user_properties = dict(properties.UserProperty)
+            span_kwargs["span_links"] = [user_properties["traceparent"]]
+        except (AttributeError, KeyError):
             # There was ultimately no traceparent in that message, ignore
             pass
         with self.span_ctxmgr_cb(**span_kwargs) as span:
@@ -297,10 +343,22 @@ def __on_message(
                 key="iot3.core.mqtt.payload_size",
                 value=len(message.payload),
             )
+            if message.retain:
+                span.set_attribute(key="iot3.core.mqtt.retain", value=True)
+                try:
+                    retain = properties.MessageExpiryInterval
+                    if retain < 1:
+                        # Glitch, it should not happen: can't be zero
+                        retain = 1
+                except AttributeError:
+                    retain = True
+            else:
+                retain = False
             self.msg_cb(
                 data=self.msg_cb_data,
                 topic=message.topic,
                 payload=message.payload,
+                retain=retain,
             )
 
     def __on_connect(
@@ -313,6 +371,4 @@ def __on_connect(
     ):
         with self.subscriptions_lock:
             if self.subscriptions:
-                self.client.subscribe(
-                    list(map(lambda t: (t, 0), self.subscriptions)),
-                )
+                self._do_subscribe(topics=self.subscriptions)

python/its-interqueuemanager/its-iqm.cfg

@@ -27,6 +27,41 @@ socket-path = /run/mosquitto/mqtt.socket
 # Name of the local inter-queue; default: interQueue
 #interqueue = interQueue
 
+# Sections which name starts with 'filtering.' (with the dot) followed
+# by an arbitrary identifier, define filtering rules and actions.
+#[filter.NAME]
+# The prefix or regexp to match topic against; only one kind of match can
+# be specified for a given filter (either by prefix or by regexp), but
+# multiple such match can be specified with a multi-line value.
+# Placeholders {{XXX}} will be replaced before attempting a match:
+#  - {{instance-id}}: replaced by general.instance-id
+#  - {{prefix}}: if general.prefix is set, this placeholder is replaced
+#    with it, with a trailing '/' appended; otherwise, an empty string;
+#  - {{suffix}}: replaced by general.suffix if set, an empty string otherwise;
+#  - {{inQueue}}, {{outQueue}}, {{interQueue}}: placeholders matching the
+#    inQueue, outQueue, and interQueue, respectively.
+#in_prefix =
+#   prefix-str-1
+#   prefix-str-2
+#in_regex = regex-1
+#   regex-2
+#out_prefix = prefix-str
+#out_regex = regex
+# What to do with the message:
+# - drop the message if the topic matches:
+#drop
+# - set the message as retained or as not retained:
+#retain = True / False
+# - set the message as retained and set property Message Expiry Interval
+#   to the specified value (in seconds):
+#retain = INT
+# - set the message as retained and set property Message Expiry Interval
+#   to the value from the JSON payload object at the specified dotted
+#   path; if the path does not exist or is not an integer, use the
+#   fallback value if specified, otherwise same as if retain was not
+#   specified:
+#retain = json:DOTTED.PATH.IN.JSON.PAYLOAD [FALLBACK_INT]
+
 [authority]
 # Type of central authority, one of: file, http, mqtt
 type = file

python/its-interqueuemanager/src/its_iqm/filters.py

@@ -0,0 +1,147 @@
+# Software Name: its-interqueuemanager
+# SPDX-FileCopyrightText: Copyright (c) 2025 Orange
+# SPDX-License-Identifier: MIT
+# Author: Yann E. MORIN <yann.morin@orange.com>
+
+import itertools
+import json
+import logging
+import re
+
+
+class Filter:
+    def __init__(
+        self,
+        *,
+        name: str,
+        filter_cfg: dict,
+        instance_id: str,
+        prefix: str,
+        suffix: str,
+        queues: dict,
+    ):
+        self.name = name
+
+        placeholders = {
+            "instance-id": instance_id,
+            "prefix": prefix,
+            "suffix": suffix,
+        }
+        placeholders.update(queues)
+
+        self._patterns = None
+        for filter_type, filter_kind in itertools.product(
+            ["in", "out"],
+            ["prefix", "regex"],
+        ):
+            try:
+                patterns = filter_cfg[filter_type + "_" + filter_kind]
+            except KeyError:
+                continue
+
+            if self._patterns is not None:
+                raise ValueError(f"Filter {name} defines multiple patterns")
+            patterns = list(filter(None, patterns.split("\n")))
+            for ph in placeholders:
+                patterns = list(
+                    map(
+                        lambda s: s.replace("{{" + ph + "}}", placeholders[ph]),
+                        patterns,
+                    ),
+                )
+            if filter_kind == "regex":
+                patterns = list(map(re.compile, patterns))
+
+            self._patterns = patterns
+            self._type = filter_type
+            self._kind = filter_kind
+
+        if patterns is None:
+            raise ValueError(f"Filter '{name}' does not define patterns")
+
+        self._drop = "drop" in filter_cfg
+
+        match filter_cfg.get("retain"):
+            case None:
+                self._retain = None
+            case "True" | "true" | "False" | "false" as b:
+                self._retain = b in ["True", "true"]
+            case str(s) if re.match("^\d+$", s):
+                self._retain = int(s)
+            case str(s) if s.startswith("json:") and re.match(".* \d+$", s):
+                path, fallback = s[5:].split(" ")
+                self._retain = {
+                    "path": path.split("."),
+                    "fallback": int(fallback),
+                }
+            case str(s) if s.startswith("json:"):
+                self._retain = {
+                    "path": s[5:].split("."),
+                    "fallback": None,
+                }
+            case _v:
+                raise ValueError(f"Unable to parse retain value '{_v}'")
+
+        logging.debug(f"Created new filter {name}:")
+        logging.debug(f"  - type: {self._type}")
+        logging.debug(f"  - kind: {self._kind}")
+        logging.debug(f"  - patterns: {self._patterns}")
+        logging.debug(f"  - drop: {self._drop}")
+        logging.debug(f"  - retain: {self._retain}")
+
+    @property
+    def type(self):
+        return self._type
+
+    def apply(
+        self,
+        *,
+        topic: str,
+        payload: bytes,
+        retain: bool | int,
+    ):
+        for pattern in self._patterns:
+            if self._kind == "prefix" and topic.startswith(pattern):
+                break
+            if self._kind == "regex" and pattern.match(topic):
+                break
+        else:
+            logging.debug(f"{self.name}[{self._type}]: no match for {topic}")
+            return topic, payload, retain
+
+        logging.debug(
+            f"{self.name}[{self._type}]: match for {topic} with {self._kind} {pattern}"
+        )
+
+        if self._drop:
+            logging.debug(f"{self.name}:   - dropped")
+            return None, None, None
+
+        match self._retain:
+            case None:
+                logging.debug("{self.name}:   - retain: None")
+            case bool(b):
+                logging.debug(f"{self.name}:   - retain: {b}")
+                retain = b
+            case int(i):
+                logging.debug(f"{self.name}:   - retain: {i}")
+                retain = i
+            case {"path": path, "fallback": fallback}:
+                logging.debug(f"{self.name}:   - retain: {path}")
+                try:
+                    data = json.loads(payload)
+                    for p in path:
+                        data = data[p]
+                except (json.decoder.JSONDecodeError, KeyError, TypeError):
+                    if fallback is not None:
+                        logging.debug(f"{self.name}:     - using fallback {fallback}")
+                        retain = fallback
+                else:
+                    retain = data
+
+        logging.debug(f"{self.name}: Result of filter:")
+        logging.debug(f"{self.name}:   - topic: {topic}")
+        logging.debug(f"{self.name}:   - payload: {payload}")
+        logging.debug(f"{self.name}:   - retain: {retain}")
+
+        return topic, payload, retain