Skip to content

METRICS.md

Latest commit

 

History

History
403 lines (331 loc) · 21.4 KB

METRICS.md

File metadata and controls

403 lines (331 loc) · 21.4 KB
Raw

Python SDK Metrics

The Conductor Python SDK can expose Prometheus metrics for worker polling, task execution, task updates, workflow starts, external payload usage, and generated API-client HTTP calls.

The SDK currently has two mutually exclusive metric surfaces:

  • Legacy metrics are the default. They preserve the pre-harmonization Python SDK names and shapes, including sliding-window quantile gauges for timing metrics.
  • Canonical metrics are opt-in with WORKER_CANONICAL_METRICS=true. They use the cross-SDK canonical names, labels, units, and Prometheus histogram shapes.

Only one collector is active in a worker process. The SDK does not emit legacy and canonical metrics at the same time.

Metric names below are the names exposed in Prometheus text output. The Python prometheus_client library appends _total to counters in exposition output.

Configuration

Enable metrics by passing MetricsSettings to TaskHandler.

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration
from conductor.client.configuration.settings.metrics_settings import MetricsSettings

config = Configuration()
metrics = MetricsSettings(
    directory="/tmp/conductor-metrics",
    http_port=8000,
)

with TaskHandler(
    configuration=config,
    metrics_settings=metrics,
    scan_for_annotated_workers=True,
) as task_handler:
    task_handler.start_processes()
    task_handler.join_processes()

With http_port set, the SDK starts a Prometheus-compatible HTTP endpoint:

curl http://localhost:8000/metrics
curl http://localhost:8000/health

Without http_port, the SDK writes Prometheus text output to {directory}/{file_name} at update_interval seconds:

metrics = MetricsSettings(
    directory="/tmp/conductor-metrics",
    file_name="conductor_metrics.prom",
    update_interval=10,
    http_port=None,
)

MetricsSettings provides opt-in cleanup of stale Prometheus multiprocess .db files. Pass clean_directory=True to remove all .db files on startup, 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:

WORKER_CANONICAL_METRICS=true python my_worker.py

Accepted true values are true, 1, and yes, case-insensitive. Any other value, or an unset variable, selects legacy metrics. The variable is read when the metrics collector is created, so changing it requires a worker restart.

WORKER_LEGACY_METRICS is reserved for future use. After the deprecation period ends and canonical becomes the default, setting WORKER_LEGACY_METRICS=true will allow opting back into legacy metrics. It is not currently read.

Canonical Metrics

Canonical timing values are seconds. Canonical size values are bytes. Label names use camelCase.

Metrics are created lazily. A metric appears in /metrics only after the corresponding worker event or collector method records it. Some low-level surface metrics, such as ack, queue-full, paused, and uncaught-exception counters, may not appear in normal worker runs unless that path is exercised.

Canonical Counters

Metric Labels Description
task_poll_total taskType Incremented each time the worker issues a poll request.
task_execution_started_total taskType Incremented when a polled task is dispatched to the worker function.
task_poll_error_total taskType, exception Incremented when a poll request fails client-side.
task_execute_error_total taskType, exception Incremented when the worker function throws.
task_update_error_total taskType, exception Incremented when updating the task result fails.
task_ack_error_total taskType, exception Collector surface for task ack errors. Not yet wired -- no runtime call site.
task_ack_failed_total taskType Collector surface for failed task ack responses. Not yet wired -- no runtime call site.
task_execution_queue_full_total taskType Collector surface for execution queue saturation events. Not yet wired -- no runtime call site.
task_paused_total taskType Collector surface for polls skipped while the worker is paused. Not yet wired -- no runtime call site.
thread_uncaught_exceptions_total exception Collector surface for uncaught worker-thread exceptions. Not yet wired -- no runtime call site.
worker_restart_total taskType Python-only counter for TaskHandler subprocess restarts.
external_payload_used_total entityName, operation, payloadType Incremented when external payload storage is used.
workflow_start_error_total workflowType, exception Incremented when starting a workflow fails client-side.

