Add SDK metrics hooks

Author: durable-workflow-ops

README.md

@@ -57,6 +57,24 @@ async def main():
 - **Polyglot**: Works alongside PHP workers on the same task queue
 - **HTTP/JSON protocol**: No gRPC, no protobuf dependencies
 - **Codec envelopes**: Avro payloads by default, with JSON decode compatibility for existing history
+- **Metrics hooks**: Pluggable counters and histograms, with an optional Prometheus adapter
+
+## Metrics
+
+Pass a recorder to `Client(metrics=...)` or `Worker(metrics=...)` to collect request, poll, and task metrics. The SDK ships a no-op default, an `InMemoryMetrics` recorder for tests or custom exporter loops, and `PrometheusMetrics` for deployments that install the optional extra:
+
+```bash
+pip install 'durable-workflow[prometheus]'
+```
+
+```python
+from durable_workflow import Client, PrometheusMetrics
+
+metrics = PrometheusMetrics()
+client = Client("http://server:8080", token="dev-token-123", metrics=metrics)
+```
+
+Custom recorders implement `increment(name, value=1.0, tags=None)` and `record(name, value, tags=None)`.
 
 ## Documentation
 

pyproject.toml

@@ -42,6 +42,9 @@ dev = [
     "mypy>=1.10",
     "ruff>=0.4",
 ]
+prometheus = [
+    "prometheus-client>=0.20",
+]
 
 [project.urls]
 Homepage = "https://github.com/durable-workflow/sdk-python"

src/durable_workflow/__init__.py

@@ -40,6 +40,12 @@
     WorkflowNotFound,
     WorkflowTerminated,
 )
+from .metrics import (
+    InMemoryMetrics,
+    MetricsRecorder,
+    NoopMetrics,
+    PrometheusMetrics,
+)
 from .worker import Worker
 from .workflow import ContinueAsNew, StartChildWorkflow
 
@@ -71,8 +77,12 @@
     "workflow",
     "DurableWorkflowError",
     "InvalidArgument",
+    "InMemoryMetrics",
+    "MetricsRecorder",
     "NamespaceNotFound",
+    "NoopMetrics",
     "QueryFailed",
+    "PrometheusMetrics",
     "ServerError",
     "Unauthorized",
     "UpdateRejected",

src/durable_workflow/client.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import time
 from dataclasses import dataclass
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as _pkg_version
@@ -16,6 +17,7 @@
     WorkflowTerminated,
     _raise_for_status,
 )
+from .metrics import CLIENT_REQUEST_DURATION_SECONDS, CLIENT_REQUESTS, NOOP_METRICS, MetricsRecorder
 from .retry_policy import RetryPolicy
 
 PROTOCOL_VERSION = "1.0"
@@ -34,6 +36,28 @@ def _default_sdk_version() -> str:
 DEFAULT_SDK_VERSION = _default_sdk_version()
 
 
+def _route_for_metrics(path: str) -> str:
+    clean_path = path.split("?", 1)[0]
+    parts = [part for part in clean_path.strip("/").split("/") if part]
+    if not parts:
+        return "/"
+
+    if parts[0] == "workflows" and len(parts) >= 2:
+        parts[1] = "{workflow_id}"
+        if len(parts) >= 4 and parts[2] in {"signal", "query", "update"}:
+            parts[3] = "{name}"
+        if len(parts) >= 4 and parts[2] == "runs":
+            parts[3] = "{run_id}"
+    elif parts[0] == "schedules" and len(parts) >= 2:
+        parts[1] = "{schedule_id}"
+    elif (
+        parts[:2] == ["worker", "workflow-tasks"] or parts[:2] == ["worker", "activity-tasks"]
+    ) and len(parts) >= 3:
+        parts[2] = "{task_id}"
+
+    return "/" + "/".join(parts)
+
+
 @dataclass
 class WorkflowExecution:
     workflow_id: str
@@ -250,11 +274,13 @@ def __init__(
         namespace: str = "default",
         timeout: float = 60.0,
         retry_policy: RetryPolicy | None = None,
+        metrics: MetricsRecorder | None = None,
     ) -> None:
         self.base_url = base_url.rstrip("/")
         self.token = token
         self.namespace = namespace
         self.retry_policy = retry_policy or RetryPolicy()
