determine metrics dir early

Author: chrishagglund-ship-it

src/conductor/client/automator/task_handler.py

@@ -232,12 +232,9 @@ def __init__(
         self._configuration = configuration
         self._metrics_settings = metrics_settings
 
-        # Resolve the metrics subdirectory and set PROMETHEUS_MULTIPROC_DIR
-        # BEFORE any worker processes start.  resolve_metrics_type is
-        # idempotent so it's safe if the caller already called it.
+        # Set PROMETHEUS_MULTIPROC_DIR BEFORE any worker processes start.
+        # MetricsSettings resolves the subdirectory eagerly at construction.
         if metrics_settings is not None:
-            from conductor.client.telemetry.metrics_factory import resolve_metrics_type
-            resolve_metrics_type(metrics_settings)
             os.environ["PROMETHEUS_MULTIPROC_DIR"] = metrics_settings.metrics_directory
             logger.info(f"Set PROMETHEUS_MULTIPROC_DIR={metrics_settings.metrics_directory}")
 

src/conductor/client/configuration/settings/metrics_settings.py

@@ -14,6 +14,16 @@
 )
 
 
+CANONICAL_SUBDIR = "canonical"
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    value = os.environ.get(name, "")
+    if not value:
+        return default
+    return value.strip().lower() in ("true", "1", "yes")
+
+
 def get_default_temporary_folder() -> str:
     return f"{Path.home()!s}/tmp/"
 