Canonical Time Histograms

All canonical time histograms use buckets: 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10.

Metric Labels Description
task_poll_time_seconds taskType, status Poll request latency. status is SUCCESS or FAILURE.
task_execute_time_seconds taskType, status Worker function execution duration. status is SUCCESS or FAILURE.
task_update_time_seconds taskType, status Task-result update latency. status is SUCCESS or FAILURE.
http_api_client_request_seconds method, uri, status Generated API-client HTTP request latency.

Each histogram exposes Prometheus series such as:

task_execute_time_seconds_bucket{taskType="my_task",status="SUCCESS",le="0.1"} 42.0
task_execute_time_seconds_count{taskType="my_task",status="SUCCESS"} 50.0
task_execute_time_seconds_sum{taskType="my_task",status="SUCCESS"} 2.3

Canonical Size Histograms

All canonical size histograms use buckets: 100, 1000, 10000, 100000, 1000000, 10000000.

Metric Labels Description
task_result_size_bytes taskType Serialized task result output size.
workflow_input_size_bytes workflowType, version Serialized workflow input size.

Canonical Gauges

Metric Labels Description
active_workers taskType Current number of workers actively executing a task.

Legacy Metrics

Legacy mode is the default so existing dashboards and alerts continue to work. Timing metrics are sliding-window quantile gauges over the latest 1,000 observations. Legacy timing metrics also expose _count and _sum gauge series for the current sliding window.

As in canonical mode, metrics are created lazily and rare or surface-only counters appear only when the corresponding code path records them.

Legacy Counters

Metric Labels Description
task_poll_total taskType Incremented each time polling is done.
task_execute_error_total taskType, exception Task execution errors. exception is str(exception).
task_update_error_total taskType, exception Task update errors. exception is str(exception).
task_ack_error_total taskType, exception Collector surface for task ack errors. Not yet wired -- no runtime call site. exception is str(exception).
task_ack_failed_total taskType Collector surface for failed task ack responses. Not yet wired -- no runtime call site.
task_execution_queue_full_total taskType Collector surface for execution queue saturation events. Not yet wired -- no runtime call site.
task_paused_total taskType Collector surface for polls skipped while the worker is paused. Not yet wired -- no runtime call site.
thread_uncaught_exceptions_total none Collector surface for uncaught worker-thread exceptions. Not yet wired -- no runtime call site.
worker_restart_total taskType TaskHandler subprocess restarts.
external_payload_used_total entityName, operation, payload_type External payload storage usage.
workflow_start_error_total workflowType, exception Workflow start errors. exception is str(exception).

Legacy mode does not emit task_poll_error_total, task_execution_started_total, or active_workers.

Legacy Time Metrics

Metric Type Labels Description
task_poll_time Gauge taskType Most recent poll duration, in seconds.
task_poll_time_seconds Quantile gauge taskType, status, quantile Sliding-window poll latency quantiles.
task_poll_time_seconds_count Gauge taskType, status Sliding-window poll observation count.
task_poll_time_seconds_sum Gauge taskType, status Sliding-window poll duration sum.
task_execute_time Gauge taskType Most recent task execution duration, in seconds.
task_execute_time_seconds Quantile gauge taskType, status, quantile Sliding-window execution latency quantiles.
task_execute_time_seconds_count Gauge taskType, status Sliding-window execution observation count.
task_execute_time_seconds_sum Gauge taskType, status Sliding-window execution duration sum.
task_update_time_seconds Quantile gauge taskType, status, quantile Sliding-window task update latency quantiles. Never emitted -- no call sites existed on main. Legacy no-ops this metric; only canonical mode records task update time.
task_update_time_seconds_count Gauge taskType, status Sliding-window task update observation count. Never emitted (see above).
task_update_time_seconds_sum Gauge taskType, status Sliding-window task update duration sum. Never emitted (see above).
http_api_client_request Quantile gauge method, uri, status, quantile Sliding-window API-client request latency quantiles.
http_api_client_request_count Gauge method, uri, status Sliding-window API-client request observation count.
http_api_client_request_sum Gauge method, uri, status Sliding-window API-client request duration sum.

