dlq: Add metrics on DLQ buffer to python-arroyo

arroyo/dlq.py

@@ -28,6 +28,7 @@
     TStrategyPayload,
     Value,
 )
+from arroyo.utils.metrics import get_metrics
 
 logger = logging.getLogger(__name__)
 
@@ -287,6 +288,11 @@ def __init__(self, dlq_policy: Optional[DlqPolicy[TStrategyPayload]]) -> None:
         self.__buffered_messages: MutableMapping[
             Partition, Deque[BrokerValue[TStrategyPayload]]
         ] = defaultdict(deque)
+        self.__metrics = get_metrics()
+
+    def report_partition_metrics(self, buffered: Deque[BrokerValue[TStrategyPayload]]) -> None:
+
+        self.__metrics.gauge("arroyo.consumer.dlq_buffer.len", len(buffered))
 
     def append(self, message: BrokerValue[TStrategyPayload]) -> None:
         """
@@ -304,6 +310,7 @@ def append(self, message: BrokerValue[TStrategyPayload]) -> None:
                 buffered.popleft()
 
         self.__buffered_messages[message.partition].append(message)
+        self.report_partition_metrics(self.__buffered_messages[message.partition])
 
     def pop(
         self, partition: Partition, offset: int
@@ -317,9 +324,14 @@ def pop(
 
             while buffered:
                 if buffered[0].offset == offset:
-                    return buffered.popleft()
+                    msg = buffered.popleft()
+                    self.report_partition_metrics(buffered)
+                    return msg
                 if buffered[0].offset > offset:
+                    self.report_partition_metrics(buffered)
                     break
+
+                self.report_partition_metrics(buffered)
                 self.__buffered_messages[partition].popleft()
 
             return None

arroyo/utils/metric_defs.py

@@ -100,5 +100,6 @@
     "arroyo.strategies.filter.dropped_messages",
     # Counter: how many messages are dropped due to errors producing to the dlq
     "arroyo.consumer.dlq.dropped_messages",
-
+    # Gauge: Current length of the DLQ buffer deque
+    "arroyo.consumer.dlq_buffer.len",
 ]