chore(producer): Add produce time by topic

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
       - id: fix-encoding-pragma
         args: ["--remove"]
   - repo: https://github.com/pycqa/flake8
-    rev: 3.8.4
+    rev: 7.1.1
     hooks:
       - id: flake8
         language_version: python3.12

arroyo/backends/kafka/consumer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from concurrent.futures import Future
 from datetime import datetime
 from enum import Enum
@@ -45,6 +46,7 @@
 )
 from arroyo.types import BrokerValue, Partition, Topic
 from arroyo.utils.concurrent import execute
+from arroyo.utils.metrics import get_metrics
 from arroyo.utils.retries import BasicRetryPolicy
 
 logger = logging.getLogger(__name__)
@@ -655,6 +657,9 @@ def __init__(self, configuration: Mapping[str, Any]) -> None:
         # are fired -- otherwise trying to produce "synchronously" via
         # ``produce(...).result()`` could result in a deadlock.
         self.__result = execute(self.__worker, daemon=True)
+        self.__metrics = get_metrics()
+        self.__last_produce_time_record = 0.0
+        self.__max_produce_time = 0.0
 
     def __worker(self) -> None:
         """
@@ -668,13 +673,33 @@ def __worker(self) -> None:
             self.__producer.poll(0.1)
         self.__producer.flush()
 
+    def __record_produce_time(self, topic: str, start_time: float) -> None:
+        now = time.time()
+        duration = now - start_time
+        self.__max_produce_time = max(self.__max_produce_time, duration)
+
+        if now - self.__last_produce_time_record <= 0.1:
+            return
+
+        self.__metrics.timing(
+            "arroyo.producer.max_produce_time",
+            self.__max_produce_time,
+            tags={"topic": topic},
+        )
+
+        self.__max_produce_time = 0.0
+        self.__last_produce_time_record = now
+
     def __delivery_callback(
         self,
         future: Future[BrokerValue[KafkaPayload]],
         payload: KafkaPayload,
         error: KafkaError,
         message: ConfluentMessage,
+        start_produce_time: float,
     ) -> None:
+        self.__record_produce_time(message.topic(), start_produce_time)
+
         if error is not None:
             future.set_exception(TransportError(error))
         else:
@@ -699,6 +724,8 @@ def produce(
         destination: Union[Topic, Partition],
         payload: KafkaPayload,
     ) -> Future[BrokerValue[KafkaPayload]]:
+        start_produce_time = time.time()
+
         if self.__shutdown_requested.is_set():
             raise RuntimeError("producer has been closed")
 
@@ -719,7 +746,9 @@ def produce(
             value=payload.value,
             key=payload.key,
             headers=payload.headers,
-            on_delivery=partial(self.__delivery_callback, future, payload),
+            on_delivery=partial(
+                self.__delivery_callback, future, payload, start_produce_time
+            ),
         )
         return future
 

arroyo/utils/metric_defs.py

@@ -98,4 +98,7 @@
     "arroyo.processing.strategies.healthcheck.touch",
     # Counter: Number of messages dropped in the FilterStep strategy
     "arroyo.strategies.filter.dropped_messages",
+    # Time: e2e time it takes to produce a message. Tagged by
+    # physical topic name.
+    "arroyo.producer.max_produce_time",
 ]