Legacy Size Gauges

Metric Labels Description
task_result_size taskType Most recent serialized task result output size, in bytes.
workflow_input_size workflowType, version Most recent serialized workflow input size, in bytes.

Labels

Label Used by Values
taskType Worker metrics Task definition name.
workflowType Workflow metrics Workflow definition name.
version workflow_input_size, workflow_input_size_bytes Workflow version as a string. If absent, the label is an empty string.
status Task time metrics SUCCESS or FAILURE. For HTTP metrics, the response code as a string, an exception status or code, or error when unavailable.
exception Error counters Legacy uses str(exception). Canonical uses the exception class name, such as TimeoutError.
entityName external_payload_used_total Task type or workflow name associated with the external payload.
operation external_payload_used_total External payload operation, such as READ or WRITE.
payload_type Legacy external_payload_used_total Payload type, such as TASK_INPUT, TASK_OUTPUT, WORKFLOW_INPUT, or WORKFLOW_OUTPUT.
payloadType Canonical external_payload_used_total Payload type, such as TASK_INPUT, TASK_OUTPUT, WORKFLOW_INPUT, or WORKFLOW_OUTPUT.
method HTTP metrics HTTP verb.
uri HTTP metrics Request path passed by the generated API client.
quantile Legacy time metrics 0.5, 0.75, 0.9, 0.95, or 0.99.

Migrating From Legacy to Canonical

Canonical mode is opt-in during the deprecation period. Before switching a production worker, update dashboards and alerts against a staging worker with WORKER_CANONICAL_METRICS=true.

Key changes:

  • Time and size distribution metrics are real Prometheus histograms in canonical mode. Query _bucket series with histogram_quantile() instead of reading {quantile="..."} gauges.
  • Legacy last-value gauges task_poll_time, task_execute_time, task_result_size, and workflow_input_size are not emitted in canonical mode.
  • Canonical adds task_poll_error_total, task_execution_started_total, and active_workers.
  • external_payload_used_total changes label payload_type to payloadType.
  • Canonical exception labels are bounded to exception class names. Legacy error counters may include raw exception messages.

Legacy metrics that change name or type in canonical mode:

Legacy metric Canonical metric Change
task_poll_time (gauge) — Removed; use the histogram instead.
task_execute_time (gauge) — Removed; use the histogram instead.
task_result_size (gauge) task_result_size_bytes (histogram) Renamed; gauge becomes histogram with buckets.
workflow_input_size (gauge) workflow_input_size_bytes (histogram) Renamed; gauge becomes histogram with buckets.
http_api_client_request (quantile gauge) http_api_client_request_seconds (histogram) Renamed with _seconds suffix; quantile gauge becomes histogram.
external_payload_used_total{payload_type=…} external_payload_used_total{payloadType=…} Label renamed from payload_type to payloadType.

Common PromQL replacements:

Legacy Canonical
task_poll_time_seconds{quantile="0.95"} histogram_quantile(0.95, sum by (le, taskType, status) (rate(task_poll_time_seconds_bucket[5m])))
task_execute_time_seconds{quantile="0.95"} histogram_quantile(0.95, sum by (le, taskType, status) (rate(task_execute_time_seconds_bucket[5m])))
task_update_time_seconds{quantile="0.95"} (never emitted in legacy) histogram_quantile(0.95, sum by (le, taskType, status) (rate(task_update_time_seconds_bucket[5m])))
http_api_client_request{quantile="0.95"} histogram_quantile(0.95, sum by (le, method, uri, status) (rate(http_api_client_request_seconds_bucket[5m])))
task_result_size task_result_size_bytes_bucket, _count, and _sum
workflow_input_size workflow_input_size_bytes_bucket, _count, and _sum
external_payload_used_total{payload_type="TASK_OUTPUT"} external_payload_used_total{payloadType="TASK_OUTPUT"}

Average latency queries continue to use _sum divided by _count, but the canonical series are cumulative histogram counters:

