Merge pull request #5335 from Agenta-AI/chore/fix-redis-exporter-stuff

junaway · web-flow · commit 5b2f8d11f2f0 · 2026-07-15T18:32:13.000+02:00
[chore] Clean up redis exporter
diff --git a/api/entrypoints/worker_queues.py b/api/entrypoints/worker_queues.py
@@ -31,7 +31,8 @@
 from taskiq import AsyncBroker, TaskiqEvents
 from taskiq.receiver import Receiver
 from taskiq.cli.worker.run import shutdown_broker
-from taskiq_redis import RedisStreamBroker
+
+from oss.src.tasks.taskiq.shared.broker import TrimOnAckRedisStreamBroker
 
 from oss.src.core.embeds.service import EmbedsService
 from oss.src.core.environments.service import EnvironmentsService
@@ -123,7 +124,7 @@ def _selected_queues() -> List[str]:
 
 
 def _build_webhooks_broker() -> tuple[AsyncBroker, int]:
-    broker = RedisStreamBroker(
+    broker = TrimOnAckRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:webhooks",
         consumer_group_name="worker-webhooks",
@@ -135,7 +136,7 @@ def _build_webhooks_broker() -> tuple[AsyncBroker, int]:
 
 
 def _build_triggers_broker() -> tuple[AsyncBroker, int]:
-    broker = RedisStreamBroker(
+    broker = TrimOnAckRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:triggers",
         consumer_group_name="worker-triggers",
@@ -175,7 +176,7 @@ def _build_triggers_broker() -> tuple[AsyncBroker, int]:
 
 
 def _build_interactions_broker() -> tuple[AsyncBroker, int]:
-    broker = RedisStreamBroker(
+    broker = TrimOnAckRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:interactions",
         consumer_group_name="worker-interactions",
diff --git a/api/entrypoints/worker_streams.py b/api/entrypoints/worker_streams.py
@@ -17,7 +17,8 @@
 from typing import List
 
 from redis.asyncio import Redis
-from taskiq_redis import RedisStreamBroker
+
+from oss.src.tasks.taskiq.shared.broker import ProducerOnlyRedisStreamBroker
 
 from oss.src.core.events.service import EventsService
 from oss.src.core.secrets.services import VaultService
@@ -85,9 +86,11 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer:
 async def _build_events_worker(redis_client: Redis) -> StreamConsumer:
     events_service = EventsService(events_dao=EventsDAO())
 
-    # Webhook dispatch runs inside the events loop as its post-hook.
+    # Webhook dispatch runs inside the events loop as its post-hook: this broker
+    # only produces (.kiq), so it must not declare a consumer group it never
+    # reads — that group sits at 0-0 and reports lag == XLEN forever.
     webhooks_dao = WebhooksDAO()
-    broker = RedisStreamBroker(
+    broker = ProducerOnlyRedisStreamBroker(
         url=env.redis.uri_durable,
         queue_name="queues:webhooks",
         consumer_group_name="worker-events-webhooks-dispatcher",
diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py
@@ -850,7 +850,6 @@ async def respond_interaction(
                 interaction_id=str(interaction_id),
                 answer=answer,
             )
-            log.tick("interactions.enqueued", dims={"queue": "interactions"})
         else:
             references = (
                 {
diff --git a/api/oss/src/apis/fastapi/triggers/router.py b/api/oss/src/apis/fastapi/triggers/router.py
@@ -1602,10 +1602,8 @@ async def ingest_composio_event(
                     ),
                     timeout=_ENQUEUE_TIMEOUT_SECONDS,
                 )
-                log.tick("triggers.enqueued", dims={"queue": "triggers"})
             except Exception as e:
                 log.error("Failed to enqueue trigger event: %s", e)
-                log.tick("triggers.enqueue_errors", dims={"queue": "triggers"})
                 raise HTTPException(
                     status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                     detail="Failed to enqueue trigger event",
diff --git a/api/oss/src/core/evaluations/runtime/broker.py b/api/oss/src/core/evaluations/runtime/broker.py
@@ -6,8 +6,8 @@
 """
 
 from taskiq import AsyncBroker
-from taskiq_redis import RedisStreamBroker
 
+from oss.src.tasks.taskiq.shared.broker import TrimOnAckRedisStreamBroker
 from oss.src.core.evaluations.runtime.runner import TaskiqEvaluationTaskRunner
 from oss.src.core.evaluations.service import EvaluationsService
 from oss.src.core.evaluators.service import SimpleEvaluatorsService
@@ -22,12 +22,13 @@
 MAXLEN_QUEUES_EVALUATIONS = 100_000
 
 
-class NoRedeliveryRedisStreamBroker(RedisStreamBroker):
+class NoRedeliveryRedisStreamBroker(TrimOnAckRedisStreamBroker):
     """Stream broker that never redelivers. `listen()` reads only NEW messages
     (`>`) and skips the XAUTOCLAIM pending-replay block, so a task that crashed
     mid-run is not re-served to later workers. Evaluation tasks are not safely
     re-runnable (`retry_on_error=False`); a stuck unacked entry replaying on every
-    worker restart is worse than dropping it.
+    worker restart is worse than dropping it. Inherits XDEL-on-ack so completed
+    entries leave the stream (XLEN = backlog).
     """
 
     async def listen(self):
diff --git a/api/oss/src/core/evaluations/runtime/runner.py b/api/oss/src/core/evaluations/runtime/runner.py
@@ -38,11 +38,6 @@ async def process_run_from_source(
             kwargs["oldest"] = oldest
 
         result = await self.worker.process_run_from_source.kiq(**kwargs)
-        log.tick(
-            "evaluations.enqueued",
-            dims={"queue": "evaluations"},
-            task="run_from_source",
-        )
         return result
 
     async def process_run_from_batch(
@@ -76,9 +71,6 @@ async def process_run_from_batch(
             kwargs["input_step_key"] = input_step_key
 
         result = await self.worker.process_run_from_batch.kiq(**kwargs)
-        log.tick(
-            "evaluations.enqueued", dims={"queue": "evaluations"}, task="run_from_batch"
-        )
         return result
 
     async def process_rerun(
@@ -116,5 +108,4 @@ async def process_rerun(
             kwargs["overwrite"] = overwrite
 
         result = await self.worker.process_rerun.kiq(**kwargs)
-        log.tick("evaluations.enqueued", dims={"queue": "evaluations"}, task="rerun")
         return result
diff --git a/api/oss/src/core/events/streaming.py b/api/oss/src/core/events/streaming.py
@@ -89,13 +89,7 @@ async def publish_event(
             maxlen=MAXLEN_STREAMS_EVENTS,
             approximate=True,
         )
-        log.tick(
-            "events.published",
-            bytes=len(event_bytes),
-            dims={"stream": "events"},
-        )
         return True
     except Exception as e:
         log.error(f"[EVENTS] Failed to publish event: {e}", exc_info=True)
-        log.tick("events.publish_errors", dims={"stream": "events"})
         return False
diff --git a/api/oss/src/core/sessions/records/streaming.py b/api/oss/src/core/sessions/records/streaming.py
@@ -91,13 +91,7 @@ async def publish_record(
             maxlen=MAXLEN_STREAMS_RECORDS,
             approximate=True,
         )
-        log.tick(
-            "records.published",
-            bytes=len(event_bytes),
-            dims={"stream": "records"},
-        )
         return True
     except Exception as e:
         log.error(f"[RECORDS] Failed to publish: {e}", exc_info=True)
-        log.tick("records.publish_errors", dims={"stream": "records"})
         return False
diff --git a/api/oss/src/core/tracing/streaming.py b/api/oss/src/core/tracing/streaming.py
@@ -1,5 +1,4 @@
 import zlib
-from time import perf_counter
 from typing import List
 from uuid import UUID
 
@@ -84,7 +83,6 @@ async def publish_spans(
 
     count = 0
     total_bytes = 0
-    started = perf_counter()
 
     # Full span_dtos list is already in hand, so pipeline all XADDs into one
     # round-trip instead of one XADD per span on the firehose.
@@ -111,12 +109,4 @@ async def publish_spans(
         if count:
             await pipe.execute()
 
-    log.tick(
-        "spans.published",
-        count=count,
-        bytes=total_bytes,
-        duration_ms=(perf_counter() - started) * 1000,
-        dims={"stream": "spans"},
-    )
-
     return count
diff --git a/api/oss/src/core/triggers/service.py b/api/oss/src/core/triggers/service.py
@@ -1560,7 +1560,6 @@ async def refresh_schedules(
                     timeout=_ENQUEUE_TIMEOUT_SECONDS,
                 )
 
-                log.tick("triggers.enqueued", dims={"queue": "triggers"})
                 log.info(
                     "[SCHEDULE] Dispatched.   ",
                     project_id=project_id,
diff --git a/api/oss/src/tasks/asyncio/events/worker.py b/api/oss/src/tasks/asyncio/events/worker.py
@@ -1,5 +1,4 @@
 import asyncio
-import time
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
 from uuid import UUID
 
@@ -218,15 +217,11 @@ async def run(self):
 
         while True:
             try:
-                # 1. Read batch from stream (idle blocks return empty, never tick)
-                read_started = time.perf_counter()
+                # 1. Read batch from stream
                 batch = await self.read_batch()
                 if not batch:
                     continue
 
-                read_ms = (time.perf_counter() - read_started) * 1000
-                started = time.perf_counter()
-
                 # 2. Process batch
                 ingested, processed_ids, batches = await self.process_batch(batch)
 
@@ -245,10 +240,6 @@ async def run(self):
                             "[EVENTS] Skipping ACK/DEL due to webhook dispatch failure",
                             batch_size=len(batch),
                         )
-                        log.tick(
-                            f"{self.metric_stream}.errors",
-                            dims={"stream": self.metric_stream},
-                        )
                         continue
                 else:
                     log.info(
@@ -264,16 +255,6 @@ async def run(self):
                     batch_size=len(batch),
                     ingested=ingested,
                 )
-                log.tick(
-                    f"{self.metric_stream}.processed",
-                    count=len(processed_ids),
-                    duration_ms=(time.perf_counter() - started) * 1000,
-                    read_ms=read_ms,
-                    dims={"stream": self.metric_stream},
-                )
             except Exception:
                 log.error("[EVENTS] Worker loop error", exc_info=True)
-                log.tick(
-                    f"{self.metric_stream}.errors", dims={"stream": self.metric_stream}
-                )
                 await asyncio.sleep(1)
diff --git a/api/oss/src/tasks/asyncio/shared/consumer.py b/api/oss/src/tasks/asyncio/shared/consumer.py
@@ -182,37 +182,19 @@ async def run(self):
 
         while True:
             try:
-                read_started = time.perf_counter()
                 batch = await self.read_batch()
                 if not batch:
                     continue
 
-                # read_ms = fetch + batch accumulation (the consumer's Redis-read
-                # latency, symmetric with the producer's XADD latency); duration_ms
-                # = process + ack only. Idle 5s blocks return empty and never tick.
-                read_ms = (time.perf_counter() - read_started) * 1000
-                started = time.perf_counter()
-                processed_count, processed_message_ids = await self.process_batch(batch)
+                _, processed_message_ids = await self.process_batch(batch)
 
                 if processed_message_ids:
                     await self.ack_and_delete(processed_message_ids)
 
-                log.tick(
-                    f"{self.metric_stream}.processed",
-                    count=processed_count,
-                    duration_ms=(time.perf_counter() - started) * 1000,
-                    read_ms=read_ms,
-                    dims={"stream": self.metric_stream},
-                )
-
             except Exception:
                 log.error(
                     f"{self.log_prefix} Error in worker loop",
                     exc_info=True,
                 )
-                log.tick(
-                    f"{self.metric_stream}.errors",
-                    dims={"stream": self.metric_stream},
-                )
                 # Sleep before retry to avoid tight error loop
                 await asyncio.sleep(1)
diff --git a/api/oss/src/tasks/asyncio/webhooks/dispatcher.py b/api/oss/src/tasks/asyncio/webhooks/dispatcher.py
@@ -321,21 +321,14 @@ async def dispatch(
                             f"delivery={delivery_id} event={event.event_id} "
                             f"subscription={sub.id}"
                         )
-                        log.tick("webhooks.enqueued", dims={"queue": "webhooks"})
                     except Exception as e:
                         log.error(
                             f"[WEBHOOKS DISPATCHER] Failed to enqueue delivery "
                             f"for subscription {sub.id}: {e}"
                         )
-                        log.tick("webhooks.enqueue_errors", dims={"queue": "webhooks"})
                         enqueue_failures += 1
 
         if enqueue_failures > 0:
-            log.tick(
-                "webhooks.enqueue_failures",
-                count=enqueue_failures,
-                dims={"queue": "webhooks"},
-            )
             # Raise so the events worker skips ACK/DEL and retries the whole batch. Without this
             # the batch is acked and the failed deliveries are dropped for good. Redelivery is
             # safe: the deterministic delivery_id dedups already-enqueued deliveries.
diff --git a/api/oss/src/tasks/taskiq/shared/__init__.py b/api/oss/src/tasks/taskiq/shared/__init__.py
diff --git a/api/oss/src/tasks/taskiq/shared/broker.py b/api/oss/src/tasks/taskiq/shared/broker.py
@@ -0,0 +1,35 @@
+"""Shared TaskIQ RedisStreamBroker mixins.
+
+`TrimOnAckRedisStreamBroker` XDELs each entry right after XACK. Without it, a
+TaskIQ queue retains every acked entry until maxlen-trims it, so XLEN reflects
+retained history (a rolling maxlen window) rather than backlog — the stream
+workers already XACK+XDEL, so this makes the queues match. XDEL fires only on
+ack (a terminal state), so retries (re-kicked as new messages) and crash
+redelivery (unacked, reclaimed via XAUTOCLAIM from the PEL) are unaffected.
+
+`ProducerOnlyRedisStreamBroker` skips consumer-group declaration on startup, for
+a broker used only to register tasks and `.kiq()` — never to `.listen()`. The
+base `startup()` unconditionally XGROUP-CREATEs a group that then sits at 0-0
+forever (0 consumers), reporting lag == XLEN permanently.
+"""
+
+from typing import Awaitable, Callable
+
+from redis.asyncio import Redis
+from taskiq_redis import RedisStreamBroker
+from taskiq_redis.redis_broker import BaseRedisBroker
+
+
+class TrimOnAckRedisStreamBroker(RedisStreamBroker):
+    def _ack_generator(self, id: str, queue_name: str) -> Callable[[], Awaitable[None]]:
+        async def _ack() -> None:
+            async with Redis(connection_pool=self.connection_pool) as redis_conn:
+                await redis_conn.xack(queue_name, self.consumer_group_name, id)
+                await redis_conn.xdel(queue_name, id)
+
+        return _ack
+
+
+class ProducerOnlyRedisStreamBroker(RedisStreamBroker):
+    async def startup(self) -> None:
+        await BaseRedisBroker.startup(self)
diff --git a/api/oss/src/utils/env.py b/api/oss/src/utils/env.py
@@ -321,15 +321,6 @@ class LoggingConfig(BaseModel):
         or "INFO"
     ).upper()
 
-    # EMF metric lines (`log.tick(...)`) to stdout — the CloudWatch agent already
-    # ships container stdout, so this needs no new port/infra to light up.
-    metrics_enabled: bool = (
-        os.getenv("AGENTA_LOGGING_METRICS_ENABLED") or "true"
-    ).lower() in _TRUTHY
-    metrics_namespace: str = (
-        os.getenv("AGENTA_LOGGING_METRICS_NAMESPACE") or "Agenta/Workers"
-    )
-
     model_config = ConfigDict(extra="ignore")
 
 
diff --git a/api/oss/src/utils/logging.py b/api/oss/src/utils/logging.py

Original file line number	Diff line number	Diff line change
`@@ -850,7 +850,6 @@ async def respond_interaction(`
`850`	`850`	`interaction_id=str(interaction_id),`
`851`	`851`	`answer=answer,`
`852`	`852`	`)`
`853`		`- log.tick("interactions.enqueued", dims={"queue": "interactions"})`
`854`	`853`	`else:`
`855`	`854`	`references = (`
`856`	`855`	`{`
Original file line number	Diff line number	Diff line change
`@@ -1560,7 +1560,6 @@ async def refresh_schedules(`
`1560`	`1560`	`timeout=_ENQUEUE_TIMEOUT_SECONDS,`
`1561`	`1561`	`)`
`1562`	`1562`
`1563`		`- log.tick("triggers.enqueued", dims={"queue": "triggers"})`
`1564`	`1563`	`log.info(`
`1565`	`1564`	`"[SCHEDULE] Dispatched. ",`
`1566`	`1565`	`project_id=project_id,`