more redis cleanup

jp-agenta · jp-agenta · commit 4aaa55db2398 · 2026-07-15T17:43:45.000+02:00
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/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/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
@@ -1,12 +1,10 @@
 import os
 import re
 import sys
-import time
 import logging
 from typing import Any, Optional
 
 import structlog
-from orjson import dumps as orjson_dumps
 from structlog.typing import EventDict, WrappedLogger, Processor
 
 from oss.src.utils.env import env
@@ -34,8 +32,6 @@ def bound_logger_trace(self, *args, **kwargs):
 # ENV VARS
 AGENTA_LOG_CONSOLE_ENABLED = env.agenta.logging.console_enabled
 AGENTA_LOG_CONSOLE_LEVEL = env.agenta.logging.console_level
-AGENTA_LOG_METRICS_ENABLED = env.agenta.logging.metrics_enabled
-AGENTA_LOG_METRICS_NAMESPACE = env.agenta.logging.metrics_namespace
 
 # COLORS
 LEVEL_COLORS = {
@@ -189,54 +185,6 @@ def render(_, __, event_dict: EventDict) -> str:
         sdk_logger.propagate = False
 
 
-def _emf_unit_for(field: str) -> str:
-    if field.endswith("_ms"):
-        return "Milliseconds"
-    if field == "bytes" or field.endswith("_bytes"):
-        return "Bytes"
-    return "None"
-
-
-def _emit_emf_metric(
-    name: str,
-    *,
-    count: float,
-    unit: str,
-    dims: dict,
-    fields: dict,
-) -> None:
-    """Write one CloudWatch Embedded Metric Format line to stdout.
-
-    No new port/infra: the CloudWatch agent already ships container stdout, and
-    EMF lines embedded in it become metrics automatically.
-    """
-    dim_names = list(dims.keys())
-    # Register every numeric field (duration_ms, bytes, ...) as its own metric so
-    # it graphs in CloudWatch; a name-based unit keeps ms/bytes sensible. Non-numeric
-    # fields stay as EMF properties (context only).
-    metrics = [{"Name": name, "Unit": unit}]
-    for key, value in fields.items():
-        if isinstance(value, (int, float)) and not isinstance(value, bool):
-            metrics.append({"Name": key, "Unit": _emf_unit_for(key)})
-    record = {
-        "_aws": {
-            "Timestamp": int(time.time() * 1000),
-            "CloudWatchMetrics": [
-                {
-                    "Namespace": AGENTA_LOG_METRICS_NAMESPACE,
-                    "Dimensions": [dim_names] if dim_names else [[]],
-                    "Metrics": metrics,
-                }
-            ],
-        },
-        name: count,
-        **dims,
-        **fields,
-    }
-    sys.stdout.write(orjson_dumps(record).decode() + "\n")
-    sys.stdout.flush()
-
-
 class MultiLogger:
     def __init__(self, *loggers: structlog.stdlib.BoundLogger):
         self._loggers = loggers
@@ -245,30 +193,6 @@ def _log(self, level: str, *args: Any, **kwargs: Any):
         for lgr in self._loggers:
             getattr(lgr, level)(*args, **kwargs)
 
-    def tick(
-        self,
-        name: str,
-        *,
-        count: float = 1,
-        unit: str = "Count",
-        dims: Optional[dict] = None,
-        **fields: Any,
-    ) -> None:
-        """Emit a CloudWatch EMF metric line beside the adjacent log call.
-
-        `dims` (e.g. `{"stream": "spans"}`) become CloudWatch dimensions;
-        `**fields` are extra structured context only (not dimensioned).
-        """
-        if not AGENTA_LOG_METRICS_ENABLED:
-            return
-        try:
-            _emit_emf_metric(
-                name, count=count, unit=unit, dims=dims or {}, fields=fields
-            )
-        except Exception:
-            # Metrics must never break the calling code path.
-            pass
-
     def debug(self, *a, **k):
         self._log("debug", *a, **k)
 

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,`