sum(rate(task_execute_time_seconds_sum[5m])) by (taskType)
/
sum(rate(task_execute_time_seconds_count[5m])) by (taskType)

Troubleshooting

Metrics Are Empty

  • Verify that metrics_settings is passed to TaskHandler.
  • Verify workers have polled or executed tasks. Metrics are created lazily when the relevant event occurs.
  • Check that the metrics directory is writable.

Stale or Unexpected Series

  • Legacy metrics use the base directory unchanged from prior releases. Canonical metrics use a canonical/ subdirectory, so switching 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. 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

  • Prefer canonical mode for bounded exception labels.
  • Watch the uri label on HTTP metrics. The Python SDK records the request path available at the generated API-client call site.
  • Avoid embedding user identifiers or unbounded values in task type, workflow type, or external payload labels.

Detailed Technical Notes -- Unreleased

Implementation details, internal design decisions, and migration notes for the unreleased metrics harmonization work. For a summary, see the project CHANGELOG.

Added

  • Metrics harmonization -- canonical metric surface aligned with the cross-SDK catalog, opt-in via WORKER_CANONICAL_METRICS=true
    • New CanonicalMetricsCollector emits the harmonized cross-SDK catalog using real Prometheus Histograms for timing and size, replacing the legacy quantile-gauge timing shape. New canonical-only metrics: task_poll_error_total, task_execution_started_total, task_result_size_bytes, workflow_input_size_bytes, http_api_client_request_seconds, active_workers. Time buckets 0.001…10s; size buckets 100…10_000_000 bytes.
    • metrics_factory.create_metrics_collector(settings) selects LegacyMetricsCollector (default) or CanonicalMetricsCollector based on WORKER_CANONICAL_METRICS (truthy: true, 1, yes, case-insensitive, whitespace-trimmed). WORKER_LEGACY_METRICS is reserved for future use and is not currently read.
    • New abstract MetricsCollectorBase consolidates Prometheus infrastructure (lazy prometheus_client imports, multiprocess NoPidCollector aggregation, HTTP server, exception-label cardinality bounding) and event handlers shared by both collectors.
    • TaskRunner and AsyncTaskRunner now record task_update_time (status="SUCCESS" / "FAILURE") on every update path.
    • OrkesWorkflowClient.start_workflow* calls measure_workflow_input_payload_size and measure_workflow_start_error on the collector; canonical records workflow input size and start errors, legacy no-ops (preserving existing behavior). OrkesClients / OrkesBaseClient accept an optional metrics_collector.
    • Legacy metrics use the base directory unchanged from prior releases; canonical metrics use a canonical/ subdirectory so that switching implementations never produces stale metric names. MetricsSettings.metrics_directory is a computed property that derives the correct path from WORKER_CANONICAL_METRICS, ensuring consistent paths 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. 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.
    • Harness manifest sets WORKER_CANONICAL_METRICS=true; harness/main.py logs which collector is active.

Changed

  • Metrics harmonization -- defaults preserved; legacy metrics emit unchanged when WORKER_CANONICAL_METRICS is unset
    • MetricLabel.PAYLOAD_TYPE retains its original value "payload_type"; a new PAYLOAD_TYPE_CAMEL = "payloadType" constant is used only by the canonical collector on external_payload_used_total.
    • metrics_collector.py is now a thin compatibility shim: MetricsCollector = LegacyMetricsCollector, so from conductor.client.telemetry.metrics_collector import MetricsCollector continues to work.
    • Default behavior is unchanged: with no env var set, the legacy metric names, label conventions, and quantile-gauge timing shape from prior releases are preserved.
    • Rewrote METRICS.md to document both surfaces, the env-var gate, full canonical and legacy catalogs, labels, a "Migrating From Legacy to Canonical" mapping (including the payload_type → payloadType label change and PromQL replacements), and troubleshooting.
    • Updated README.md, WORKER_CONFIGURATION.md, and docs/design/WORKER_DESIGN.md to point at METRICS.md.