Fix for spawn-mode subprocess metrics directory db file wiping (#417)

Author: chrishagglund-ship-it

.github/workflows/harness-image.yml

@@ -29,6 +29,8 @@ jobs:
     permissions:
       contents: read
       packages: write
+    outputs:
+      image-tag: ${{ steps.vars.outputs.image-tag }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -53,6 +55,10 @@ jobs:
           BRANCH="${{ github.ref_name }}"
           CLEANED_BRANCH_NAME=$(echo "$BRANCH" | tr '/' '-' | tr '[:upper:]' '[:lower:]')
           echo "cleaned-branch-name=$CLEANED_BRANCH_NAME" >> "$GITHUB_OUTPUT"
+          # The tag the downstream deploy should pull. dispatch-deploy only runs
+          # on workflow_dispatch, so this always matches the branch image pushed
+          # below (<branch>-latest), never the unrelated release tag.
+          echo "image-tag=${CLEANED_BRANCH_NAME}-latest" >> "$GITHUB_OUTPUT"
 
       - name: Docker metadata
         id: meta
@@ -92,4 +98,4 @@ jobs:
           repository: conductor-oss/oss-ci-util
           event-type: sdk_release
           client-payload: |-
-            {"tag": "${{ github.event.release.tag_name || 'latest' }}", "repo": "${{ github.repository }}"}
+            {"tag": "${{ needs.build-and-push.outputs.image-tag }}", "repo": "${{ github.repository }}"}

METRICS.md

@@ -68,6 +68,27 @@ or `clean_dead_pids=True` to remove only files from PIDs that no longer exist.
 Both default to `False`. Use a dedicated metrics directory per worker process
 group.
 
+Cleanup is owned by the parent process and performed by
+`MetricsSettings.clean_metrics_directory()`, which is idempotent per process
+(the destructive step runs at most once per directory). Spawned workers only
+ensure the directory exists (via `create_metrics_collector`) and never delete
+`.db` files, so a newly started or restarted worker can never wipe metrics
+belonging to live sibling processes.
+
+`TaskHandler.__init__` calls `clean_metrics_directory()` for you before any
+worker is spawned, which is sufficient for worker-only apps. If the parent
+process *also* collects metrics before constructing the `TaskHandler` (for
+example, registering task/workflow definitions or starting workflows through an
+instrumented client), build the parent's collector with
+`create_metrics_collector_for_parent(metrics_settings)` instead of
+`create_metrics_collector`. The parent factory cleans the directory up front --
+before the collector's first write -- and then creates the collector. Because
+`clean_metrics_directory()` is idempotent, `TaskHandler`'s later call is then a
+no-op and cannot orphan the parent's own `.db` files. For a long-lived parent
+that shares the directory, prefer `clean_dead_pids=True` over
+`clean_directory=True` -- it never deletes a live process's file regardless of
+ordering.
+
 ## Selecting Canonical Metrics
 
 Set `WORKER_CANONICAL_METRICS` before the worker starts:
@@ -290,7 +311,13 @@ sum(rate(task_execute_time_seconds_count[5m])) by (taskType)
   implementations never mixes stale metric names.
 - Pass `clean_dead_pids=True` to `MetricsSettings` to remove `.db` files from
   PIDs that no longer exist.  Use `clean_directory=True` only when you are sure
-  no other live process shares the same directory.
+  no other live process shares the same directory. Cleanup is idempotent per
+  process and runs in the parent (via `clean_metrics_directory()`, which
+  `TaskHandler.__init__` calls) before workers spawn; workers never wipe the
+  directory, so restarts and newly spawned workers preserve sibling metrics. If
+  the parent also collects, build its collector with
+  `create_metrics_collector_for_parent()` (cleans up front, then creates) so it
+  doesn't orphan the parent's own files.
 - Restart workers after changing `WORKER_CANONICAL_METRICS`.
 
 ### High Cardinality
@@ -342,8 +369,15 @@ unreleased metrics harmonization work. For a summary, see the project
     across all processes (main, workers, MetricsProvider).
   - `MetricsSettings` gains `clean_directory` (default `False`) to wipe all
     `.db` files and `clean_dead_pids` (default `False`) to remove only `.db`
-    files from PIDs that no longer exist. Both are executed by the factory
-    against the resolved metrics directory.
+    files from PIDs that no longer exist. Cleanup is applied by
+    `MetricsSettings.clean_metrics_directory()`, which is idempotent per process
+    and invoked by the parent (`TaskHandler.__init__` calls it, or a
+    parent that also collects builds its collector via
+    `create_metrics_collector_for_parent()`, which cleans up front then creates)
+    before workers spawn.
+    `create_metrics_collector` (the per-worker path) is non-destructive and
+    only ensures the directory exists, so spawned/restarted workers never wipe
+    live sibling metrics.
   - `CONDUCTOR_MP_START_METHOD` env var (`spawn` / `fork` / `forkserver`;
     default `fork` on POSIX, `spawn` on Windows) to control the worker pool's
     multiprocessing start method (motivated by a `prometheus_client` lock-fork

harness/main.py

@@ -9,7 +9,7 @@
 from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from conductor.client.http.models.task_def import TaskDef
 from conductor.client.orkes_clients import OrkesClients
-from conductor.client.telemetry.metrics_factory import create_metrics_collector
+from conductor.client.telemetry.metrics_factory import create_metrics_collector_for_parent
 from conductor.client.workflow.conductor_workflow import ConductorWorkflow
 from conductor.client.workflow.task.simple_task import SimpleTask
 
@@ -77,7 +77,12 @@ def main() -> None:
         clean_dead_pids=True,
     )
 
-    metrics_collector = create_metrics_collector(metrics_settings)
+    # The parent collects metrics too (task/workflow registration below and the
+    # WorkflowGovernor), so use the parent factory: it cleans the directory up
+    # front -- before the collector's first write -- then creates the collector,
+    # avoiding orphaning the parent's own .db files. clean_metrics_directory() is
+    # idempotent per process, so TaskHandler's later call is a no-op.
+    metrics_collector = create_metrics_collector_for_parent(metrics_settings)
     print(f"Prometheus metrics server started on port {metrics_port} ({metrics_collector.collector_name()} metrics)")
     clients = OrkesClients(configuration, metrics_collector=metrics_collector)
 

src/conductor/client/automator/task_handler.py

@@ -251,6 +251,14 @@ def __init__(
         if metrics_settings is not None:
             os.environ["PROMETHEUS_MULTIPROC_DIR"] = metrics_settings.metrics_directory
             logger.info(f"Set PROMETHEUS_MULTIPROC_DIR={metrics_settings.metrics_directory}")
+            # Clean the shared metrics directory in the parent, before any
+            # worker process is spawned. This is a convenience for worker-only
+            # apps; clean_metrics_directory() is idempotent per process, so if
+            # the entrypoint already cleaned up front (recommended when the
+            # parent itself collects metrics), this call is a no-op and will not
+            # wipe .db files the parent already began writing. Spawned workers
+            # call create_metrics_collector (non-destructive) and never clean.
+            metrics_settings.clean_metrics_directory()
 
         # Store event listeners to pass to each worker process
         self.event_listeners = event_listeners or []

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

@@ -17,6 +17,20 @@
 CANONICAL_SUBDIR = "canonical"
 
 
+# Directories already cleaned in THIS process. Cleanup must run at most once
+# per directory per process so that a second caller (e.g. TaskHandler after the
+# owning entrypoint already cleaned up front) cannot wipe .db files that a live
+# process started writing after the first clean. Safe under spawn because only
+# the parent ever cleans; spawned workers never call clean_metrics_directory.
+_cleaned_metrics_directories: set = set()
+
+
+def _reset_cleaned_metrics_directories() -> None:
+    """Clear the per-process cleanup guard. Intended for tests so each starts
+    from a known state and does not leak entries via this module-level global."""
+    _cleaned_metrics_directories.clear()
+
+
 def _env_bool(name: str, default: bool) -> bool:
     value = os.environ.get(name, "")
     if not value:
@@ -100,6 +114,39 @@ def metrics_directory(self) -> str:
             return os.path.join(self.directory, self._subdir)
         return self.directory
 
+    def clean_metrics_directory(self) -> None:
+        """Prepare the shared metrics directory, at most once per process.
+
+        This is the destructive counterpart to metrics collection and must be
+        invoked only by the process that owns the metrics lifecycle, and as
+        early as possible -- before that process (or any worker it spawns)
+        creates a collector and starts writing ``.db`` files. It must never be
+        called by a spawned worker: a worker cannot know whether sibling
+        processes are already live and sharing this directory, so it must never
+        wipe ``.db`` files.
+
+        Idempotent per process: the destructive cleanup runs only on the first
+        call for a given directory. Later calls (e.g. ``TaskHandler`` invoking
+        this after the entrypoint already cleaned up front) only ensure the
+        directory exists, so they cannot wipe files a live process began
+        writing after the first clean.
+
+        On the first call, ensures the directory exists and applies the
+        configured cleanup:
+          - ``clean_directory``: remove all prometheus_client ``.db`` files.
+          - ``clean_dead_pids``: remove only ``.db`` files whose owning PID no
+            longer exists.
+        Both are no-ops when their respective flag is ``False``.
+        """
+        os.makedirs(self.metrics_directory, exist_ok=True)
+        if self.metrics_directory in _cleaned_metrics_directories:
+            return
+        _cleaned_metrics_directories.add(self.metrics_directory)
+        if self.clean_directory:
+            self._clean_stale_db_files()
+        if self.clean_dead_pids:
+            self._clean_dead_pid_files()
+
     def __set_dir(self, dir: str) -> None:
         if not os.path.isdir(dir):
             try:

src/conductor/client/telemetry/metrics_factory.py

@@ -22,16 +22,6 @@
 )
 
 
-_cleaned_directories: set = set()
-
-
-def _reset_cleaned_directories() -> None:
-    """Clear the per-process cleanup guard.  Intended for tests so that each
-    test starts from a known-clean state and does not inherit or leak entries
-    via this module-level global."""
-    _cleaned_directories.clear()
-
-
 def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
     """
     Create the metrics collector indicated by ``settings.is_canonical``.
@@ -41,17 +31,17 @@ def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
 
     Returns a fully-initialised collector (legacy or canonical) that satisfies
     the MetricsCollector Protocol and can be registered as an event listener.
+
+    This is non-destructive: it only ensures the directory exists.  It never
+    deletes ``.db`` files, because it runs in every spawned worker and a worker
+    must not wipe metrics belonging to live sibling processes.  Directory
+    cleanup is owned by the parent via
+    ``MetricsSettings.clean_metrics_directory()`` (invoked once by
+    ``TaskHandler`` before workers spawn).
     """
     metrics_dir = settings.metrics_directory
     os.makedirs(metrics_dir, exist_ok=True)
 
-    if metrics_dir not in _cleaned_directories:
-        _cleaned_directories.add(metrics_dir)
-        if settings.clean_directory:
-            settings._clean_stale_db_files()
-        if settings.clean_dead_pids:
-            settings._clean_dead_pid_files()
-
     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)
@@ -60,3 +50,22 @@ def create_metrics_collector(settings: MetricsSettings) -> MetricsCollectorBase:
     from conductor.client.telemetry.legacy_metrics_collector import LegacyMetricsCollector
     logger.info("Using LegacyMetricsCollector (dir=%s; set WORKER_CANONICAL_METRICS=true for canonical)", metrics_dir)
     return LegacyMetricsCollector(settings)
+
+
+def create_metrics_collector_for_parent(settings: MetricsSettings) -> MetricsCollectorBase:
+    """Owner/parent entrypoint: prepare the shared metrics directory, then build
+    the collector.
+
+    Use this from the process that owns the metrics lifecycle (the one that
+    spawns workers, and that may itself collect metrics before spawning). It
+    cleans the directory up front -- honoring ``settings.clean_directory`` /
+    ``settings.clean_dead_pids`` -- so the parent's own first write is not
+    orphaned, then delegates to ``create_metrics_collector``.
+
+    Do NOT call this from a spawned worker: workers must use
+    ``create_metrics_collector`` (non-destructive) so they can never wipe ``.db``
+    files belonging to live sibling processes. ``clean_metrics_directory()`` is
+    idempotent per process, so a later ``TaskHandler`` call is a no-op.
+    """
+    settings.clean_metrics_directory()
+    return create_metrics_collector(settings)

tests/unit/automator/test_task_handler.py

@@ -1,11 +1,13 @@
 import multiprocessing
+import tempfile
 import unittest
 from unittest.mock import Mock
 from unittest.mock import patch
 
 from conductor.client.automator.task_handler import TaskHandler
 from conductor.client.automator.task_runner import TaskRunner
 from conductor.client.configuration.configuration import Configuration
+from conductor.client.configuration.settings.metrics_settings import MetricsSettings
 from tests.unit.resources.workers import ClassWorker
 
 
@@ -41,6 +43,23 @@ def test_start_processes(self):
                         isinstance(process, multiprocessing.Process)
                     )
 
+    def test_metrics_directory_cleaned_once_in_parent_init(self):
+        """TaskHandler.__init__ (parent, pre-spawn) invokes
+        clean_metrics_directory exactly once. Spawned workers rely on the
+        non-destructive create_metrics_collector path and never clean, so this
+        parent-owned call is the single point where .db files are wiped."""
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            metrics_settings = MetricsSettings(directory=tmp_dir, clean_directory=True)
+            metrics_settings.clean_metrics_directory = Mock()
+
+            task_handler = TaskHandler(
+                configuration=Configuration(),
+                workers=[ClassWorker('task')],
+                metrics_settings=metrics_settings,
+            )
+            with task_handler:
+                metrics_settings.clean_metrics_directory.assert_called_once_with()
+
 
 def _get_valid_task_handler():
     return TaskHandler(