+        self.metrics = metrics or NOOP_METRICS
         self._http = httpx.AsyncClient(base_url=self.base_url, timeout=timeout)
 
     async def aclose(self) -> None:
@@ -287,6 +313,12 @@ async def _request(
         timeout: float | None = None,
         context: str = "",
     ) -> Any:
+        start = time.perf_counter()
+        route = _route_for_metrics(path)
+        plane = "worker" if worker else "control"
+        status_code = "none"
+        outcome = "pending"
+
         async def _do_request() -> httpx.Response:
             resp = await self._http.request(
                 method,
@@ -300,19 +332,40 @@ async def _do_request() -> httpx.Response:
             return resp
 
         try:
-            resp = await self.retry_policy.execute(_do_request)
-        except httpx.HTTPStatusError as exc:
-            # Convert to our custom exception types
             try:
-                body = exc.response.json()
-            except ValueError:
-                body = exc.response.text
-            _raise_for_status(exc.response.status_code, body, context=context)
-            raise  # unreachable, but keeps type checker happy
-
-        if resp.status_code == 204 or not resp.content:
-            return None
-        return resp.json()
+                resp = await self.retry_policy.execute(_do_request)
+            except httpx.HTTPStatusError as exc:
+                status_code = str(exc.response.status_code)
+                outcome = "http_error"
+                # Convert to our custom exception types
+                try:
+                    body = exc.response.json()
+                except ValueError:
+                    body = exc.response.text
+                _raise_for_status(exc.response.status_code, body, context=context)
+                raise  # unreachable, but keeps type checker happy
+
+            status_code = str(resp.status_code)
+            if resp.status_code == 204 or not resp.content:
+                outcome = "ok"
+                return None
+            result = resp.json()
+            outcome = "ok"
+            return result
+        except Exception as exc:
+            if outcome == "pending":
+                outcome = type(exc).__name__
+            raise
+        finally:
+            tags = {
+                "method": method.upper(),
+                "route": route,
+                "plane": plane,
+                "status_code": status_code,
+                "outcome": outcome,
+            }
+            self.metrics.increment(CLIENT_REQUESTS, tags=tags)
+            self.metrics.record(CLIENT_REQUEST_DURATION_SECONDS, time.perf_counter() - start, tags=tags)
 
     async def get_cluster_info(self) -> dict[str, Any]:
         """Fetch server build identity, capabilities, and protocol manifests."""

src/durable_workflow/metrics.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from importlib import import_module
+from typing import Any, Protocol, TypeAlias
+
+MetricTags: TypeAlias = Mapping[str, str]
+MetricKey: TypeAlias = tuple[str, tuple[tuple[str, str], ...]]
+
+CLIENT_REQUESTS = "durable_workflow_client_requests"
+CLIENT_REQUEST_DURATION_SECONDS = "durable_workflow_client_request_duration_seconds"
+WORKER_POLLS = "durable_workflow_worker_polls"
+WORKER_POLL_DURATION_SECONDS = "durable_workflow_worker_poll_duration_seconds"
+WORKER_TASKS = "durable_workflow_worker_tasks"
+WORKER_TASK_DURATION_SECONDS = "durable_workflow_worker_task_duration_seconds"
+
+
+class MetricsRecorder(Protocol):
+    """Pluggable counter and histogram recorder used by the client and worker."""
+
+    def increment(self, name: str, value: float = 1.0, tags: MetricTags | None = None) -> None:
+        """Increment a counter metric."""
+
+    def record(self, name: str, value: float, tags: MetricTags | None = None) -> None:
+        """Record a histogram/sample metric."""
+
+
+class NoopMetrics:
+    """Default metrics recorder that intentionally drops all observations."""
+
+    def increment(self, name: str, value: float = 1.0, tags: MetricTags | None = None) -> None:
+        pass
+
+    def record(self, name: str, value: float, tags: MetricTags | None = None) -> None:
+        pass
+
+
+NOOP_METRICS = NoopMetrics()
+
+
+def _metric_key(name: str, tags: MetricTags | None) -> MetricKey:
+    return name, tuple(sorted((str(k), str(v)) for k, v in (tags or {}).items()))
+
+
+@dataclass
+class InMemoryMetrics:
+    """Simple recorder useful for tests and custom exporter loops."""
+
+    counters: dict[MetricKey, float] = field(default_factory=dict)
+    histograms: dict[MetricKey, list[float]] = field(default_factory=dict)
+
+    def increment(self, name: str, value: float = 1.0, tags: MetricTags | None = None) -> None:
+        key = _metric_key(name, tags)
+        self.counters[key] = self.counters.get(key, 0.0) + value
+
+    def record(self, name: str, value: float, tags: MetricTags | None = None) -> None:
+        self.histograms.setdefault(_metric_key(name, tags), []).append(value)
+
+    def counter_value(self, name: str, tags: MetricTags | None = None) -> float:
+        return self.counters.get(_metric_key(name, tags), 0.0)
+
+    def observations(self, name: str, tags: MetricTags | None = None) -> list[float]:
+        return list(self.histograms.get(_metric_key(name, tags), []))
+
+
+class PrometheusMetrics:
+    """Metrics recorder backed by the optional prometheus-client package."""
+
+    def __init__(self, *, registry: Any | None = None) -> None:
+        try:
+            prometheus_client = import_module("prometheus_client")
+        except ImportError as exc:
+            raise RuntimeError(
+                "PrometheusMetrics requires prometheus-client. "
+                "Install it with `pip install durable-workflow[prometheus]`."
+            ) from exc
+
+        self._counter_cls: Any = prometheus_client.Counter
+        self._histogram_cls: Any = prometheus_client.Histogram
+        self._registry = registry
+        self._counters: dict[str, Any] = {}
+        self._histograms: dict[str, Any] = {}
+        self._label_names: dict[tuple[str, str], tuple[str, ...]] = {}
+
+    def increment(self, name: str, value: float = 1.0, tags: MetricTags | None = None) -> None:
+        tag_values = dict(_metric_key(name, tags)[1])
+        counter = self._metric("counter", name, tuple(tag_values))
+        if tag_values:
+            counter.labels(**tag_values).inc(value)
+        else:
+            counter.inc(value)
+
+    def record(self, name: str, value: float, tags: MetricTags | None = None) -> None:
+        tag_values = dict(_metric_key(name, tags)[1])
+        histogram = self._metric("histogram", name, tuple(tag_values))
+        if tag_values:
+            histogram.labels(**tag_values).observe(value)
+        else:
+            histogram.observe(value)
+
+    def _metric(self, kind: str, name: str, label_names: tuple[str, ...]) -> Any:
+        key = (kind, name)
+        existing = self._label_names.get(key)
+        if existing is not None and existing != label_names:
+            raise ValueError(
+                f"metric {name!r} was already registered as a {kind} "
+                f"with labels {existing!r}; got {label_names!r}"
+            )
+        self._label_names[key] = label_names
+
+        store = self._counters if kind == "counter" else self._histograms
+        if name in store:
+            return store[name]
+
+        kwargs: dict[str, Any] = {}
+        if self._registry is not None:
+            kwargs["registry"] = self._registry
+        metric_cls = self._counter_cls if kind == "counter" else self._histogram_cls
+        metric = metric_cls(name, f"{name} {kind}", label_names, **kwargs)
+        store[name] = metric
+        return metric

src/durable_workflow/sync.py

@@ -17,6 +17,8 @@
     WorkflowHandle,
     WorkflowList,
 )
+from .metrics import MetricsRecorder
+from .retry_policy import RetryPolicy
 
 
 def _run(coro: Any) -> Any:
@@ -139,8 +141,17 @@ def __init__(
         token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
+        retry_policy: RetryPolicy | None = None,
+        metrics: MetricsRecorder | None = None,
     ) -> None:
-        self._async = AsyncClient(base_url, token=token, namespace=namespace, timeout=timeout)
+        self._async = AsyncClient(
+            base_url,
+            token=token,
+            namespace=namespace,
+            timeout=timeout,
+            retry_policy=retry_policy,
+            metrics=metrics,
+        )
 
     def close(self) -> None:
         _run(self._async.aclose())