@@ -30,6 +40,11 @@ def __init__(
         """
         Configure metrics collection settings.
 
+        The ``WORKER_CANONICAL_METRICS`` env var is read at construction time
+        to decide whether ``.db`` files go in *directory* (legacy) or in a
+        ``canonical/`` subdirectory (canonical mode).  Set the env var before
+        creating this object.
+
         Args:
             directory: Base directory for storing multiprocess metrics .db files.
                       Legacy metrics use this directory directly (unchanged from
@@ -62,16 +77,24 @@ def __init__(
         self.http_port = http_port
         self.clean_directory = clean_directory
         self.clean_dead_pids = clean_dead_pids
-        self._subdir: Optional[str] = None
+        self._subdir: str = (
+            CANONICAL_SUBDIR
+            if _env_bool("WORKER_CANONICAL_METRICS", False)
+            else ""
+        )
+
+    @property
+    def is_canonical(self) -> bool:
+        return self._subdir == CANONICAL_SUBDIR
 
     @property
     def metrics_directory(self) -> str:
         """Full path where .db files live (base directory + optional subdir).
 
-        The subdirectory is set by ``metrics_factory.resolve_metrics_type``
-        before any child processes are forked.  Legacy leaves it empty (base
-        directory unchanged from prior releases); canonical sets it to
-        ``"canonical"`` to avoid stale metric-name collisions.
+        Legacy leaves the subdirectory empty (base directory unchanged from
+        prior releases); canonical sets it to ``"canonical"`` to avoid stale
+        metric-name collisions.  Resolved eagerly at construction time from
+        the ``WORKER_CANONICAL_METRICS`` env var.
         """
         if self._subdir:
             return os.path.join(self.directory, self._subdir)

src/conductor/client/telemetry/canonical_metrics_collector.py

@@ -11,6 +11,8 @@
 WORKER_CANONICAL_METRICS=true.
 """
 
+from typing import Optional
+
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from conductor.client.telemetry.metrics_collector_base import (
     MetricsCollectorBase,
@@ -193,7 +195,7 @@ def record_task_update_time(self, task_type: str, time_spent: float, status: str
         )
 
     def record_api_request_time(self, method: str, uri: str, status: str, time_spent: float,
-                                metric_uri: str = None) -> None:
+                                metric_uri: Optional[str] = None) -> None:
         resolved_uri = (metric_uri or uri).replace("{tasktype}", "{taskType}")
         self._observe_histogram(
             name=MetricName.API_REQUEST_TIME_CANONICAL,

src/conductor/client/telemetry/legacy_metrics_collector.py

@@ -7,7 +7,7 @@
 """
 
 from collections import deque
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from conductor.client.telemetry import metrics_collector_base as _mcb
@@ -176,7 +176,7 @@ def record_task_update_time(self, task_type: str, time_spent: float, status: str
         pass  # canonical-only; call sites were added in this branch and didn't exist on main
 
     def record_api_request_time(self, method: str, uri: str, status: str, time_spent: float,
-                                metric_uri: str = None) -> None:
+                                metric_uri: Optional[str] = None) -> None:
         self._record_quantiles(
             name=MetricName.API_REQUEST_TIME,
             documentation=MetricDocumentation.API_REQUEST_TIME,

src/conductor/client/telemetry/metrics_collector_base.py

@@ -5,7 +5,7 @@
 import signal
 import threading
 import time
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 # Lazy imports - these will be imported when first needed.
 # PROMETHEUS_MULTIPROC_DIR must be set before prometheus_client is imported.
@@ -419,7 +419,7 @@ def record_task_update_time(self, task_type: str, time_spent: float, status: str
 
     @abc.abstractmethod
     def record_api_request_time(self, method: str, uri: str, status: str, time_spent: float,
-                                metric_uri: str = None) -> None: ...
+                                metric_uri: Optional[str] = None) -> None: ...
 
     @abc.abstractmethod
     def record_task_result_payload_size(self, task_type: str, payload_size: int) -> None: ...

src/conductor/client/telemetry/metrics_factory.py

@@ -22,16 +22,6 @@
 )
 
 
-_CANONICAL_SUBDIR = "canonical"
-
-
-def _env_bool(name: str, default: bool) -> bool:
-    value = os.environ.get(name, "")
-    if not value:
-        return default
-    return value.strip().lower() in ("true", "1", "yes")
-
-
 _cleaned_directories: set = set()
 
 
@@ -42,35 +32,16 @@ def _reset_cleaned_directories() -> None:
     _cleaned_directories.clear()
 
 
-def resolve_metrics_type(settings: MetricsSettings) -> None:
-    """Read ``WORKER_CANONICAL_METRICS`` once and store the result on *settings*.
-
-    Idempotent -- subsequent calls are no-ops.  Must be called before
-    ``settings.metrics_directory`` is read so that the subdirectory is
-    resolved in the main process before any child processes are forked.
-
-    Both ``TaskHandler.__init__`` and ``create_metrics_collector`` call this.
-    """
-    if settings._subdir is not None:
-        return
-    if _env_bool("WORKER_CANONICAL_METRICS", default=False):
-        settings._subdir = _CANONICAL_SUBDIR
-    else:
-        settings._subdir = ""
-
-
 def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
     """
-    Create the metrics collector selected by environment variables.
+    Create the metrics collector indicated by ``settings.is_canonical``.
 
-    Calls ``resolve_metrics_type`` to ensure ``settings.metrics_directory``
-    returns the correct path, then instantiates the appropriate collector.
+    ``MetricsSettings`` reads ``WORKER_CANONICAL_METRICS`` at construction
+    time, so the directory and collector type are already determined.
 
     Returns a fully-initialised collector (legacy or canonical) that satisfies
     the MetricsCollector Protocol and can be registered as an event listener.
     """
-    resolve_metrics_type(settings)
-
     metrics_dir = settings.metrics_directory
     os.makedirs(metrics_dir, exist_ok=True)
 
@@ -81,7 +52,7 @@ def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
         if settings.clean_dead_pids:
             settings._clean_dead_pid_files()
 
-    if settings._subdir == _CANONICAL_SUBDIR:
+    if settings.is_canonical:
         from conductor.client.telemetry.canonical_metrics_collector import CanonicalMetricsCollector
         logger.info("WORKER_CANONICAL_METRICS is true — using CanonicalMetricsCollector (dir=%s)", metrics_dir)
         return CanonicalMetricsCollector(settings)

tests/unit/api_client/test_async_api_client.py

@@ -0,0 +1,58 @@
+import asyncio
+import unittest
+from unittest.mock import Mock, AsyncMock
+
+from conductor.client.http.async_api_client import AsyncApiClient
+from conductor.client.configuration.configuration import Configuration
+
+
+class TestAsyncApiClientMetricUri(unittest.TestCase):
+
+    def setUp(self):
+        self.config = Configuration(
+            base_url="http://localhost:8080",
+            authentication_settings=None,
+        )
+
+    def _run(self, coro):
+        loop = asyncio.new_event_loop()
+        try:
+            return loop.run_until_complete(coro)
+        finally:
+            loop.close()
+
+    def test_request_passes_metric_uri_to_collector(self):
+        """metric_uri kwarg is forwarded to record_api_request_time on success."""
+        metrics_collector = Mock()
+        client = AsyncApiClient(configuration=self.config, metrics_collector=metrics_collector)
+        client.async_rest_client.GET = AsyncMock(return_value=Mock(status=200))
+
+        self._run(client.request(
+            'GET', 'http://localhost:8080/api/workflow/test-id',
+            metric_uri='/api/workflow/{workflowId}',
+        ))
+
+        call_args = metrics_collector.record_api_request_time.call_args
+        self.assertEqual(call_args[1]['metric_uri'], '/api/workflow/{workflowId}')
+
+    def test_request_passes_metric_uri_to_collector_on_error(self):
+        """metric_uri kwarg is forwarded to record_api_request_time on error."""
+        metrics_collector = Mock()
+        client = AsyncApiClient(configuration=self.config, metrics_collector=metrics_collector)
+
+        error = Exception('Test error')
+        error.status = 500
+        client.async_rest_client.GET = AsyncMock(side_effect=error)
+
+        with self.assertRaises(Exception):
+            self._run(client.request(
+                'GET', 'http://localhost:8080/api/workflow/test-id',
+                metric_uri='/api/workflow/{workflowId}',
+            ))
+
+        call_args = metrics_collector.record_api_request_time.call_args
+        self.assertEqual(call_args[1]['metric_uri'], '/api/workflow/{workflowId}')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/unit/telemetry/test_metrics_factory.py

@@ -21,11 +21,10 @@ class TestMetricsFactory(unittest.TestCase):
 
     def setUp(self):
         _reset_cleaned_directories()
-        self.metrics_dir = tempfile.mkdtemp()
-        self.settings = MetricsSettings(directory=self.metrics_dir)
         self._saved_env = {}
         for key in ("WORKER_CANONICAL_METRICS", "WORKER_LEGACY_METRICS"):
             self._saved_env[key] = os.environ.pop(key, None)
+        self.metrics_dir = tempfile.mkdtemp()
 
     def tearDown(self):
         for key, val in self._saved_env.items():
@@ -37,62 +36,62 @@ def tearDown(self):
             shutil.rmtree(self.metrics_dir)
         _reset_cleaned_directories()
 
+    def _make_settings(self, **kwargs):
+        kwargs.setdefault("directory", self.metrics_dir)
+        return MetricsSettings(**kwargs)
+
     def test_default_returns_legacy(self):
         """With no env vars set, factory returns LegacyMetricsCollector."""
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertIsInstance(collector, LegacyMetricsCollector)
 
     def test_canonical_true_returns_canonical(self):
         """WORKER_CANONICAL_METRICS=true selects CanonicalMetricsCollector."""
         os.environ["WORKER_CANONICAL_METRICS"] = "true"
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertIsInstance(collector, CanonicalMetricsCollector)
 
     def test_canonical_1_returns_canonical(self):
         """WORKER_CANONICAL_METRICS=1 selects CanonicalMetricsCollector."""
         os.environ["WORKER_CANONICAL_METRICS"] = "1"
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertIsInstance(collector, CanonicalMetricsCollector)
 
     def test_canonical_yes_returns_canonical(self):
         """WORKER_CANONICAL_METRICS=yes selects CanonicalMetricsCollector."""
         os.environ["WORKER_CANONICAL_METRICS"] = "yes"
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertIsInstance(collector, CanonicalMetricsCollector)
 
     def test_canonical_false_returns_legacy(self):
         """WORKER_CANONICAL_METRICS=false selects LegacyMetricsCollector."""
         os.environ["WORKER_CANONICAL_METRICS"] = "false"
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertIsInstance(collector, LegacyMetricsCollector)
 
     def test_canonical_takes_priority_over_legacy(self):
         """WORKER_CANONICAL_METRICS=true wins even if WORKER_LEGACY_METRICS=true."""
         os.environ["WORKER_CANONICAL_METRICS"] = "true"
         os.environ["WORKER_LEGACY_METRICS"] = "true"
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertIsInstance(collector, CanonicalMetricsCollector)
 
     def test_legacy_collector_name(self):
         """LegacyMetricsCollector.collector_name() returns 'legacy'."""
-        collector = create_metrics_collector(self.settings)
+        collector = create_metrics_collector(self._make_settings())
         self.assertEqual(collector.collector_name(), "legacy")
 
     def test_canonical_collector_name(self):
         """CanonicalMetricsCollector.collector_name() returns 'canonical'."""
         os.environ["WORKER_CANONICAL_METRICS"] = "true"
-        collector = create_metrics_collector(
-            MetricsSettings(directory=tempfile.mkdtemp())
-        )
+        collector = create_metrics_collector(self._make_settings())
         self.assertEqual(collector.collector_name(), "canonical")
 
     def test_both_implementations_satisfy_same_interface(self):
         """Both implementations have the same public method surface."""
-        legacy = LegacyMetricsCollector(self.settings)
+        legacy = LegacyMetricsCollector(self._make_settings())
         os.environ["WORKER_CANONICAL_METRICS"] = "true"
-        canonical = create_metrics_collector(
-            MetricsSettings(directory=tempfile.mkdtemp())
-        )
+        canonical = create_metrics_collector(self._make_settings())
 
         required_methods = [
             "collector_name",