diff --git a/.rubocop.yml b/.rubocop.yml index 98804e9..a384d8f 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -21,6 +21,7 @@ Style/MixinUsage: # Metrics settings Metrics/BlockLength: + Max: 40 Exclude: - 'spec/**/*' - '*.gemspec' diff --git a/AGENTS.md b/AGENTS.md index d06e457..dbff989 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -300,46 +300,17 @@ end ### Metrics Collection -The MetricsCollector listens to events and tracks metrics: +The SDK supports legacy and canonical metric surfaces, selected by the +`WORKER_CANONICAL_METRICS` environment variable. `MetricsCollector.create` +returns the appropriate collector: ```ruby -class MetricsCollector - def on_poll_started(event) - increment("task_poll_total", task_type: event.task_type) - end - - def on_poll_completed(event) - observe("task_poll_time_seconds", event.duration_ms / 1000.0, - task_type: event.task_type) - end - - def on_task_execution_completed(event) - observe("task_execute_time_seconds", event.duration_ms / 1000.0, - task_type: event.task_type) - observe("task_result_size_bytes", event.output_size_bytes, - task_type: event.task_type) - end - - def on_task_execution_failure(event) - increment("task_execute_error_total", - task_type: event.task_type, - exception: event.cause.class.name, - retryable: event.is_retryable.to_s) - end -end +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) ``` -### Prometheus Metrics - -| Metric | Type | Labels | Description | -|--------|------|--------|-------------| -| `task_poll_total` | Counter | `task_type` | Number of poll operations | -| `task_poll_time_seconds` | Histogram | `task_type` | Poll latency | -| `task_poll_error_total` | Counter | `task_type`, `error` | Poll failures | -| `task_execute_time_seconds` | Histogram | `task_type` | Execution time | -| `task_execute_error_total` | Counter | `task_type`, `exception`, `retryable` | Execution failures | -| `task_result_size_bytes` | Histogram | `task_type` | Output size | -| `task_update_failed_total` | Counter | `task_type` | CRITICAL: Update failures | +See [docs/METRICS_AND_INTERCEPTORS.md](docs/METRICS_AND_INTERCEPTORS.md) for +the full legacy and canonical metrics catalogs, label reference, and migration +guide. ### Worker Configuration (3-Tier Hierarchy) @@ -436,8 +407,11 @@ lib/conductor/ │ │ ├── listeners.rb # Listener protocol │ │ └── listener_registry.rb # Registration helper │ └── telemetry/ # Metrics -│ ├── metrics_collector.rb # Event-based metrics -│ └── prometheus_backend.rb # Prometheus integration +│ ├── metrics_collector.rb # Factory (WORKER_CANONICAL_METRICS gate) +│ ├── legacy_metrics_collector.rb # Legacy metric set +│ ├── canonical_metrics_collector.rb # Canonical metric set +│ ├── prometheus_backend.rb # Legacy Prometheus backend +│ └── canonical_prometheus_backend.rb # Canonical Prometheus backend └── workflow/ ├── dsl/ # Workflow DSL │ ├── workflow_builder.rb # Core DSL engine (~1000 lines) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ebc1c6..371e386 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Canonical metrics mode: opt-in harmonized metric surface via `WORKER_CANONICAL_METRICS=true` -- [details](docs/METRICS_AND_INTERCEPTORS.md#detailed-technical-notes----unreleased) +- Bounded `uri` label on `http_api_client_request_seconds`: uses path templates (e.g. `/workflow/{workflowId}`) instead of fully-resolved paths, preventing metric cardinality explosion +- `WorkflowStatusProbe` in harness: opt-in probe (via `HARNESS_PROBE_RATE_PER_SEC`) that exercises UUID-bearing endpoints to validate template URI metrics + ### Changed - **BREAKING: Workflow DSL Redesign** - Complete redesign of the workflow DSL for Ruby-idiomatic syntax @@ -17,6 +23,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Control flow blocks: `parallel do`, `decide expr do`, `loop_over items do` - Auto-generated task reference names - Simplified LLM task methods with hash-to-ChatMessage auto-conversion +- `MetricsCollector.new(...)` is deprecated; use `MetricsCollector.create(...)` instead. `.new` still works but logs a deprecation warning. The previous implementation is preserved as `LegacyMetricsCollector` and remains the default. +- Legacy metrics emit unchanged by default; existing dashboards and alerts continue to work without modification +- HTTP request timing (`http_api_client_request_seconds`) is now zero-overhead in legacy mode: `RestClient` only enters the timing path when a canonical collector is subscribed to `GlobalDispatcher` +- `thread_uncaught_exceptions_total` is no longer incremented for caught exceptions in the polling loop; the metric surface is retained but unwired, matching the Python and JavaScript SDKs +- Both `CanonicalMetricsCollector` and `LegacyMetricsCollector` now respond to `stop`; `TaskHandler#stop` calls it automatically to unsubscribe from process-wide dispatchers +- `MetricsCollector.create` accepts `measure_payload_size:` (default `true` for canonical, `false` for legacy) to opt out of `workflow_input_size_bytes` JSON serialization overhead ### Removed diff --git a/Gemfile b/Gemfile index 32c9354..8e86acf 100644 --- a/Gemfile +++ b/Gemfile @@ -4,6 +4,9 @@ source 'https://rubygems.org' gemspec +install_if -> { RUBY_VERSION >= '3.3' } do + gem 'async', '~> 2.0', group: :test +end gem 'prometheus-client', '~> 4.0' gem 'rake', '~> 13.0' gem 'webrick', '~> 1.8' diff --git a/Gemfile.lock b/Gemfile.lock index 65cf156..97fbb26 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -14,11 +14,21 @@ GEM addressable (2.8.8) public_suffix (>= 2.0.2, < 8.0) ast (2.4.3) + async (2.39.0) + console (~> 1.29) + fiber-annotation + io-event (~> 1.11) + metrics (~> 0.12) + traces (~> 0.18) base64 (0.3.0) bigdecimal (4.0.1) coderay (1.1.3) concurrent-ruby (1.3.6) connection_pool (2.5.5) + console (1.35.1) + fiber-annotation + fiber-local (~> 1.1) + json crack (1.0.1) bigdecimal rexml @@ -33,10 +43,16 @@ GEM net-http-persistent (>= 4.0.4, < 5) faraday-retry (2.4.0) faraday (~> 2.0) + fiber-annotation (0.2.0) + fiber-local (1.1.0) + fiber-storage + fiber-storage (1.0.1) hashdiff (1.2.1) io-console (0.8.2) + io-event (1.16.0) json (2.7.6) method_source (1.1.0) + metrics (0.15.0) net-http-persistent (4.0.8) connection_pool (>= 2.2.4, < 4) parallel (1.24.0) @@ -89,6 +105,7 @@ GEM rubocop-capybara (~> 2.17) ruby-progressbar (1.13.0) ruby2_keywords (0.0.5) + traces (0.18.2) unicode-display_width (2.6.0) vcr (6.1.0) webmock (3.26.1) @@ -101,6 +118,7 @@ PLATFORMS ruby DEPENDENCIES + async (~> 2.0) conductor_ruby! prometheus-client (~> 4.0) pry (~> 0.14) diff --git a/docs/METRICS_AND_INTERCEPTORS.md b/docs/METRICS_AND_INTERCEPTORS.md index d20a9a6..ada3e8c 100644 --- a/docs/METRICS_AND_INTERCEPTORS.md +++ b/docs/METRICS_AND_INTERCEPTORS.md @@ -1,107 +1,483 @@ # Metrics and Interceptors Guide -This guide explains how to use the metrics and interceptor system in the Conductor Ruby SDK to monitor worker performance, track errors, and implement custom observability. +The Conductor Ruby SDK can expose Prometheus metrics for worker polling, task +execution, task result updates, payload sizes, workflow starts, and HTTP API +client latency. It also provides an event-driven interceptor system for custom +logging, error tracking, and observability. + +This document covers the Ruby SDK metrics emitted by `MetricsCollector.create`, +`LegacyMetricsCollector`, and `CanonicalMetricsCollector`. It does not cover +Conductor server metrics or metrics emitted by other SDKs. ## Table of Contents -- [Overview](#overview) +- [Legacy and Canonical Modes](#legacy-and-canonical-modes) - [Quick Start](#quick-start) +- [Canonical Metrics Catalog](#canonical-metrics-catalog) +- [Legacy Metrics Catalog](#legacy-metrics-catalog) +- [Metrics Not Applicable to Ruby](#metrics-not-applicable-to-ruby) +- [Labels](#labels) +- [Migration from Legacy to Canonical](#migration-from-legacy-to-canonical) +- [Prometheus Integration](#prometheus-integration) +- [Custom Metrics Backends](#custom-metrics-backends) +- [Troubleshooting](#troubleshooting) +- [Interceptor System](#interceptor-system) - [Event Types](#event-types) - [Creating Custom Interceptors](#creating-custom-interceptors) -- [Metrics Collection](#metrics-collection) -- [Prometheus Integration](#prometheus-integration) - [Advanced Use Cases](#advanced-use-cases) - [Best Practices](#best-practices) +- [Reference](#reference) --- -## Overview +## Legacy and Canonical Modes -The Conductor Ruby SDK provides an event-driven interceptor system that allows you to: +The Ruby SDK currently supports two mutually exclusive metric surfaces: -- **Monitor performance** - Track polling times, execution durations, error rates -- **Implement custom logging** - Add structured logging for task execution -- **Track errors** - Send failures to error tracking services (Sentry, Bugsnag, etc.) -- **Collect metrics** - Export to Prometheus, Datadog, or custom backends -- **Build alerting** - Monitor SLAs and trigger alerts on violations +- **Legacy metrics** are the default. They preserve the original Ruby SDK names + and labels, including snake_case label keys like `task_type`. +- **Canonical metrics** are opt-in with `WORKER_CANONICAL_METRICS=true`. They + use the cross-SDK canonical names, labels, units, and Prometheus histogram + bucket boundaries. -### How It Works +`MetricsCollector.create` reads `WORKER_CANONICAL_METRICS` when the collector +is created: -``` -TaskRunner - │ - │ publishes events - ▼ -SyncEventDispatcher ──────► Listener 1 (MetricsCollector) - ──────► Listener 2 (LoggingInterceptor) - ──────► Listener 3 (SentryInterceptor) +| Environment variable | Values | Effect | +|---|---|---| +| `WORKER_CANONICAL_METRICS` | `true`, `1`, or `yes` (case-insensitive, surrounding whitespace ignored) | Selects `CanonicalMetricsCollector`. | +| `WORKER_CANONICAL_METRICS` | unset, blank, `false`, `0`, `no`, or any other value | Selects `LegacyMetricsCollector`. | + +Only one implementation is active at a time. The SDK does not emit legacy and +canonical metrics simultaneously. Restart workers after changing +`WORKER_CANONICAL_METRICS` so the factory creates the desired collector. + +`WORKER_LEGACY_METRICS` is reserved for a future default-flip phase and is not +currently read by the Ruby SDK factory. + +### Payload Size Metrics + +Recording `workflow_input_size_bytes` requires JSON-serializing the workflow +input to measure its byte size. This is enabled by default in canonical mode +and disabled in legacy mode. Override explicitly via the factory: + +```ruby +# Canonical mode, but skip the JSON serialization for large payloads +metrics = MetricsCollector.create(backend: :prometheus, measure_payload_size: false) ``` -When a worker polls for tasks, executes them, or encounters errors, events are published to all registered listeners. Listeners can then process these events independently. +### Collector Lifecycle + +Both `CanonicalMetricsCollector` and `LegacyMetricsCollector` respond to +`stop`. Call `stop` to unsubscribe from process-wide dispatchers (e.g. the +`GlobalDispatcher` used for HTTP metrics). `TaskHandler#stop` calls `stop` +on all registered event listeners automatically. If you manage a collector +outside of `TaskHandler`, call `stop` when the collector is no longer needed. --- ## Quick Start -### Basic Logging Interceptor +### Enabling Metrics (Legacy, Default) ```ruby require 'conductor' -require 'logger' -# Create a simple logging interceptor -class LoggingInterceptor - def initialize - @logger = Logger.new($stdout) +# MetricsCollector.create checks WORKER_CANONICAL_METRICS and returns +# the appropriate collector. Default (unset) selects legacy metrics. +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) + +# Start metrics HTTP server +metrics_server = Conductor::Worker::Telemetry::MetricsServer.new(port: 9090) +metrics_server.start + +# Create handler with metrics +handler = Conductor::Worker::TaskHandler.new( + configuration: config, + event_listeners: [metrics] +) + +handler.start +handler.join + +# Cleanup +metrics_server.stop +``` + +### Enabling Canonical Metrics + +Set the environment variable before the worker starts: + +```shell +WORKER_CANONICAL_METRICS=true ruby my_worker.rb +``` + +The same code above will now return a `CanonicalMetricsCollector` instead. + +--- + +## Canonical Metrics Catalog + +Canonical timing values are seconds. Canonical size values are bytes. Label +names use camelCase. Metrics are created lazily and appear in `/metrics` only +after the corresponding event records them. + +### 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_paused_total` | `taskType` | Incremented when a worker is paused and skips acting on a poll. | +| `thread_uncaught_exceptions_total` | `exception` | Incremented when a worker thread raises an uncaught exception. | +| `workflow_start_error_total` | `workflowType`, `exception` | Incremented when starting a workflow fails client-side. | + +### Canonical Time Histograms + +All canonical time histograms use buckets (in seconds): + +```text +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` | HTTP API client request latency. `status` is the HTTP status code as a string, or `"0"` on network failure. | + +Each histogram exposes Prometheus series such as: + +```prometheus +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 (in bytes): + +```text +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. `version` is an empty string when the workflow version is absent. | + +### Canonical Gauges + +| Metric | Labels | Description | +|---|---|---| +| `active_workers` | `taskType` | Current number of worker threads actively executing tasks. | + +--- + +## Legacy Metrics Catalog + +Legacy mode is the default so existing dashboards and alerts continue to work. +Legacy labels use snake_case (`task_type`). Legacy histograms do not carry a +`status` label. Legacy poll failure does not record poll time -- only the error +counter is incremented. + +### Legacy Counters + +| Metric | Labels | Description | +|---|---|---| +| `task_poll_total` | `task_type` | Incremented each time polling is done. | +| `task_poll_error_total` | `task_type`, `error` | Poll failures. `error` is the exception class name. | +| `task_execute_error_total` | `task_type`, `exception`, `retryable` | Task execution errors. `retryable` is `true` or `false`. | +| `task_update_failed_total` | `task_type` | Failed task result updates (critical -- task result lost). | + +### Legacy Time Histograms + +Legacy time histograms use buckets (in seconds): + +```text +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` | `task_type` | Poll request latency. Recorded on successful polls only. | +| `task_execute_time_seconds` | `task_type` | Worker function execution duration. | + +### Legacy Size Histograms + +| Metric | Labels | Description | +|---|---|---| +| `task_result_size_bytes` | `task_type` | Serialized task result output size. Uses the same bucket set as canonical. | + +Legacy mode does not emit `task_execution_started_total`, +`task_update_time_seconds`, `task_paused_total`, +`thread_uncaught_exceptions_total`, `workflow_start_error_total`, +`http_api_client_request_seconds`, `workflow_input_size_bytes`, or +`active_workers`. + +--- + +## Metrics Not Applicable to Ruby + +The cross-SDK canonical catalog defines additional metrics that are not +applicable to the Ruby SDK's runtime model: + +| Canonical metric | Why N/A for Ruby | +|---|---| +| `task_ack_error_total` | Batch-poll response is the ack; there is no separate ack call. | +| `task_ack_failed_total` | Same reason. | +| `task_execution_queue_full_total` | Ruby uses `fallback_policy: :caller_runs` which back-pressures the polling thread instead of rejecting tasks. | +| `worker_restart_total` | Python-only. Its multi-process supervisor restarts child processes. Ruby uses threads, fibers, or ractors. | +| `external_payload_used_total` | Ruby SDK has no external-payload-storage integration. | + +Users cross-referencing the harmonization spec or documentation from other +Conductor SDKs may notice these metrics in other catalogs. Their absence in +the Ruby SDK is intentional. + +### `thread_uncaught_exceptions_total` + +The `thread_uncaught_exceptions_total` counter and its collector handler exist +in the Ruby SDK for API completeness, but the metric is not currently wired to +any runtime event. The harmonization spec defines it as "incremented when a +worker thread terminates with an uncaught exception." In SDKs that implement it +(Java, Go, Rust), it fires only at the thread/goroutine death boundary. Python +and JavaScript also define the metric surface but do not wire it. Future Ruby +SDK versions may connect it to `Thread.report_on_exception` or a similar +mechanism. + +### Ractor Runner Limitations + +The `RactorTaskRunner` does not currently emit `active_workers` gauge updates +because each Ractor processes tasks sequentially with no shared count. All +other canonical and legacy metrics are emitted by the Ractor runner. + +--- + +## Labels + +| Label | Used by | Values | +|---|---|---| +| `task_type` | Legacy worker metrics | Task definition name. Replaced by `taskType` in canonical mode. | +| `taskType` | Canonical worker metrics | Task definition name. | +| `workflowType` | Canonical workflow metrics | Workflow definition name. | +| `version` | `workflow_input_size_bytes` | Workflow version as a string. Empty string when the version is absent. | +| `status` | Canonical task time metrics | `SUCCESS` or `FAILURE`. For `http_api_client_request_seconds`, the HTTP status code as a string (e.g. `"200"`), or `"0"` on network failure. | +| `exception` | Canonical error counters | Exception class name, such as `Faraday::TimeoutError`. | +| `error` | Legacy `task_poll_error_total` | Exception class name. Renamed to `exception` in canonical mode. | +| `retryable` | Legacy `task_execute_error_total` | `true` or `false`. Dropped in canonical mode. | +| `method` | HTTP metrics | HTTP verb (`GET`, `POST`, etc.). | +| `uri` | HTTP metrics | API-relative path template (e.g. `/tasks/poll/batch/{taskType}`). Dynamic path segments retain `{placeholder}` tokens so label cardinality is bounded. | + +--- + +## Migration from Legacy to Canonical + +Switching to canonical metrics is an explicit metrics-surface cutover. Enable +`WORKER_CANONICAL_METRICS=true` in a lower environment first, then update +dashboards, recording rules, and alerts before enabling it in production. + +Key changes: + +- Legacy task labels use `task_type`; canonical task labels use `taskType`. +- Legacy poll failure only increments the error counter; canonical also records + poll time with `status=FAILURE`. +- Legacy execution errors carry an extra `retryable` label; canonical drops it. +- Legacy poll errors use the `error` label; canonical uses `exception`. +- Legacy `task_update_failed_total` becomes `task_update_error_total` with an + added `exception` label. +- Canonical time histogram buckets start at 0.001s; legacy starts at 0.005s. +- Canonical adds metrics that legacy never emits: `task_execution_started_total`, + `task_update_time_seconds`, `task_paused_total`, + `thread_uncaught_exceptions_total`, `workflow_start_error_total`, + `http_api_client_request_seconds`, `workflow_input_size_bytes`, and + `active_workers`. +- Canonical and legacy collectors are mutually exclusive. During a migration, + compare scrape output by running separate worker instances or environments + with and without `WORKER_CANONICAL_METRICS=true`. + +Legacy-to-canonical replacements: + +| Legacy metric | Canonical replacement | +|---|---| +| `task_poll_total{task_type}` | `task_poll_total{taskType}` | +| `task_poll_time_seconds{task_type}` | `task_poll_time_seconds{taskType,status}` | +| `task_poll_error_total{task_type,error}` | `task_poll_error_total{taskType,exception}` | +| `task_execute_time_seconds{task_type}` | `task_execute_time_seconds{taskType,status}` | +| `task_execute_error_total{task_type,exception,retryable}` | `task_execute_error_total{taskType,exception}` | +| `task_result_size_bytes{task_type}` | `task_result_size_bytes{taskType}` | +| `task_update_failed_total{task_type}` | `task_update_error_total{taskType,exception}` | + +--- + +## Prometheus Integration + +### Setup + +Add the `prometheus-client` gem to your Gemfile: + +```ruby +gem 'prometheus-client', '~> 4.0' +``` + +### Basic Usage + +```ruby +require 'conductor' + +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) + +metrics_server = Conductor::Worker::Telemetry::MetricsServer.new(port: 9090) +metrics_server.start + +handler = Conductor::Worker::TaskHandler.new( + configuration: config, + event_listeners: [metrics] +) + +handler.start +handler.join + +metrics_server.stop +``` + +### Metrics Endpoints + +The MetricsServer exposes: + +- `GET /metrics` - Prometheus metrics in text format +- `GET /health` - Health check endpoint (`{"status":"healthy"}`) + +### Kubernetes Integration + +```yaml +# Pod annotations for Prometheus scraping +metadata: + annotations: + prometheus.io/scrape: "true" + prometheus.io/port: "9090" + prometheus.io/path: "/metrics" +``` + +### Custom Prometheus Registry + +```ruby +require 'prometheus/client' + +registry = Prometheus::Client::Registry.new +backend = Conductor::Worker::Telemetry::PrometheusBackend.new(registry: registry) +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: backend) +``` + +--- + +## Custom Metrics Backends + +You can create a custom backend by implementing three methods: + +```ruby +class DatadogBackend + def initialize(statsd_client) + @statsd = statsd_client end - def on_task_execution_started(event) - @logger.info("Task started: #{event.task_id} (#{event.task_type})") + def increment(name, labels: {}) + tags = labels.map { |k, v| "#{k}:#{v}" } + @statsd.increment(name, tags: tags) end - def on_task_execution_completed(event) - @logger.info("Task completed: #{event.task_id} in #{event.duration_ms.round(2)}ms") + def observe(name, value, labels: {}) + tags = labels.map { |k, v| "#{k}:#{v}" } + @statsd.histogram(name, value, tags: tags) end - def on_task_execution_failure(event) - @logger.error("Task failed: #{event.task_id} - #{event.cause.message}") + def set(name, value, labels: {}) + tags = labels.map { |k, v| "#{k}:#{v}" } + @statsd.gauge(name, value, tags: tags) end end -# Create configuration -config = Conductor::Configuration.new( - server_api_url: ENV['CONDUCTOR_SERVER_URL'], - auth_key: ENV['CONDUCTOR_AUTH_KEY'], - auth_secret: ENV['CONDUCTOR_AUTH_SECRET'] +# Use with MetricsCollector +require 'datadog/statsd' +statsd = Datadog::Statsd.new('localhost', 8125) +metrics = Conductor::Worker::Telemetry::MetricsCollector.create( + backend: DatadogBackend.new(statsd) ) +``` -# Define a worker -Conductor::Worker.define('my_task') do |task| - # Worker logic - { result: 'success' } -end +--- -# Create handler with interceptor -handler = Conductor::Worker::TaskHandler.new( - configuration: config, - event_listeners: [LoggingInterceptor.new] -) +## Troubleshooting -# Start workers -handler.start -handler.join +### Metrics Are Empty + +- Verify that `MetricsCollector.create` is called and the collector is passed + to `TaskHandler` via `event_listeners:`. +- Verify workers have polled or executed tasks. Metrics are created lazily when + the relevant event occurs. +- Confirm the scrape endpoint is reachable at the expected host and port. + +### Missing HTTP or Workflow Metrics + +- `http_api_client_request_seconds` requires canonical mode. Legacy mode does + not emit HTTP metrics. The canonical collector auto-subscribes to + `GlobalDispatcher` for `HttpApiRequest` events from the HTTP layer. +- `workflow_input_size_bytes` and `workflow_start_error_total` require canonical + mode and only record when the corresponding `WorkflowExecutor` events fire. + +### High Cardinality + +- The `uri` label on `http_api_client_request_seconds` uses path templates + (e.g. `/workflow/{workflowId}`) to keep cardinality bounded. If you see + fully-resolved paths in your metrics, verify that HTTP requests are going + through the SDK's `ApiClient` rather than a standalone `RestClient`. +- Prefer canonical mode for bounded `exception` labels using exception class + names instead of raw error messages. +- Avoid embedding user identifiers or unbounded values in task type, workflow + type, or other label values. + +--- + +## Interceptor System + +The Conductor Ruby SDK provides an event-driven interceptor system that allows +you to: + +- **Monitor performance** - Track polling times, execution durations, error rates +- **Implement custom logging** - Add structured logging for task execution +- **Track errors** - Send failures to error tracking services (Sentry, Bugsnag, etc.) +- **Collect metrics** - Export to Prometheus, Datadog, or custom backends +- **Build alerting** - Monitor SLAs and trigger alerts on violations + +``` +TaskRunner + │ + │ publishes events + ▼ +SyncEventDispatcher ──────► Listener 1 (MetricsCollector) + ──────► Listener 2 (LoggingInterceptor) + ──────► Listener 3 (SentryInterceptor) ``` +When a worker polls for tasks, executes them, or encounters errors, events are +published to all registered listeners. Listeners can then process these events +independently. + --- ## Event Types -The SDK publishes the following events during worker execution: +The SDK publishes events during worker execution. Any object that responds to +the corresponding `on_*` method can listen for these events. ### Poll Events | Event | When Published | Key Attributes | -|-------|---------------|----------------| +|---|---|---| | `PollStarted` | Before polling for tasks | `task_type`, `worker_id`, `poll_count` | | `PollCompleted` | After successful poll | `task_type`, `duration_ms`, `tasks_received` | | `PollFailure` | When poll fails | `task_type`, `duration_ms`, `cause` | @@ -109,18 +485,42 @@ The SDK publishes the following events during worker execution: ### Execution Events | Event | When Published | Key Attributes | -|-------|---------------|----------------| +|---|---|---| | `TaskExecutionStarted` | Before task execution | `task_type`, `task_id`, `worker_id`, `workflow_instance_id` | | `TaskExecutionCompleted` | After successful execution | `task_type`, `task_id`, `duration_ms`, `output_size_bytes` | | `TaskExecutionFailure` | When execution fails | `task_type`, `task_id`, `duration_ms`, `cause`, `is_retryable` | -### Critical Events +### Update Events | Event | When Published | Key Attributes | -|-------|---------------|----------------| -| `TaskUpdateFailure` | When result update fails after all retries | `task_type`, `task_id`, `retry_count`, `task_result` | +|---|---|---| +| `TaskUpdateCompleted` | After successful result update | `task_type`, `task_id`, `duration_ms` | +| `TaskUpdateFailure` | When result update fails after all retries | `task_type`, `task_id`, `retry_count`, `task_result`, `cause` | + +### Worker State Events -**Important**: `TaskUpdateFailure` is a critical event indicating that a task result was lost. You should always handle this event to prevent silent data loss. +| Event | When Published | Key Attributes | +|---|---|---| +| `TaskPaused` | When a paused worker skips a poll | `task_type` | +| `ThreadUncaughtException` | When a worker thread raises an uncaught exception | `cause` | +| `ActiveWorkersChanged` | When the active worker count changes | `task_type`, `count` | + +### Workflow Events + +| Event | When Published | Key Attributes | +|---|---|---| +| `WorkflowStartError` | When starting a workflow fails client-side | `workflow_type`, `cause` | +| `WorkflowInputSize` | When a workflow is started | `workflow_type`, `version`, `size_bytes` | + +### HTTP Events + +| Event | When Published | Key Attributes | +|---|---|---| +| `HttpApiRequest` | After every HTTP API client request | `method`, `uri`, `status`, `duration_ms` | + +**Important**: `TaskUpdateFailure` is a critical event indicating that a task +result was lost. You should always handle this event to prevent silent data +loss. --- @@ -132,8 +532,6 @@ An interceptor is any object that responds to one or more `on_*` methods: ```ruby class MyInterceptor - # Implement only the methods you need - def on_poll_started(event) # Called before each poll end @@ -239,139 +637,6 @@ end --- -## Metrics Collection - -### Using the Built-in MetricsCollector - -The SDK includes a `MetricsCollector` that tracks key metrics: - -```ruby -require 'conductor' - -# Create metrics collector (uses NullBackend by default) -metrics = Conductor::Worker::Telemetry::MetricsCollector.new - -# Use with TaskHandler -handler = Conductor::Worker::TaskHandler.new( - configuration: config, - event_listeners: [metrics] -) -``` - -### Tracked Metrics - -| Metric | Type | Labels | Description | -|--------|------|--------|-------------| -| `task_poll_total` | Counter | `task_type` | Total poll operations | -| `task_poll_time_seconds` | Histogram | `task_type` | Poll latency | -| `task_poll_error_total` | Counter | `task_type`, `error` | Poll failures | -| `task_execute_time_seconds` | Histogram | `task_type` | Execution time | -| `task_execute_error_total` | Counter | `task_type`, `exception`, `retryable` | Execution failures | -| `task_result_size_bytes` | Histogram | `task_type` | Output size | -| `task_update_failed_total` | Counter | `task_type` | Failed result updates | - -### Custom Metrics Backend - -You can create a custom backend by implementing three methods: - -```ruby -class DatadogBackend - def initialize(statsd_client) - @statsd = statsd_client - end - - def increment(name, labels: {}) - tags = labels.map { |k, v| "#{k}:#{v}" } - @statsd.increment(name, tags: tags) - end - - def observe(name, value, labels: {}) - tags = labels.map { |k, v| "#{k}:#{v}" } - @statsd.histogram(name, value, tags: tags) - end - - def set(name, value, labels: {}) - tags = labels.map { |k, v| "#{k}:#{v}" } - @statsd.gauge(name, value, tags: tags) - end -end - -# Use with MetricsCollector -require 'datadog/statsd' -statsd = Datadog::Statsd.new('localhost', 8125) -metrics = Conductor::Worker::Telemetry::MetricsCollector.new( - backend: DatadogBackend.new(statsd) -) -``` - ---- - -## Prometheus Integration - -### Setup - -Add the `prometheus-client` gem to your Gemfile: - -```ruby -gem 'prometheus-client', '~> 4.0' -``` - -### Basic Usage - -```ruby -require 'conductor' - -# Create metrics collector with Prometheus backend -metrics = Conductor::Worker::Telemetry::MetricsCollector.new(backend: :prometheus) - -# Start metrics HTTP server -metrics_server = Conductor::Worker::Telemetry::MetricsServer.new(port: 9090) -metrics_server.start - -# Create handler with metrics -handler = Conductor::Worker::TaskHandler.new( - configuration: config, - event_listeners: [metrics] -) - -handler.start -handler.join - -# Cleanup -metrics_server.stop -``` - -### Metrics Endpoints - -The MetricsServer exposes: - -- `GET /metrics` - Prometheus metrics in text format -- `GET /health` - Health check endpoint (`{"status":"healthy"}`) - -### Kubernetes Integration - -```yaml -# Pod annotations for Prometheus scraping -metadata: - annotations: - prometheus.io/scrape: "true" - prometheus.io/port: "9090" - prometheus.io/path: "/metrics" -``` - -### Custom Prometheus Registry - -```ruby -require 'prometheus/client' - -# Use a custom registry -registry = Prometheus::Client::Registry.new -backend = Conductor::Worker::Telemetry::PrometheusBackend.new(registry: registry) -metrics = Conductor::Worker::Telemetry::MetricsCollector.new(backend: backend) -``` - ---- - ## Advanced Use Cases ### SLA Monitor @@ -416,7 +681,7 @@ Track compute costs per task type: ```ruby class CostTracker def initialize(cost_per_ms: {}) - @cost_per_ms = cost_per_ms # { 'task_type' => cost_per_ms } + @cost_per_ms = cost_per_ms @costs = Hash.new(0.0) @mutex = Mutex.new end @@ -472,7 +737,8 @@ end ### 1. Keep Interceptors Fast -Interceptors run synchronously in the worker thread. Keep processing fast to avoid impacting task execution: +Interceptors run synchronously in the worker thread. Keep processing fast to +avoid impacting task execution: ```ruby # BAD: Slow synchronous HTTP call @@ -488,34 +754,25 @@ end ### 2. Handle Errors in Interceptors -Errors in interceptors are caught and logged but don't affect other interceptors: +Errors in interceptors are caught and logged but don't affect other +interceptors: ```ruby def on_task_execution_completed(event) - # Safe to raise - won't crash the worker external_service.track(event) rescue => e - # Optionally log internally @logger.warn("Failed to track: #{e.message}") end ``` ### 3. Always Handle TaskUpdateFailure -This is a critical event - task results are lost: +This is a critical event -- task results are lost: ```ruby def on_task_update_failure(event) - # Log for debugging @logger.fatal("Task result lost: #{event.task_id}") - - # Alert operations team - PagerDuty.trigger( - severity: :critical, - summary: "Task result lost: #{event.task_id}" - ) - - # Optionally store for manual recovery + PagerDuty.trigger(severity: :critical, summary: "Task result lost: #{event.task_id}") FailedTaskStore.save(event.task_result) end ``` @@ -528,19 +785,16 @@ Separate concerns into different interceptors: handler = Conductor::Worker::TaskHandler.new( configuration: config, event_listeners: [ - MetricsCollector.new(backend: :prometheus), # Metrics - StructuredLoggingInterceptor.new, # Logging - SentryInterceptor.new, # Error tracking - SLAMonitor.new(thresholds: sla_config), # SLA monitoring - AuditLogger.new(log_file: 'audit.log') # Compliance + Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus), + StructuredLoggingInterceptor.new, + SentryInterceptor.new, + SLAMonitor.new(thresholds: sla_config), ] ) ``` ### 5. Test Your Interceptors -Write tests for your interceptors: - ```ruby RSpec.describe SentryInterceptor do let(:interceptor) { described_class.new } @@ -568,24 +822,26 @@ end ## Reference -### Event Classes +### Telemetry Classes -All events are in the `Conductor::Worker::Events` namespace: +- `Conductor::Worker::Telemetry::MetricsCollector` - Factory module; `.create` returns the appropriate collector +- `Conductor::Worker::Telemetry::LegacyMetricsCollector` - Pre-harmonization metric set with `task_type` labels +- `Conductor::Worker::Telemetry::CanonicalMetricsCollector` - Canonical metric set with `taskType` labels +- `Conductor::Worker::Telemetry::NullBackend` - No-op metrics backend +- `Conductor::Worker::Telemetry::PrometheusBackend` - Legacy Prometheus backend +- `Conductor::Worker::Telemetry::CanonicalPrometheusBackend` - Canonical Prometheus backend +- `Conductor::Worker::Telemetry::MetricsServer` - WEBrick HTTP server for `/metrics` and `/health` endpoints -- `Conductor::Worker::Events::PollStarted` -- `Conductor::Worker::Events::PollCompleted` -- `Conductor::Worker::Events::PollFailure` -- `Conductor::Worker::Events::TaskExecutionStarted` -- `Conductor::Worker::Events::TaskExecutionCompleted` -- `Conductor::Worker::Events::TaskExecutionFailure` -- `Conductor::Worker::Events::TaskUpdateFailure` +### Event Classes -### Telemetry Classes +All events are in the `Conductor::Worker::Events` namespace: -- `Conductor::Worker::Telemetry::MetricsCollector` - Event listener that collects metrics -- `Conductor::Worker::Telemetry::NullBackend` - No-op metrics backend -- `Conductor::Worker::Telemetry::PrometheusBackend` - Prometheus metrics backend -- `Conductor::Worker::Telemetry::MetricsServer` - HTTP server for `/metrics` endpoint +- `PollStarted`, `PollCompleted`, `PollFailure` +- `TaskExecutionStarted`, `TaskExecutionCompleted`, `TaskExecutionFailure` +- `TaskUpdateCompleted`, `TaskUpdateFailure` +- `TaskPaused`, `ThreadUncaughtException`, `ActiveWorkersChanged` +- `WorkflowStartError`, `WorkflowInputSize` +- `HttpApiRequest` ### Registration Methods @@ -604,3 +860,73 @@ Or register manually with the event dispatcher: dispatcher = handler.event_dispatcher dispatcher.register(Conductor::Worker::Events::PollStarted, ->(event) { puts event }) ``` + +--- + +## Detailed Technical Notes -- [Unreleased] + +### Path template `uri` label (metric_uri) + +The `uri` label on `http_api_client_request_seconds` now carries the **path +template** (e.g. `/workflow/{workflowId}`) rather than the fully-resolved +request path (e.g. `/api/workflow/abc-123-def`). This keeps metric label +cardinality bounded regardless of how many unique workflow IDs, task types, or +other dynamic path segments pass through the SDK. + +**Data flow:** + +1. Every API resource method calls `ApiClient#call_api` with a `resource_path` + that contains `{placeholder}` tokens (e.g. `/tasks/poll/batch/{taskType}`). +2. `ApiClient#call_api_no_retry` saves `resource_path` as `metric_uri` *before* + substituting path parameters. +3. The substituted path is concatenated with `server_url` to form the HTTP URL. + `metric_uri` is passed alongside as a keyword argument to + `RestClient#request`. +4. `RestClient#emit_http_event` prefers `metric_uri` when present; it only + falls back to extracting the path from the full URL when `metric_uri` is + `nil` (e.g. for direct `RestClient` calls outside of `ApiClient`). +5. The `HttpApiRequest` event carries the template string as its `uri` field. + `CanonicalMetricsCollector#on_http_api_request` records it as-is into the + histogram. + +This approach mirrors the Python SDK (`metric_uri` parameter), the Java SDK +(`PathTemplateTag` on the OkHttp request), and the Go SDK (`WithPathTemplate` +context value). The base-URL path prefix (e.g. `/api`) is never included +because `metric_uri` is always the raw API-relative resource path. + +### Canonical metrics factory + +`MetricsCollector.create(backend:)` selects the collector implementation based +on the `WORKER_CANONICAL_METRICS` environment variable: + +- **Unset / falsy** -- `LegacyMetricsCollector` (default). Emits the 0.1.0 + metric names and `snake_case` label conventions unchanged. +- **Truthy** (`true`, `1`, `yes`, case-insensitive) -- + `CanonicalMetricsCollector`. Emits the harmonized cross-SDK catalog with + `camelCase` domain labels, Prometheus histograms with explicit bucket + boundaries, and the `exception` label derived from the Ruby exception class + name. + +`WORKER_LEGACY_METRICS` is reserved for a future phase where canonical becomes +the default. + +### Event system additions + +The following event types were added for the canonical collector: + +- `HttpApiRequest` -- emitted by `RestClient` via the process-wide + `GlobalDispatcher`, but only when at least one `HttpApiRequest` listener + is subscribed (i.e. a `CanonicalMetricsCollector` is active). In legacy + mode or with no collector, `RestClient` skips all timing overhead. +- `WorkflowStartError`, `WorkflowInputSize` -- emitted by + `WorkflowExecutor`. +- `TaskUpdateCompleted`, `TaskPaused`, `ActiveWorkersChanged` -- emitted + by `TaskRunner`. +- `ThreadUncaughtException` -- event class and collector handler exist for + API completeness but are not currently emitted by any runner (see + [thread_uncaught_exceptions_total](#thread_uncaught_exceptions_total)). + +All events flow through the `SyncEventDispatcher` -> listener registry -> +collector pattern. The `GlobalDispatcher` singleton provides a secondary +channel so that `RestClient` (which has no direct reference to the task +handler's dispatcher) can still emit HTTP events. diff --git a/docs/design/EVENT_INTERCEPTOR_SYSTEM.md b/docs/design/EVENT_INTERCEPTOR_SYSTEM.md index 488dd28..d21ccb3 100644 --- a/docs/design/EVENT_INTERCEPTOR_SYSTEM.md +++ b/docs/design/EVENT_INTERCEPTOR_SYSTEM.md @@ -125,8 +125,11 @@ TaskRunner SyncEventDispatcher Listeners | `SyncEventDispatcher` | `events/sync_event_dispatcher.rb` | Thread-safe event router | | `TaskRunnerEventsListener` | `events/listeners.rb` | Listener protocol (duck typing) | | `ListenerRegistry` | `events/listener_registry.rb` | Bulk listener registration | -| `MetricsCollector` | `telemetry/metrics_collector.rb` | Event-based metrics | -| `PrometheusBackend` | `telemetry/prometheus_backend.rb` | Prometheus integration | +| `MetricsCollector` | `telemetry/metrics_collector.rb` | Factory (WORKER_CANONICAL_METRICS gate) | +| `LegacyMetricsCollector` | `telemetry/legacy_metrics_collector.rb` | Legacy metric set | +| `CanonicalMetricsCollector` | `telemetry/canonical_metrics_collector.rb` | Canonical metric set | +| `PrometheusBackend` | `telemetry/prometheus_backend.rb` | Legacy Prometheus backend | +| `CanonicalPrometheusBackend` | `telemetry/canonical_prometheus_backend.rb` | Canonical Prometheus backend | | `NullBackend` | `telemetry/metrics_collector.rb` | No-op backend | --- @@ -462,61 +465,19 @@ handler = Conductor::Worker::TaskHandler.new( ### MetricsCollector -The `MetricsCollector` implements `TaskRunnerEventsListener` to collect metrics: +The SDK supports legacy and canonical metric surfaces, selected by the +`WORKER_CANONICAL_METRICS` environment variable. `MetricsCollector.create` +returns the appropriate collector (`LegacyMetricsCollector` or +`CanonicalMetricsCollector`): ```ruby -module Conductor::Worker::Telemetry - class MetricsCollector - include Events::TaskRunnerEventsListener - - def initialize(backend: :null) - @backend = load_backend(backend) - end - - def on_poll_started(event) - @backend.increment('task_poll_total', labels: { task_type: event.task_type }) - end - - def on_poll_completed(event) - @backend.observe('task_poll_time_seconds', event.duration_ms / 1000.0, - labels: { task_type: event.task_type }) - end - - def on_poll_failure(event) - @backend.increment('task_poll_error_total', - labels: { - task_type: event.task_type, - error: event.cause.class.name - }) - end - - def on_task_execution_completed(event) - @backend.observe('task_execute_time_seconds', event.duration_ms / 1000.0, - labels: { task_type: event.task_type }) - - return unless event.output_size_bytes - - @backend.observe('task_result_size_bytes', event.output_size_bytes, - labels: { task_type: event.task_type }) - end - - def on_task_execution_failure(event) - @backend.increment('task_execute_error_total', - labels: { - task_type: event.task_type, - exception: event.cause.class.name, - retryable: event.is_retryable.to_s - }) - end - - def on_task_update_failure(event) - @backend.increment('task_update_failed_total', - labels: { task_type: event.task_type }) - end - end -end +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) ``` +See [docs/METRICS_AND_INTERCEPTORS.md](../METRICS_AND_INTERCEPTORS.md) for the +full legacy and canonical metrics catalogs, label reference, and migration +guide. + ### Backend Protocol Metrics backends must implement these methods: @@ -559,54 +520,16 @@ end ## Prometheus Integration -### PrometheusBackend - -The `PrometheusBackend` integrates with the `prometheus-client` gem: - -```ruby -module Conductor::Worker::Telemetry - class PrometheusBackend - # Default histogram buckets for time measurements (in seconds) - TIME_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10].freeze - - # Default histogram buckets for size measurements (in bytes) - SIZE_BUCKETS = [100, 1000, 10_000, 100_000, 1_000_000, 10_000_000].freeze - - def initialize(registry: nil) - require 'prometheus/client' - @registry = registry || Prometheus::Client.registry - setup_metrics - end - - def increment(name, labels: {}, value: 1) - metric = get_or_create_counter(name) - metric.increment(labels: normalize_labels(labels), by: value) - end - - def observe(name, value, labels: {}) - metric = get_or_create_histogram(name) - metric.observe(value, labels: normalize_labels(labels)) - end +### Prometheus Backends - def set(name, value, labels: {}) - metric = get_or_create_gauge(name) - metric.set(value, labels: normalize_labels(labels)) - end - end -end -``` +The SDK ships two Prometheus backends: -### Prometheus Metrics +- `PrometheusBackend` -- legacy metric registrations with `task_type` labels. +- `CanonicalPrometheusBackend` -- canonical metric registrations with `taskType` labels, `status` on time histograms, and canonical bucket boundaries. -| Metric Name | Type | Labels | Description | -|-------------|------|--------|-------------| -| `task_poll_total` | Counter | `task_type` | Total number of poll operations | -| `task_poll_time_seconds` | Histogram | `task_type` | Poll latency in seconds | -| `task_poll_error_total` | Counter | `task_type`, `error` | Total poll failures | -| `task_execute_time_seconds` | Histogram | `task_type` | Task execution time in seconds | -| `task_execute_error_total` | Counter | `task_type`, `exception`, `retryable` | Total execution failures | -| `task_result_size_bytes` | Histogram | `task_type` | Task output size in bytes | -| `task_update_failed_total` | Counter | `task_type` | **CRITICAL**: Failed task updates | +Both implement `increment`, `observe`, and `set` and integrate with the +`prometheus-client` gem. See [docs/METRICS_AND_INTERCEPTORS.md](../METRICS_AND_INTERCEPTORS.md) +for the full metric catalog emitted by each backend. ### MetricsServer @@ -663,7 +586,7 @@ config = Conductor::Configuration.new( ) # Create metrics collector with Prometheus backend -metrics = Conductor::Worker::Telemetry::MetricsCollector.new(backend: :prometheus) +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) # Start metrics server metrics_server = Conductor::Worker::Telemetry::MetricsServer.new(port: 9090) @@ -756,7 +679,7 @@ end handler = Conductor::Worker::TaskHandler.new( configuration: config, event_listeners: [ - Conductor::Worker::Telemetry::MetricsCollector.new(backend: :prometheus), + Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus), LoggingInterceptor.new, SentryInterceptor.new ] @@ -948,8 +871,11 @@ lib/conductor/worker/ │ ├── listeners.rb # TaskRunnerEventsListener protocol │ └── listener_registry.rb # Bulk listener registration helper ├── telemetry/ -│ ├── metrics_collector.rb # MetricsCollector + NullBackend -│ └── prometheus_backend.rb # PrometheusBackend + MetricsServer +│ ├── metrics_collector.rb # Factory (WORKER_CANONICAL_METRICS gate) + NullBackend +│ ├── legacy_metrics_collector.rb # Legacy metric set +│ ├── canonical_metrics_collector.rb # Canonical metric set +│ ├── prometheus_backend.rb # Legacy PrometheusBackend + MetricsServer +│ └── canonical_prometheus_backend.rb # Canonical PrometheusBackend ├── task_runner.rb # Publishes events during polling/execution └── task_handler.rb # Creates dispatcher, registers listeners @@ -961,7 +887,10 @@ spec/conductor/worker/ │ └── listener_registry_spec.rb └── telemetry/ ├── metrics_collector_spec.rb - └── prometheus_backend_spec.rb + ├── legacy_metrics_collector_spec.rb + ├── canonical_metrics_collector_spec.rb + ├── prometheus_backend_spec.rb + └── canonical_prometheus_backend_spec.rb ``` --- diff --git a/docs/design/WORKER_DESIGN.md b/docs/design/WORKER_DESIGN.md index 5229894..1f190c4 100644 --- a/docs/design/WORKER_DESIGN.md +++ b/docs/design/WORKER_DESIGN.md @@ -1067,26 +1067,22 @@ def update_task_with_retry(task_result) RETRY_BACKOFFS.each_with_index do |backoff, attempt| sleep(backoff) if backoff > 0 + start_time = Time.now begin @task_client.update_task(task_result) + duration_ms = (Time.now - start_time) * 1000 + + publish_task_update_completed(task_result, duration_ms) return # Success rescue StandardError => e + duration_ms = (Time.now - start_time) * 1000 @logger.error("Task update failed (attempt #{attempt + 1}/#{RETRY_BACKOFFS.size}): #{e.message}") if attempt == RETRY_BACKOFFS.size - 1 # All retries exhausted - CRITICAL: task result is lost @logger.fatal("CRITICAL: Task update failed after #{RETRY_BACKOFFS.size} attempts. " \ "Task #{task_result.task_id} result is LOST.") - - @event_dispatcher.publish(Events::TaskUpdateFailure.new( - task_type: @worker.task_definition_name, - task_id: task_result.task_id, - worker_id: @worker_id, - workflow_instance_id: task_result.workflow_instance_id, - cause: e, - retry_count: RETRY_BACKOFFS.size, - task_result: task_result # Include result for recovery - )) + publish_task_update_failure(task_result, e, duration_ms) end end end @@ -1379,88 +1375,18 @@ end ### MetricsCollector -```ruby -module Conductor - module Worker - class MetricsCollector - include TaskRunnerEventsListener - - def initialize(backend: :prometheus) - @backend = load_backend(backend) - end - - def on_poll_started(event) - @backend.increment("task_poll_total", labels: { task_type: event.task_type }) - end - - def on_poll_completed(event) - @backend.observe("task_poll_time_seconds", event.duration_ms / 1000.0, - labels: { task_type: event.task_type }) - end - - def on_poll_failure(event) - @backend.increment("task_poll_error_total", - labels: { task_type: event.task_type, error: event.cause.class.name }) - end - - def on_task_execution_completed(event) - @backend.observe("task_execute_time_seconds", event.duration_ms / 1000.0, - labels: { task_type: event.task_type }) - @backend.observe("task_result_size_bytes", event.output_size_bytes, - labels: { task_type: event.task_type }) - end - - def on_task_execution_failure(event) - @backend.increment("task_execute_error_total", - labels: { - task_type: event.task_type, - exception: event.cause.class.name, - retryable: event.is_retryable.to_s - }) - end - - def on_task_update_failure(event) - @backend.increment("task_update_failed_total", - labels: { task_type: event.task_type }) - end - - private - - def load_backend(backend) - case backend - when :prometheus - require_prometheus - PrometheusBackend.new - when :null, nil - NullBackend.new - else - backend # Custom backend instance - end - end +The SDK supports legacy and canonical metric surfaces, selected by the +`WORKER_CANONICAL_METRICS` environment variable. `MetricsCollector.create` +returns the appropriate collector (`LegacyMetricsCollector` or +`CanonicalMetricsCollector`): - def require_prometheus - require 'prometheus/client' - rescue LoadError - raise ConfigurationError, - "The 'prometheus-client' gem is required for Prometheus metrics. " \ - "Add `gem 'prometheus-client'` to your Gemfile." - end - end - end -end +```ruby +metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) ``` -### Prometheus Metric Names - -| Metric | Type | Labels | Description | -|--------|------|--------|-------------| -| `task_poll_total` | Counter | `task_type` | Number of poll operations | -| `task_poll_time_seconds` | Histogram | `task_type` | Poll latency | -| `task_poll_error_total` | Counter | `task_type`, `error` | Poll failures | -| `task_execute_time_seconds` | Histogram | `task_type` | Execution time | -| `task_execute_error_total` | Counter | `task_type`, `exception`, `retryable` | Execution failures | -| `task_result_size_bytes` | Histogram | `task_type` | Output size | -| `task_update_failed_total` | Counter | `task_type` | CRITICAL: Update failures | +See [docs/METRICS_AND_INTERCEPTORS.md](../METRICS_AND_INTERCEPTORS.md) for the +full legacy and canonical metrics catalogs, label reference, and migration +guide. --- @@ -1637,8 +1563,11 @@ lib/conductor/ │ ├── listener_registry.rb # Listener registration helper │ └── listeners.rb # Listener protocol module ├── worker/telemetry/ -│ ├── metrics_collector.rb # Event-based metrics -│ ├── prometheus_backend.rb # Prometheus integration +│ ├── metrics_collector.rb # Factory (WORKER_CANONICAL_METRICS gate) +│ ├── legacy_metrics_collector.rb # Legacy metric set +│ ├── canonical_metrics_collector.rb # Canonical metric set +│ ├── prometheus_backend.rb # Legacy Prometheus backend +│ ├── canonical_prometheus_backend.rb # Canonical Prometheus backend │ └── null_backend.rb # No-op backend └── exceptions.rb # Add NonRetryableError ``` diff --git a/examples/metrics_example.rb b/examples/metrics_example.rb index ef130aa..ad6e21b 100644 --- a/examples/metrics_example.rb +++ b/examples/metrics_example.rb @@ -130,7 +130,7 @@ def main if PROMETHEUS_AVAILABLE # Create metrics collector with Prometheus backend - metrics = Conductor::Worker::Telemetry::MetricsCollector.new(backend: :prometheus) + metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) # Start metrics HTTP server metrics_server = Conductor::Worker::Telemetry::MetricsServer.new(port: metrics_port) @@ -141,7 +141,7 @@ def main puts "Health check: http://localhost:#{metrics_port}/health" else # Fall back to null metrics (logging only) - metrics = Conductor::Worker::Telemetry::MetricsCollector.new(backend: :null) + metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :null) metrics_server = nil puts 'Metrics mode: Null (prometheus-client gem not installed)' @@ -240,7 +240,7 @@ def main_with_custom_backend ) custom_backend = CustomMetricsBackend.new - metrics = Conductor::Worker::Telemetry::MetricsCollector.new(backend: custom_backend) + metrics = Conductor::Worker::Telemetry::MetricsCollector.create(backend: custom_backend) puts 'Using custom metrics backend...' diff --git a/harness/main.rb b/harness/main.rb index 5227d52..3c51035 100644 --- a/harness/main.rb +++ b/harness/main.rb @@ -6,6 +6,7 @@ require 'conductor/worker/telemetry/prometheus_backend' require_relative 'simulated_task_worker' require_relative 'workflow_governor' +require_relative 'workflow_status_probe' module Harness WORKFLOW_NAME = 'ruby_simulated_tasks_workflow' @@ -28,19 +29,20 @@ def self.env_int(name, default) def self.main $stdout.sync = true - workflows_per_sec = env_int('HARNESS_WORKFLOWS_PER_SEC', 2) - batch_size = env_int('HARNESS_BATCH_SIZE', 20) - poll_interval_ms = env_int('HARNESS_POLL_INTERVAL_MS', 100) + workflows_per_sec = env_int('HARNESS_WORKFLOWS_PER_SEC', 2) + batch_size = env_int('HARNESS_BATCH_SIZE', 20) + poll_interval_ms = env_int('HARNESS_POLL_INTERVAL_MS', 100) + probe_rate_per_sec = env_int('HARNESS_PROBE_RATE_PER_SEC', 0) metrics_port = env_int('HARNESS_METRICS_PORT', 9991) configuration = Conductor::Configuration.new register_metadata(configuration) - metrics_collector = Conductor::Worker::Telemetry::MetricsCollector.new(backend: :prometheus) + metrics_collector = Conductor::Worker::Telemetry::MetricsCollector.create(backend: :prometheus) metrics_server = Conductor::Worker::Telemetry::MetricsServer.new(port: metrics_port) metrics_server.start - puts "Prometheus metrics server started on port #{metrics_port}" + puts "Prometheus metrics server started on port #{metrics_port} (#{metrics_collector.collector_name} metrics)" workers = SIMULATED_WORKERS.map do |def_entry| sim = SimulatedTaskWorker.new( @@ -68,13 +70,23 @@ def self.main ) task_handler.start - workflow_executor = Conductor::Workflow::WorkflowExecutor.new(configuration) - governor = WorkflowGovernor.new(workflow_executor, WORKFLOW_NAME, workflows_per_sec) + workflow_executor = Conductor::Workflow::WorkflowExecutor.new( + configuration, + event_dispatcher: task_handler.event_dispatcher + ) + + workflow_client = Conductor::Client::WorkflowClient.new(configuration) + probe = WorkflowStatusProbe.new(workflow_client, probe_rate_per_sec) + probe.start + + governor = WorkflowGovernor.new(workflow_executor, WORKFLOW_NAME, workflows_per_sec, + id_sink: probe.method(:offer)) governor.start shutdown = proc do puts 'Shutting down...' governor.stop + probe.stop task_handler.stop exit(0) end diff --git a/harness/manifests/deployment.yaml b/harness/manifests/deployment.yaml index 0036ec9..0a0dbfa 100644 --- a/harness/manifests/deployment.yaml +++ b/harness/manifests/deployment.yaml @@ -53,6 +53,14 @@ spec: - name: HARNESS_POLL_INTERVAL_MS value: "100" + # === METRICS IMPLEMENTATION === + # Set to "true" to use the canonical (harmonized) metric set. + # Default "false" uses legacy metrics during the deprecation period. + # In a future release, canonical will become the default and + # WORKER_LEGACY_METRICS will allow opting back into legacy. + - name: WORKER_CANONICAL_METRICS + value: "true" + ports: - name: metrics containerPort: 9991 diff --git a/harness/workflow_governor.rb b/harness/workflow_governor.rb index 14cbe88..7f6f156 100644 --- a/harness/workflow_governor.rb +++ b/harness/workflow_governor.rb @@ -4,10 +4,11 @@ module Harness # WorkflowGovernor -- starts a configurable number of workflow instances # per second in a background thread, feeding work to the simulated workers. class WorkflowGovernor - def initialize(workflow_executor, workflow_name, workflows_per_second) + def initialize(workflow_executor, workflow_name, workflows_per_second, id_sink: nil) @workflow_executor = workflow_executor @workflow_name = workflow_name @workflows_per_second = workflows_per_second + @id_sink = id_sink @running = false @thread = nil end @@ -38,8 +39,9 @@ def run_loop def start_batch @workflows_per_second.times do - request = Conductor::Http::Models::StartWorkflowRequest.new(name: @workflow_name) - @workflow_executor.start_workflow(request) + request = Conductor::Http::Models::StartWorkflowRequest.new(name: @workflow_name, version: 1) + id = @workflow_executor.start_workflow(request) + @id_sink&.call(id) if id end puts "Governor: started #{@workflows_per_second} workflow(s)" rescue StandardError => e diff --git a/harness/workflow_status_probe.rb b/harness/workflow_status_probe.rb new file mode 100644 index 0000000..2b4fc03 --- /dev/null +++ b/harness/workflow_status_probe.rb @@ -0,0 +1,86 @@ +# frozen_string_literal: true + +module Harness + MAX_TRACKED_IDS = 256 + + # WorkflowStatusProbe exercises UUID-bearing workflow lookup endpoints so + # http_api_client_request_seconds picks up entries with + # uri=/workflow/{workflowId} and uri=/workflow/{workflowId}/status. + # + # Default harness traffic only hits bounded, no-path-param URLs (poll/update), + # making the high-cardinality concern on the uri label invisible without this + # probe. + # + # Default off. Runs only when HARNESS_PROBE_RATE_PER_SEC > 0. + # Side-effect-free: only issues read calls (get_execution_status, get_workflow_status). + # Self-bounded: fixed-size FIFO of workflow IDs. + class WorkflowStatusProbe + def initialize(workflow_client, calls_per_second) + @workflow_client = workflow_client + @calls_per_second = calls_per_second + @recent_ids = [] + @mutex = Mutex.new + @rng = Random.new + @running = false + @thread = nil + end + + # Capture a workflow ID for later probing. Thread-safe. + def offer(workflow_id) + return if workflow_id.nil? || workflow_id.empty? + + @mutex.synchronize do + @recent_ids << workflow_id + @recent_ids.shift(@recent_ids.size - MAX_TRACKED_IDS) if @recent_ids.size > MAX_TRACKED_IDS + end + end + + def start + if @calls_per_second <= 0 + puts 'WorkflowStatusProbe disabled (HARNESS_PROBE_RATE_PER_SEC<=0)' + return self + end + + @running = true + puts "WorkflowStatusProbe started: rate=#{@calls_per_second}/sec, retainedIds<=#{MAX_TRACKED_IDS}" + + @thread = Thread.new { run_loop } + @thread.name = 'workflow-status-probe' + self + end + + def stop + @running = false + @thread&.join(5) + puts 'WorkflowStatusProbe stopped' + end + + private + + def run_loop + while @running + tick + sleep(1) + end + end + + def tick + ids = @mutex.synchronize do + budget = [@calls_per_second, @recent_ids.size].min + return if budget.zero? + + Array.new(budget) { @recent_ids[@rng.rand(@recent_ids.size)] } + end + + ids.each do |id| + if @rng.rand < 0.5 + @workflow_client.workflow_api.get_execution_status(id, include_tasks: false) + else + @workflow_client.workflow_api.get_workflow_status(id) + end + rescue StandardError => e + puts "probe: #{id}: #{e.message}" + end + end + end +end diff --git a/lib/conductor.rb b/lib/conductor.rb index 64fc84d..3c8bf3e 100644 --- a/lib/conductor.rb +++ b/lib/conductor.rb @@ -90,7 +90,10 @@ # Worker Infrastructure require_relative 'conductor/worker/events/conductor_event' require_relative 'conductor/worker/events/task_runner_events' +require_relative 'conductor/worker/events/workflow_events' +require_relative 'conductor/worker/events/http_events' require_relative 'conductor/worker/events/sync_event_dispatcher' +require_relative 'conductor/worker/events/global_dispatcher' require_relative 'conductor/worker/events/listeners' require_relative 'conductor/worker/events/listener_registry' require_relative 'conductor/worker/task_context' diff --git a/lib/conductor/http/api_client.rb b/lib/conductor/http/api_client.rb index 12ff1b4..91fc33f 100644 --- a/lib/conductor/http/api_client.rb +++ b/lib/conductor/http/api_client.rb @@ -197,6 +197,8 @@ def call_api_no_retry(resource_path, method, opts = {}) return_type = opts[:return_type] return_http_data_only = opts[:return_http_data_only] || false + metric_uri = resource_path + # Replace path parameters path_params.each do |key, value| resource_path = resource_path.sub("{#{key}}", URI.encode_www_form_component(value.to_s)) @@ -220,7 +222,8 @@ def call_api_no_retry(resource_path, method, opts = {}) url, query: query_params, headers: header_params, - body: body ? JSON.generate(body) : nil + body: body ? JSON.generate(body) : nil, + metric_uri: metric_uri ) @last_response = response diff --git a/lib/conductor/http/rest_client.rb b/lib/conductor/http/rest_client.rb index 6ffcd5e..c51ff98 100644 --- a/lib/conductor/http/rest_client.rb +++ b/lib/conductor/http/rest_client.rb @@ -4,6 +4,8 @@ require 'faraday/net_http_persistent' require 'faraday/retry' require 'json' +require 'logger' +require 'uri' module Conductor module Http @@ -11,29 +13,44 @@ module Http class RestClient attr_reader :connection - def initialize(configuration = nil) + def initialize(configuration = nil, logger: nil) @configuration = configuration + @logger = logger || Logger.new(File::NULL) @connection = build_connection end # Main request method - def request(method, url, query: nil, headers: nil, body: nil) + def request(method, url, query: nil, headers: nil, body: nil, metric_uri: nil) method = method.to_s.upcase raise ArgumentError, "Invalid HTTP method: #{method}" unless valid_method?(method) headers ||= {} headers['Content-Type'] ||= 'application/json' if %w[POST PUT PATCH DELETE OPTIONS].include?(method) - response = @connection.run_request(method.downcase.to_sym, url, nil, headers) do |req| - req.params = query if query - req.body = serialize_body(body, headers['Content-Type']) if body - end + timing = http_metrics_enabled? + start_time = Time.now if timing + status_code = '0' - handle_response(response) - rescue Faraday::TimeoutError => e - raise ApiError.new("Request timeout: #{e.message}", status: 0, reason: 'Timeout') - rescue Faraday::ConnectionFailed => e - raise ApiError.new("Connection error: #{e.message}", status: 0, reason: 'ConnectionFailed') + begin + response = @connection.run_request(method.downcase.to_sym, url, nil, headers) do |req| + req.params = query if query + req.body = serialize_body(body, headers['Content-Type']) if body + end + status_code = response.status.to_s + + result = handle_response(response) + emit_http_event(method, url, status_code, start_time, metric_uri: metric_uri) if timing + result + rescue Faraday::TimeoutError => e + emit_http_event(method, url, '0', start_time, metric_uri: metric_uri) if timing + raise ApiError.new("Request timeout: #{e.message}", status: 0, reason: 'Timeout') + rescue Faraday::ConnectionFailed => e + emit_http_event(method, url, '0', start_time, metric_uri: metric_uri) if timing + raise ApiError.new("Connection error: #{e.message}", status: 0, reason: 'ConnectionFailed') + rescue ApiError, AuthorizationError + emit_http_event(method, url, status_code, start_time, metric_uri: metric_uri) if timing + raise + end end # Convenience methods @@ -71,6 +88,22 @@ def close private + def http_metrics_enabled? + defined?(Conductor::Worker::Events::GlobalDispatcher) && + Conductor::Worker::Events::GlobalDispatcher.http_metrics_enabled? + end + + def emit_http_event(method, url, status, start_time, metric_uri: nil) + duration_ms = (Time.now - start_time) * 1000 + uri_path = metric_uri || URI.parse(url).request_uri + event = Conductor::Worker::Events::HttpApiRequest.new( + method: method, uri: uri_path, status: status, duration_ms: duration_ms + ) + Conductor::Worker::Events::GlobalDispatcher.publish(event) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + def build_connection Faraday.new do |conn| # HTTP/2 adapter with persistent connections diff --git a/lib/conductor/worker/events/global_dispatcher.rb b/lib/conductor/worker/events/global_dispatcher.rb new file mode 100644 index 0000000..aba3211 --- /dev/null +++ b/lib/conductor/worker/events/global_dispatcher.rb @@ -0,0 +1,37 @@ +# frozen_string_literal: true + +require_relative 'sync_event_dispatcher' + +module Conductor + module Worker + module Events + # Process-wide default event dispatcher. + # + # Used by library layers (e.g. HTTP client) that don't have an obvious + # owner to receive a dispatcher reference. Listeners can subscribe to this + # singleton to receive events regardless of which RestClient/ApiClient + # instance generated them. + class GlobalDispatcher + class << self + def instance + @instance ||= SyncEventDispatcher.new + end + + def reset! + @instance = SyncEventDispatcher.new + end + + def publish(event) + instance.publish(event) + end + + def http_metrics_enabled? + instance.has_listeners?(HttpApiRequest) + rescue StandardError + false + end + end + end + end + end +end diff --git a/lib/conductor/worker/events/http_events.rb b/lib/conductor/worker/events/http_events.rb new file mode 100644 index 0000000..46257cd --- /dev/null +++ b/lib/conductor/worker/events/http_events.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +require_relative 'conductor_event' + +module Conductor + module Worker + module Events + class HttpApiRequest < ConductorEvent + attr_reader :method, :uri, :status, :duration_ms + + def initialize(method:, uri:, status:, duration_ms:) + super() + @method = method.to_s.upcase + @uri = uri.to_s + @status = status.to_s + @duration_ms = duration_ms + end + + def to_h + super.merge(method: @method, uri: @uri, status: @status, duration_ms: @duration_ms) + end + end + end + end +end diff --git a/lib/conductor/worker/events/listener_registry.rb b/lib/conductor/worker/events/listener_registry.rb index 5733430..b7610c0 100644 --- a/lib/conductor/worker/events/listener_registry.rb +++ b/lib/conductor/worker/events/listener_registry.rb @@ -1,14 +1,13 @@ # frozen_string_literal: true require_relative 'task_runner_events' +require_relative 'workflow_events' +require_relative 'http_events' module Conductor module Worker module Events - # Helper class to register listener objects with event dispatchers - # Uses duck typing to detect which methods a listener implements class ListenerRegistry - # Mapping of event classes to listener method names EVENT_METHOD_MAP = { PollStarted => :on_poll_started, PollCompleted => :on_poll_completed, @@ -16,28 +15,24 @@ class ListenerRegistry TaskExecutionStarted => :on_task_execution_started, TaskExecutionCompleted => :on_task_execution_completed, TaskExecutionFailure => :on_task_execution_failure, - TaskUpdateFailure => :on_task_update_failure + TaskUpdateCompleted => :on_task_update_completed, + TaskUpdateFailure => :on_task_update_failure, + TaskPaused => :on_task_paused, + ThreadUncaughtException => :on_thread_uncaught_exception, + ActiveWorkersChanged => :on_active_workers_changed, + WorkflowStartError => :on_workflow_start_error, + WorkflowInputSize => :on_workflow_input_size, + HttpApiRequest => :on_http_api_request }.freeze - # Register a listener object with the dispatcher - # Auto-detects implemented methods via respond_to? - # @param listener [Object] Object implementing TaskRunnerEventsListener methods - # @param dispatcher [SyncEventDispatcher] Event dispatcher - # @return [void] def self.register_task_runner_listener(listener, dispatcher) EVENT_METHOD_MAP.each do |event_class, method_name| dispatcher.register(event_class, ->(event) { listener.send(method_name, event) }) if listener.respond_to?(method_name) end end - # Register multiple listeners with the dispatcher - # @param listeners [Array] Array of listener objects - # @param dispatcher [SyncEventDispatcher] Event dispatcher - # @return [void] def self.register_all(listeners, dispatcher) - listeners.each do |listener| - register_task_runner_listener(listener, dispatcher) - end + listeners.each { |listener| register_task_runner_listener(listener, dispatcher) } end end end diff --git a/lib/conductor/worker/events/listeners.rb b/lib/conductor/worker/events/listeners.rb index 85ee75d..5d53b0b 100644 --- a/lib/conductor/worker/events/listeners.rb +++ b/lib/conductor/worker/events/listeners.rb @@ -3,38 +3,31 @@ module Conductor module Worker module Events - # Listener protocol for task runner events - # Include this module to document the expected interface - # All methods are optional - only implement the ones you need - # The dispatcher uses duck typing (respond_to?) to check for methods + # Listener protocol for task runner events. + # All methods are optional - the dispatcher uses duck typing (respond_to?). module TaskRunnerEventsListener - # Called when polling starts - # @param event [PollStarted] def on_poll_started(event); end - - # Called when polling completes successfully - # @param event [PollCompleted] def on_poll_completed(event); end - - # Called when polling fails - # @param event [PollFailure] def on_poll_failure(event); end - - # Called when task execution starts - # @param event [TaskExecutionStarted] def on_task_execution_started(event); end - - # Called when task execution completes successfully - # @param event [TaskExecutionCompleted] def on_task_execution_completed(event); end - - # Called when task execution fails - # @param event [TaskExecutionFailure] def on_task_execution_failure(event); end - - # Called when task update fails after all retries (CRITICAL) - # @param event [TaskUpdateFailure] + def on_task_update_completed(event); end def on_task_update_failure(event); end + def on_task_paused(event); end + def on_thread_uncaught_exception(event); end + def on_active_workers_changed(event); end + end + + # Listener protocol for workflow-lifecycle events + module WorkflowEventsListener + def on_workflow_start_error(event); end + def on_workflow_input_size(event); end + end + + # Listener protocol for HTTP API client events + module HttpEventsListener + def on_http_api_request(event); end end end end diff --git a/lib/conductor/worker/events/task_runner_events.rb b/lib/conductor/worker/events/task_runner_events.rb index 1e458e6..ab28368 100644 --- a/lib/conductor/worker/events/task_runner_events.rb +++ b/lib/conductor/worker/events/task_runner_events.rb @@ -187,31 +187,34 @@ def to_h end end + # Published when task update completes successfully + class TaskUpdateCompleted < TaskRunnerEvent + attr_reader :task_id, :worker_id, :workflow_instance_id, :duration_ms + + def initialize(task_type:, task_id:, worker_id:, workflow_instance_id:, duration_ms:) + super(task_type: task_type) + @task_id = task_id + @worker_id = worker_id + @workflow_instance_id = workflow_instance_id + @duration_ms = duration_ms + end + + def to_h + super.merge( + task_id: @task_id, worker_id: @worker_id, + workflow_instance_id: @workflow_instance_id, duration_ms: @duration_ms + ) + end + end + # Published when task update fails after all retries # This is a CRITICAL event - the task result is lost class TaskUpdateFailure < TaskRunnerEvent - # @return [String] Unique task identifier - attr_reader :task_id - # @return [String] Unique worker identifier - attr_reader :worker_id - # @return [String] Workflow instance identifier - attr_reader :workflow_instance_id - # @return [Exception] The exception that caused the failure - attr_reader :cause - # @return [Integer] Number of retry attempts made - attr_reader :retry_count - # @return [TaskResult] The task result that failed to update (for recovery) - attr_reader :task_result + attr_reader :task_id, :worker_id, :workflow_instance_id, + :cause, :retry_count, :task_result, :duration_ms - # @param task_type [String] Task definition name - # @param task_id [String] Unique task identifier - # @param worker_id [String] Unique worker identifier - # @param workflow_instance_id [String] Workflow instance identifier - # @param cause [Exception] The exception that caused the failure - # @param retry_count [Integer] Number of retry attempts made - # @param task_result [TaskResult] The task result that failed to update def initialize(task_type:, task_id:, worker_id:, workflow_instance_id:, - cause:, retry_count:, task_result:) + cause:, retry_count:, task_result:, duration_ms: nil) super(task_type: task_type) @task_id = task_id @worker_id = worker_id @@ -219,19 +222,50 @@ def initialize(task_type:, task_id:, worker_id:, workflow_instance_id:, @cause = cause @retry_count = retry_count @task_result = task_result + @duration_ms = duration_ms end def to_h super.merge( - task_id: @task_id, - worker_id: @worker_id, + task_id: @task_id, worker_id: @worker_id, workflow_instance_id: @workflow_instance_id, - cause: @cause.class.name, - cause_message: @cause.message, - retry_count: @retry_count + cause: @cause.class.name, cause_message: @cause.message, + retry_count: @retry_count, duration_ms: @duration_ms ) end end + + # Published when a poll iteration is skipped because the worker is paused + class TaskPaused < TaskRunnerEvent; end + + # Published when a worker thread terminates with an uncaught exception + class ThreadUncaughtException < ConductorEvent + attr_reader :cause, :task_type + + def initialize(cause:, task_type: nil) + super() + @cause = cause + @task_type = task_type + end + + def to_h + super.merge(cause: @cause.class.name, cause_message: @cause.message, task_type: @task_type) + end + end + + # Published when the active-worker count changes for a task type + class ActiveWorkersChanged < TaskRunnerEvent + attr_reader :count + + def initialize(task_type:, count:) + super(task_type: task_type) + @count = count + end + + def to_h + super.merge(count: @count) + end + end end end end diff --git a/lib/conductor/worker/events/workflow_events.rb b/lib/conductor/worker/events/workflow_events.rb new file mode 100644 index 0000000..b117396 --- /dev/null +++ b/lib/conductor/worker/events/workflow_events.rb @@ -0,0 +1,49 @@ +# frozen_string_literal: true + +require_relative 'conductor_event' + +module Conductor + module Worker + module Events + class WorkflowEvent < ConductorEvent + attr_reader :workflow_type, :version + + def initialize(workflow_type:, version: nil) + super() + @workflow_type = workflow_type + @version = version + end + + def to_h + super.merge(workflow_type: @workflow_type, version: @version) + end + end + + class WorkflowStartError < WorkflowEvent + attr_reader :cause + + def initialize(workflow_type:, cause:, version: nil) + super(workflow_type: workflow_type, version: version) + @cause = cause + end + + def to_h + super.merge(cause: @cause.class.name, cause_message: @cause.message) + end + end + + class WorkflowInputSize < WorkflowEvent + attr_reader :size_bytes + + def initialize(workflow_type:, size_bytes:, version: nil) + super(workflow_type: workflow_type, version: version) + @size_bytes = size_bytes + end + + def to_h + super.merge(size_bytes: @size_bytes) + end + end + end + end +end diff --git a/lib/conductor/worker/fiber_executor.rb b/lib/conductor/worker/fiber_executor.rb index cf3b97f..57c56b6 100644 --- a/lib/conductor/worker/fiber_executor.rb +++ b/lib/conductor/worker/fiber_executor.rb @@ -227,6 +227,7 @@ def run_once(executor) tasks.each do |task| executor.submit { execute_and_update(task) } end + publish_active_workers(executor) end end @@ -258,7 +259,10 @@ def calculate_adaptive_backoff end def batch_poll(count) - return [] if @worker.paused + if @worker.paused + @event_dispatcher.publish(Events::TaskPaused.new(task_type: @worker.task_definition_name)) + return [] + end if @auth_failures.positive? && @last_auth_failure_time backoff_seconds = [2**@auth_failures, MAX_AUTH_BACKOFF_SECONDS].min @@ -438,27 +442,63 @@ def update_task_with_retry(task_result) RETRY_BACKOFFS.each_with_index do |backoff, attempt| sleep(backoff) if backoff.positive? + start_time = Time.now begin @task_client.update_task(task_result) + duration_ms = (Time.now - start_time) * 1000 + + publish_task_update_completed(task_result, duration_ms) return rescue StandardError => e + duration_ms = (Time.now - start_time) * 1000 @logger.error("Update failed (attempt #{attempt + 1}): #{e.message}") - if attempt == RETRY_BACKOFFS.size - 1 - @event_dispatcher.publish(Events::TaskUpdateFailure.new( - task_type: @worker.task_definition_name, - task_id: task_result.task_id, - worker_id: @worker_id, - workflow_instance_id: task_result.workflow_instance_id, - cause: e, - retry_count: RETRY_BACKOFFS.size, - task_result: task_result - )) - end + publish_task_update_failure(task_result, e, duration_ms) if attempt == RETRY_BACKOFFS.size - 1 end end end + def publish_task_update_completed(task_result, duration_ms) + @event_dispatcher.publish(Events::TaskUpdateCompleted.new( + task_type: @worker.task_definition_name, + task_id: task_result.task_id, + worker_id: @worker_id, + workflow_instance_id: task_result.workflow_instance_id, + duration_ms: duration_ms + )) + end + + def publish_task_update_failure(task_result, error, duration_ms) + @event_dispatcher.publish(Events::TaskUpdateFailure.new( + task_type: @worker.task_definition_name, + task_id: task_result.task_id, + worker_id: @worker_id, + workflow_instance_id: task_result.workflow_instance_id, + cause: error, + retry_count: RETRY_BACKOFFS.size, + task_result: task_result, + duration_ms: duration_ms + )) + end + + def publish_active_workers(executor) + @event_dispatcher.publish(Events::ActiveWorkersChanged.new( + task_type: @worker.task_definition_name, + count: executor.running_count + )) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + + def publish_uncaught_exception(error) + @event_dispatcher.publish(Events::ThreadUncaughtException.new( + cause: error, + task_type: @worker&.task_definition_name + )) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + def cleanup @executor&.shutdown @event_dispatcher.clear diff --git a/lib/conductor/worker/ractor_task_runner.rb b/lib/conductor/worker/ractor_task_runner.rb index cd5978b..e129a3b 100644 --- a/lib/conductor/worker/ractor_task_runner.rb +++ b/lib/conductor/worker/ractor_task_runner.rb @@ -183,7 +183,10 @@ def calculate_adaptive_backoff # Poll for a single task # @return [Hash, nil] Task data or nil def poll_task - return nil if @worker.paused + if @worker.paused + publish_event(Events::TaskPaused.new(task_type: @worker.task_definition_name)) + return nil + end # Auth failure backoff if @auth_failures.positive? && @last_auth_failure_time @@ -401,29 +404,57 @@ def update_task_with_retry(task_result) RETRY_BACKOFFS.each_with_index do |backoff, attempt| sleep(backoff) if backoff.positive? + start_time = Time.now begin @task_client.update_task(task_result) + duration_ms = (Time.now - start_time) * 1000 + + publish_task_update_completed(task_result, duration_ms) return rescue StandardError => e + duration_ms = (Time.now - start_time) * 1000 @logger.error("[Ractor #{@ractor_id}] Update failed (attempt #{attempt + 1}): #{e.message}") if attempt == RETRY_BACKOFFS.size - 1 @logger.fatal("[Ractor #{@ractor_id}] CRITICAL: Task #{task_result.task_id} result LOST") - - publish_event(Events::TaskUpdateFailure.new( - task_type: @worker.task_definition_name, - task_id: task_result.task_id, - worker_id: @worker_id, - workflow_instance_id: task_result.workflow_instance_id, - cause: e, - retry_count: RETRY_BACKOFFS.size, - task_result: task_result - )) + publish_task_update_failure(task_result, e, duration_ms) end end end end + def publish_task_update_completed(task_result, duration_ms) + publish_event(Events::TaskUpdateCompleted.new( + task_type: @worker.task_definition_name, + task_id: task_result.task_id, + worker_id: @worker_id, + workflow_instance_id: task_result.workflow_instance_id, + duration_ms: duration_ms + )) + end + + def publish_task_update_failure(task_result, error, duration_ms) + publish_event(Events::TaskUpdateFailure.new( + task_type: @worker.task_definition_name, + task_id: task_result.task_id, + worker_id: @worker_id, + workflow_instance_id: task_result.workflow_instance_id, + cause: error, + retry_count: RETRY_BACKOFFS.size, + task_result: task_result, + duration_ms: duration_ms + )) + end + + def publish_uncaught_exception(error) + publish_event(Events::ThreadUncaughtException.new( + cause: error, + task_type: @worker&.task_definition_name + )) + rescue StandardError => e + @logger&.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + # Publish event - sends to main Ractor if configured, otherwise logs # @param event [ConductorEvent] Event to publish def publish_event(event) diff --git a/lib/conductor/worker/task_handler.rb b/lib/conductor/worker/task_handler.rb index 8b42121..eef1986 100644 --- a/lib/conductor/worker/task_handler.rb +++ b/lib/conductor/worker/task_handler.rb @@ -48,6 +48,7 @@ def initialize( @running = false @mutex = Mutex.new @register_task_definitions = register_task_definitions + @event_listeners = [] # Register event listeners register_listeners(event_listeners) if event_listeners @@ -260,6 +261,8 @@ def stop(timeout: 5) @runners.clear @threads.clear @ractors.clear + + stop_event_listeners @running = false @logger.info('TaskHandler stopped') @@ -316,10 +319,19 @@ def create_default_logger # @param listeners [Array] Listeners to register def register_listeners(listeners) listeners.each do |listener| + @event_listeners << listener Events::ListenerRegistry.register_task_runner_listener(listener, @event_dispatcher) end end + def stop_event_listeners + @event_listeners.each do |listener| + listener.stop if listener.respond_to?(:stop) + rescue StandardError => e + @logger.debug { "Error stopping listener: #{e.class}: #{e.message}" } + end + end + # Import worker modules from file paths # @param modules [Array] File paths or module names to require def import_worker_modules(modules) diff --git a/lib/conductor/worker/task_runner.rb b/lib/conductor/worker/task_runner.rb index 2dd8172..32fe0c5 100644 --- a/lib/conductor/worker/task_runner.rb +++ b/lib/conductor/worker/task_runner.rb @@ -180,9 +180,14 @@ def apply_resolved_config(config) # Cleanup completed task futures def cleanup_completed_tasks + removed = false @running_tasks.each do |future| - @running_tasks.delete(future) if future.fulfilled? || future.rejected? + if future.fulfilled? || future.rejected? + @running_tasks.delete(future) + removed = true + end end + publish_active_workers if removed end # Calculate adaptive backoff for empty polls @@ -196,8 +201,10 @@ def calculate_adaptive_backoff # @param count [Integer] Number of tasks to poll for # @return [Array] Array of task hashes def batch_poll(count) - # Skip if worker is paused - return [] if @worker.paused + if @worker.paused + @event_dispatcher.publish(Events::TaskPaused.new(task_type: @worker.task_definition_name)) + return [] + end # Auth failure exponential backoff if @auth_failures.value.positive? && @last_auth_failure_time @@ -292,6 +299,7 @@ def submit_task(task) execute_and_update(task) end @running_tasks << future + publish_active_workers end # Execute a task and update the result @@ -448,31 +456,67 @@ def update_task_with_retry(task_result) RETRY_BACKOFFS.each_with_index do |backoff, attempt| sleep(backoff) if backoff.positive? + start_time = Time.now begin @task_client.update_task(task_result) + duration_ms = (Time.now - start_time) * 1000 + + publish_task_update_completed(task_result, duration_ms) return # Success rescue StandardError => e + duration_ms = (Time.now - start_time) * 1000 @logger.error("Task update failed (attempt #{attempt + 1}/#{RETRY_BACKOFFS.size}): #{e.message}") if attempt == RETRY_BACKOFFS.size - 1 - # All retries exhausted - CRITICAL: task result is lost @logger.fatal("CRITICAL: Task update failed after #{RETRY_BACKOFFS.size} attempts. " \ "Task #{task_result.task_id} result is LOST.") - - @event_dispatcher.publish(Events::TaskUpdateFailure.new( - task_type: @worker.task_definition_name, - task_id: task_result.task_id, - worker_id: @worker_id, - workflow_instance_id: task_result.workflow_instance_id, - cause: e, - retry_count: RETRY_BACKOFFS.size, - task_result: task_result - )) + publish_task_update_failure(task_result, e, duration_ms) end end end end + def publish_task_update_completed(task_result, duration_ms) + @event_dispatcher.publish(Events::TaskUpdateCompleted.new( + task_type: @worker.task_definition_name, + task_id: task_result.task_id, + worker_id: @worker_id, + workflow_instance_id: task_result.workflow_instance_id, + duration_ms: duration_ms + )) + end + + def publish_task_update_failure(task_result, error, duration_ms) + @event_dispatcher.publish(Events::TaskUpdateFailure.new( + task_type: @worker.task_definition_name, + task_id: task_result.task_id, + worker_id: @worker_id, + workflow_instance_id: task_result.workflow_instance_id, + cause: error, + retry_count: RETRY_BACKOFFS.size, + task_result: task_result, + duration_ms: duration_ms + )) + end + + def publish_active_workers + @event_dispatcher.publish(Events::ActiveWorkersChanged.new( + task_type: @worker.task_definition_name, + count: @running_tasks.size + )) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + + def publish_uncaught_exception(error) + @event_dispatcher.publish(Events::ThreadUncaughtException.new( + cause: error, + task_type: @worker&.task_definition_name + )) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + # Register task definition if configured def register_task_definition @logger.info('Task definition registration not yet implemented') diff --git a/lib/conductor/worker/telemetry/canonical_metrics_collector.rb b/lib/conductor/worker/telemetry/canonical_metrics_collector.rb new file mode 100644 index 0000000..948d6a0 --- /dev/null +++ b/lib/conductor/worker/telemetry/canonical_metrics_collector.rb @@ -0,0 +1,182 @@ +# frozen_string_literal: true + +require 'logger' +require_relative '../events/listeners' +require_relative '../events/global_dispatcher' + +module Conductor + module Worker + module Telemetry + # CanonicalMetricsCollector - Canonical SDK worker metrics from the + # harmonization spec (sdk-metrics-harmonization.md). + # + # Selected when WORKER_CANONICAL_METRICS is truthy. Uses camelCase domain + # labels (taskType, workflowType) and includes status labels on time + # histograms. + # + # Legacy-only event handlers that have no canonical equivalent are + # implemented as no-ops so this collector satisfies the full listener + # interface and can be used interchangeably with LegacyMetricsCollector. + class CanonicalMetricsCollector + include Events::TaskRunnerEventsListener + include Events::WorkflowEventsListener + include Events::HttpEventsListener + + STATUS_SUCCESS = 'SUCCESS' + STATUS_FAILURE = 'FAILURE' + + # @param backend [Symbol, Object] :null, :prometheus, or a custom backend + # @param subscribe_global_http [Boolean] Auto-subscribe to GlobalDispatcher + # for HttpApiRequest events from the HTTP layer (default true). + # @param measure_payload_size [Boolean] Record workflow_input_size_bytes + # (requires JSON serialization; default true). Set false to skip + # serialization overhead for large payloads. + def initialize(backend: :null, subscribe_global_http: true, measure_payload_size: true, logger: nil) + @backend = load_backend(backend) + @logger = logger || Logger.new(File::NULL) + @measure_payload_size = measure_payload_size + @http_listener = nil + subscribe_to_global_http_events if subscribe_global_http + end + + attr_reader :backend, :measure_payload_size + + def stop + return unless @http_listener + + Events::GlobalDispatcher.instance.unregister(Events::HttpApiRequest, @http_listener) + @http_listener = nil + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + + def collector_name + 'canonical' + end + + # --- Task Runner Event Handlers --- + + def on_poll_started(event) + @backend.increment('task_poll_total', labels: { taskType: event.task_type }) + end + + def on_poll_completed(event) + observe_time('task_poll_time_seconds', event.duration_ms, + { taskType: event.task_type, status: STATUS_SUCCESS }) + end + + def on_poll_failure(event) + @backend.increment('task_poll_error_total', + labels: { taskType: event.task_type, exception: event.cause.class.name }) + observe_time('task_poll_time_seconds', event.duration_ms, + { taskType: event.task_type, status: STATUS_FAILURE }) + end + + def on_task_execution_started(event) + @backend.increment('task_execution_started_total', labels: { taskType: event.task_type }) + end + + def on_task_execution_completed(event) + observe_time('task_execute_time_seconds', event.duration_ms, + { taskType: event.task_type, status: STATUS_SUCCESS }) + + return unless event.output_size_bytes + + @backend.observe('task_result_size_bytes', event.output_size_bytes, + labels: { taskType: event.task_type }) + end + + def on_task_execution_failure(event) + @backend.increment('task_execute_error_total', + labels: { taskType: event.task_type, exception: event.cause.class.name }) + observe_time('task_execute_time_seconds', event.duration_ms, + { taskType: event.task_type, status: STATUS_FAILURE }) + end + + def on_task_update_completed(event) + observe_time('task_update_time_seconds', event.duration_ms, + { taskType: event.task_type, status: STATUS_SUCCESS }) + end + + def on_task_update_failure(event) + @backend.increment('task_update_error_total', + labels: { taskType: event.task_type, exception: event.cause.class.name }) + + return unless event.respond_to?(:duration_ms) && event.duration_ms + + observe_time('task_update_time_seconds', event.duration_ms, + { taskType: event.task_type, status: STATUS_FAILURE }) + end + + def on_task_paused(event) + @backend.increment('task_paused_total', labels: { taskType: event.task_type }) + end + + def on_thread_uncaught_exception(event) + @backend.increment('thread_uncaught_exceptions_total', + labels: { exception: event.cause.class.name }) + end + + def on_active_workers_changed(event) + @backend.set('active_workers', event.count, labels: { taskType: event.task_type }) + end + + # --- Workflow Event Handlers --- + + def on_workflow_start_error(event) + @backend.increment('workflow_start_error_total', + labels: { workflowType: event.workflow_type, + exception: event.cause.class.name }) + end + + def on_workflow_input_size(event) + return unless @measure_payload_size + + @backend.observe('workflow_input_size_bytes', event.size_bytes, + labels: { workflowType: event.workflow_type, + version: (event.version || '').to_s }) + end + + # --- HTTP Event Handlers --- + + def on_http_api_request(event) + observe_time('http_api_client_request_seconds', event.duration_ms, + { method: event.method, uri: event.uri, status: event.status }) + end + + private + + def observe_time(name, duration_ms, labels) + @backend.observe(name, duration_ms / 1000.0, labels: labels) + end + + def subscribe_to_global_http_events + @http_listener = ->(event) { on_http_api_request(event) } + Events::GlobalDispatcher.instance.register(Events::HttpApiRequest, @http_listener) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + + def load_backend(backend) + case backend + when :null, nil + NullBackend.new + when :prometheus + load_prometheus_backend + else + backend + end + end + + def load_prometheus_backend + require_relative 'canonical_prometheus_backend' + CanonicalPrometheusBackend.new + rescue LoadError + raise ConfigurationError, + "The 'prometheus-client' gem is required for Prometheus metrics. " \ + "Add `gem 'prometheus-client'` to your Gemfile." + end + end + end + end +end diff --git a/lib/conductor/worker/telemetry/canonical_prometheus_backend.rb b/lib/conductor/worker/telemetry/canonical_prometheus_backend.rb new file mode 100644 index 0000000..a0401ef --- /dev/null +++ b/lib/conductor/worker/telemetry/canonical_prometheus_backend.rb @@ -0,0 +1,173 @@ +# frozen_string_literal: true + +module Conductor + module Worker + module Telemetry + # CanonicalPrometheusBackend - Prometheus backend for the canonical SDK metric catalog. + # + # Pre-registers every metric from the harmonization spec with its canonical + # label set and bucket configuration. Uses camelCase domain labels (taskType, + # workflowType) per the canonical convention. + class CanonicalPrometheusBackend + TIME_BUCKETS = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10].freeze + SIZE_BUCKETS = [100, 1000, 10_000, 100_000, 1_000_000, 10_000_000].freeze + + COUNTER_LABELS = { + 'task_poll_total' => %i[taskType], + 'task_execution_started_total' => %i[taskType], + 'task_poll_error_total' => %i[taskType exception], + 'task_execute_error_total' => %i[taskType exception], + 'task_update_error_total' => %i[taskType exception], + 'task_paused_total' => %i[taskType], + 'thread_uncaught_exceptions_total' => %i[exception], + 'workflow_start_error_total' => %i[workflowType exception] + }.freeze + + HISTOGRAM_LABELS = { + 'task_poll_time_seconds' => %i[taskType status], + 'task_execute_time_seconds' => %i[taskType status], + 'task_update_time_seconds' => %i[taskType status], + 'http_api_client_request_seconds' => %i[method uri status], + 'task_result_size_bytes' => %i[taskType], + 'workflow_input_size_bytes' => %i[workflowType version] + }.freeze + + GAUGE_LABELS = { + 'active_workers' => %i[taskType] + }.freeze + + HISTOGRAM_BUCKETS = { + 'task_result_size_bytes' => SIZE_BUCKETS, + 'workflow_input_size_bytes' => SIZE_BUCKETS + }.freeze + + def initialize(registry: nil) + load_prometheus_client + @registry = registry || Prometheus::Client.registry + @counters = {} + @histograms = {} + @gauges = {} + setup_metrics + end + + def increment(name, labels: {}, value: 1) + metric = get_or_create_counter(name) + metric.increment(labels: normalize_labels(name, labels, COUNTER_LABELS), by: value) + end + + def observe(name, value, labels: {}) + metric = get_or_create_histogram(name) + metric.observe(value, labels: normalize_labels(name, labels, HISTOGRAM_LABELS)) + end + + def set(name, value, labels: {}) + metric = get_or_create_gauge(name) + metric.set(value, labels: normalize_labels(name, labels, GAUGE_LABELS)) + end + + attr_reader :registry + + private + + def load_prometheus_client + require 'prometheus/client' + rescue LoadError + raise ConfigurationError, + "The 'prometheus-client' gem is required for Prometheus metrics. " \ + "Add `gem 'prometheus-client'` to your Gemfile." + end + + def setup_metrics + COUNTER_LABELS.each do |name, _| + register_counter(name, "Counter for #{name}") + end + + HISTOGRAM_LABELS.each do |name, _| + register_histogram(name, "Histogram for #{name}") + end + + GAUGE_LABELS.each do |name, _| + register_gauge(name, "Gauge for #{name}") + end + end + + def register_counter(name, docstring) + metric_name = name.to_sym + labels = COUNTER_LABELS.fetch(name, %i[taskType]) + @counters[name] = register_or_reuse(metric_name) do + Prometheus::Client::Counter.new(metric_name, docstring: docstring, labels: labels) + end + end + + def register_histogram(name, docstring) + metric_name = name.to_sym + labels = HISTOGRAM_LABELS.fetch(name, %i[taskType]) + buckets = HISTOGRAM_BUCKETS[name] || TIME_BUCKETS + @histograms[name] = register_or_reuse(metric_name) do + Prometheus::Client::Histogram.new(metric_name, docstring: docstring, + labels: labels, buckets: buckets) + end + end + + def register_gauge(name, docstring) + metric_name = name.to_sym + labels = GAUGE_LABELS.fetch(name, %i[taskType]) + @gauges[name] = register_or_reuse(metric_name) do + Prometheus::Client::Gauge.new(metric_name, docstring: docstring, labels: labels) + end + end + + def register_or_reuse(metric_name) + if @registry.exist?(metric_name) + @registry.get(metric_name) + else + metric = yield + @registry.register(metric) + metric + end + end + + def get_or_create_counter(name) + @counters[name] ||= register_or_reuse(name.to_sym) do + labels = COUNTER_LABELS.fetch(name, %i[taskType]) + Prometheus::Client::Counter.new(name.to_sym, docstring: "Counter for #{name}", labels: labels) + end + end + + def get_or_create_histogram(name) + @histograms[name] ||= register_or_reuse(name.to_sym) do + labels = HISTOGRAM_LABELS.fetch(name, %i[taskType]) + buckets = HISTOGRAM_BUCKETS[name] || TIME_BUCKETS + Prometheus::Client::Histogram.new(name.to_sym, docstring: "Histogram for #{name}", + labels: labels, buckets: buckets) + end + end + + def get_or_create_gauge(name) + @gauges[name] ||= register_or_reuse(name.to_sym) do + labels = GAUGE_LABELS.fetch(name, %i[taskType]) + Prometheus::Client::Gauge.new(name.to_sym, docstring: "Gauge for #{name}", labels: labels) + end + end + + # Align provided labels to the declared label set for the metric. + # Missing keys get empty-string defaults; unknown keys are dropped. + def normalize_labels(name, labels, schema) + symbolized = {} + labels.each do |key, value| + next if value.nil? + + symbolized[key.to_sym] = value.to_s + end + + declared = schema[name] + return symbolized unless declared + + declared.each_with_object({}) do |key, acc| + acc[key] = symbolized.key?(key) ? symbolized[key] : '' + end + end + end + end + end +end diff --git a/lib/conductor/worker/telemetry/legacy_metrics_collector.rb b/lib/conductor/worker/telemetry/legacy_metrics_collector.rb new file mode 100644 index 0000000..b7714c5 --- /dev/null +++ b/lib/conductor/worker/telemetry/legacy_metrics_collector.rb @@ -0,0 +1,116 @@ +# frozen_string_literal: true + +require_relative '../events/listeners' + +module Conductor + module Worker + module Telemetry + # LegacyMetricsCollector - The original Ruby SDK metrics implementation. + # + # Emits the pre-harmonization metric set with snake_case labels (task_type, error). + # This is the default implementation during the deprecation period while + # WORKER_CANONICAL_METRICS defaults to false. + # + # Canonical-only event handlers (on_task_update_completed, on_task_paused, etc.) + # are implemented as no-ops so this collector satisfies the full listener interface + # and can be used interchangeably with CanonicalMetricsCollector. + class LegacyMetricsCollector + include Events::TaskRunnerEventsListener + include Events::WorkflowEventsListener + include Events::HttpEventsListener + + def initialize(backend: :null, measure_payload_size: false) + @backend = load_backend(backend) + @measure_payload_size = measure_payload_size + end + + attr_reader :backend, :measure_payload_size + + def collector_name + 'legacy' + end + + def stop; end + + # --- Real legacy metrics --- + + def on_poll_started(event) + @backend.increment('task_poll_total', labels: { task_type: event.task_type }) + end + + def on_poll_completed(event) + @backend.observe('task_poll_time_seconds', event.duration_ms / 1000.0, + labels: { task_type: event.task_type }) + end + + def on_poll_failure(event) + @backend.increment('task_poll_error_total', + labels: { + task_type: event.task_type, + error: event.cause.class.name + }) + end + + def on_task_execution_started(_event) + # No legacy metric for execution-started + end + + def on_task_execution_completed(event) + @backend.observe('task_execute_time_seconds', event.duration_ms / 1000.0, + labels: { task_type: event.task_type }) + + return unless event.output_size_bytes + + @backend.observe('task_result_size_bytes', event.output_size_bytes, + labels: { task_type: event.task_type }) + end + + def on_task_execution_failure(event) + @backend.increment('task_execute_error_total', + labels: { + task_type: event.task_type, + exception: event.cause.class.name, + retryable: event.is_retryable.to_s + }) + end + + def on_task_update_failure(event) + @backend.increment('task_update_failed_total', + labels: { task_type: event.task_type }) + end + + # --- No-op stubs for canonical-only events --- + + def on_task_update_completed(_event); end + def on_task_paused(_event); end + def on_thread_uncaught_exception(_event); end + def on_active_workers_changed(_event); end + def on_workflow_start_error(_event); end + def on_workflow_input_size(_event); end + def on_http_api_request(_event); end + + private + + def load_backend(backend) + case backend + when :null, nil + NullBackend.new + when :prometheus + load_prometheus_backend + else + backend + end + end + + def load_prometheus_backend + require_relative 'prometheus_backend' + PrometheusBackend.new + rescue LoadError + raise ConfigurationError, + "The 'prometheus-client' gem is required for Prometheus metrics. " \ + "Add `gem 'prometheus-client'` to your Gemfile." + end + end + end + end +end diff --git a/lib/conductor/worker/telemetry/metrics_collector.rb b/lib/conductor/worker/telemetry/metrics_collector.rb index 6e7037e..cf9dbc9 100644 --- a/lib/conductor/worker/telemetry/metrics_collector.rb +++ b/lib/conductor/worker/telemetry/metrics_collector.rb @@ -1,123 +1,66 @@ # frozen_string_literal: true -require_relative '../events/listeners' +require_relative 'legacy_metrics_collector' +require_relative 'canonical_metrics_collector' module Conductor module Worker module Telemetry - # MetricsCollector - Collects metrics from worker events - # Implements TaskRunnerEventsListener protocol - # Uses pluggable backends (null, prometheus, etc.) - class MetricsCollector - include Events::TaskRunnerEventsListener - - # Initialize metrics collector - # @param backend [Symbol, Object] Backend type (:null, :prometheus) or custom backend - def initialize(backend: :null) - @backend = load_backend(backend) - end - - # @return [Object] The metrics backend - attr_reader :backend - - # --- Event Handlers --- - - def on_poll_started(event) - @backend.increment('task_poll_total', labels: { task_type: event.task_type }) - end - - def on_poll_completed(event) - @backend.observe('task_poll_time_seconds', event.duration_ms / 1000.0, - labels: { task_type: event.task_type }) - end - - def on_poll_failure(event) - @backend.increment('task_poll_error_total', - labels: { - task_type: event.task_type, - error: event.cause.class.name - }) - end - - def on_task_execution_started(event) - # Could track active tasks here - end - - def on_task_execution_completed(event) - @backend.observe('task_execute_time_seconds', event.duration_ms / 1000.0, - labels: { task_type: event.task_type }) - - return unless event.output_size_bytes - - @backend.observe('task_result_size_bytes', event.output_size_bytes, - labels: { task_type: event.task_type }) + # MetricsCollector - Factory for creating the appropriate metrics + # collector based on environment configuration. + # + # Currently checks WORKER_CANONICAL_METRICS (default false). When truthy, + # returns a CanonicalMetricsCollector; otherwise a LegacyMetricsCollector. + # + # In a future release, when canonical metrics become the default, + # WORKER_LEGACY_METRICS will be checked to allow opting back in to the + # legacy implementation. + module MetricsCollector + # Backward-compatible shim: delegates to .create so existing + # MetricsCollector.new(backend: ...) callers don't crash on upgrade. + def self.new(backend: :null, **_opts) + warn '[DEPRECATION] MetricsCollector.new is deprecated. Use MetricsCollector.create instead.' + create(backend: backend) end - def on_task_execution_failure(event) - @backend.increment('task_execute_error_total', - labels: { - task_type: event.task_type, - exception: event.cause.class.name, - retryable: event.is_retryable.to_s - }) - end - - def on_task_update_failure(event) - @backend.increment('task_update_failed_total', - labels: { task_type: event.task_type }) - end - - private - - # Load a metrics backend - # @param backend [Symbol, Object] Backend type or instance - # @return [Object] Backend instance - def load_backend(backend) - case backend - when :null, nil - NullBackend.new - when :prometheus - load_prometheus_backend + # Create a metrics collector instance gated by environment configuration. + # + # @param backend [Symbol, Object] Backend type (:null, :prometheus) or custom backend + # @param subscribe_global_http [Boolean] Auto-subscribe to HTTP events (canonical only) + # @param measure_payload_size [Boolean, nil] Record workflow_input_size_bytes. + # Defaults to true for canonical, false for legacy. Set explicitly to override. + # @param logger [Logger, nil] Optional logger for diagnostic output in rescue blocks + # @return [LegacyMetricsCollector, CanonicalMetricsCollector] + def self.create(backend: :null, subscribe_global_http: true, measure_payload_size: nil, logger: nil) + if canonical_metrics_enabled? + mps = measure_payload_size.nil? ? true : measure_payload_size + CanonicalMetricsCollector.new( + backend: backend, subscribe_global_http: subscribe_global_http, + measure_payload_size: mps, logger: logger + ) else - # Assume it's a custom backend instance - backend + mps = measure_payload_size.nil? ? false : measure_payload_size + LegacyMetricsCollector.new(backend: backend, measure_payload_size: mps) end end - # Load Prometheus backend (lazy loading) - # @return [PrometheusBackend] - def load_prometheus_backend - require_relative 'prometheus_backend' - PrometheusBackend.new - rescue LoadError - raise ConfigurationError, - "The 'prometheus-client' gem is required for Prometheus metrics. " \ - "Add `gem 'prometheus-client'` to your Gemfile." + # @return [Boolean] true when the canonical metric set is selected + def self.canonical_metrics_enabled? + %w[true 1 yes].include?(ENV.fetch('WORKER_CANONICAL_METRICS', 'false').downcase.strip) end end # NullBackend - No-op backend for metrics # Used when metrics are disabled or not configured class NullBackend - # Increment a counter (no-op) - # @param name [String] Metric name - # @param labels [Hash] Metric labels def increment(name, labels: {}) # No-op end - # Observe a value (no-op) - # @param name [String] Metric name - # @param value [Numeric] Value to observe - # @param labels [Hash] Metric labels def observe(name, value, labels: {}) # No-op end - # Set a gauge value (no-op) - # @param name [String] Metric name - # @param value [Numeric] Value to set - # @param labels [Hash] Metric labels def set(name, value, labels: {}) # No-op end diff --git a/lib/conductor/worker/telemetry/prometheus_backend.rb b/lib/conductor/worker/telemetry/prometheus_backend.rb index 31cdbd7..3c8bc75 100644 --- a/lib/conductor/worker/telemetry/prometheus_backend.rb +++ b/lib/conductor/worker/telemetry/prometheus_backend.rb @@ -16,7 +16,7 @@ module Telemetry # - task_update_failed_total (Counter) # # @example - # collector = MetricsCollector.new(backend: :prometheus) + # collector = MetricsCollector.create(backend: :prometheus) # # Metrics available at default prometheus registry class PrometheusBackend # Default histogram buckets for time measurements (in seconds) diff --git a/lib/conductor/workflow/workflow_executor.rb b/lib/conductor/workflow/workflow_executor.rb index f6eaeda..cef11ea 100644 --- a/lib/conductor/workflow/workflow_executor.rb +++ b/lib/conductor/workflow/workflow_executor.rb @@ -1,5 +1,7 @@ # frozen_string_literal: true +require 'json' +require 'logger' require 'securerandom' require_relative '../configuration' require_relative '../http/api_client' @@ -7,22 +9,27 @@ require_relative '../http/api/metadata_resource_api' require_relative '../http/api/task_resource_api' require_relative '../http/models/start_workflow_request' +require_relative '../worker/events/workflow_events' +require_relative '../worker/events/sync_event_dispatcher' module Conductor module Workflow # WorkflowExecutor provides a high-level interface for executing workflows # Supports both synchronous (wait for completion) and asynchronous execution class WorkflowExecutor - attr_reader :workflow_api, :metadata_api, :task_api + attr_reader :workflow_api, :metadata_api, :task_api, :event_dispatcher # Initialize WorkflowExecutor # @param [Configuration] configuration Optional configuration - def initialize(configuration = nil) + # @param [Worker::Events::SyncEventDispatcher, nil] event_dispatcher Optional event dispatcher + def initialize(configuration = nil, event_dispatcher: nil, logger: nil) @configuration = configuration || Configuration.new + @logger = logger || Logger.new(File::NULL) api_client = Http::ApiClient.new(configuration: @configuration) @workflow_api = Http::Api::WorkflowResourceApi.new(api_client) @metadata_api = Http::Api::MetadataResourceApi.new(api_client) @task_api = Http::Api::TaskResourceApi.new(api_client) + @event_dispatcher = event_dispatcher || Worker::Events::SyncEventDispatcher.new end # ========================================== @@ -46,7 +53,11 @@ def register_workflow(workflow, overwrite: true) # @param [StartWorkflowRequest] request Start workflow request # @return [String] Workflow ID def start_workflow(request) + publish_workflow_input_size(request) @workflow_api.start_workflow(request) + rescue StandardError => e + publish_workflow_start_error(request, e) + raise end # Start multiple workflows @@ -321,6 +332,42 @@ def execute_and_wait(name, input: {}, timeout_seconds: 60, **options) get_workflow(result.workflow_id) end end + + private + + def publish_workflow_input_size(request) + return unless @event_dispatcher.has_listeners?(Worker::Events::WorkflowInputSize) + + name = request.respond_to?(:name) ? request.name : nil + return unless name + + input_bytes = begin + (request.respond_to?(:input) ? request.input : nil).to_json.bytesize + rescue StandardError + 0 + end + + @event_dispatcher.publish(Worker::Events::WorkflowInputSize.new( + workflow_type: name, + version: request.respond_to?(:version) ? request.version : nil, + size_bytes: input_bytes + )) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end + + def publish_workflow_start_error(request, error) + wf_name = (request.respond_to?(:name) ? request.name : nil) || 'unknown' + wf_version = request.respond_to?(:version) ? request.version : nil + + @event_dispatcher.publish(Worker::Events::WorkflowStartError.new( + workflow_type: wf_name, + version: wf_version, + cause: error + )) + rescue StandardError => e + @logger.debug { "Telemetry error (non-fatal): #{e.class}: #{e.message}" } + end end end end diff --git a/spec/conductor/worker/canonical_metrics_collector_spec.rb b/spec/conductor/worker/canonical_metrics_collector_spec.rb new file mode 100644 index 0000000..8ba0ef2 --- /dev/null +++ b/spec/conductor/worker/canonical_metrics_collector_spec.rb @@ -0,0 +1,271 @@ +# frozen_string_literal: true + +require 'spec_helper' +require_relative '../../../lib/conductor/worker/telemetry/metrics_collector' + +RSpec.describe Conductor::Worker::Telemetry::CanonicalMetricsCollector do + let(:backend) { double('backend') } + let(:collector) { described_class.new(backend: backend, subscribe_global_http: false) } + + before do + allow(backend).to receive(:increment) + allow(backend).to receive(:observe) + allow(backend).to receive(:set) + end + + describe '#initialize' do + it 'uses NullBackend by default' do + c = described_class.new(subscribe_global_http: false) + expect(c.backend).to be_a(Conductor::Worker::Telemetry::NullBackend) + end + end + + describe '#collector_name' do + it 'returns "canonical"' do + expect(collector.collector_name).to eq('canonical') + end + end + + # --- Task Runner Events --- + + describe '#on_poll_started' do + it 'increments task_poll_total with camelCase taskType label' do + event = Conductor::Worker::Events::PollStarted.new( + task_type: 'my_task', worker_id: 'w1', poll_count: 1 + ) + collector.on_poll_started(event) + expect(backend).to have_received(:increment).with( + 'task_poll_total', labels: { taskType: 'my_task' } + ) + end + end + + describe '#on_poll_completed' do + it 'observes task_poll_time_seconds with status=SUCCESS' do + event = Conductor::Worker::Events::PollCompleted.new( + task_type: 'my_task', duration_ms: 250.0, tasks_received: 2 + ) + collector.on_poll_completed(event) + expect(backend).to have_received(:observe).with( + 'task_poll_time_seconds', 0.25, + labels: { taskType: 'my_task', status: 'SUCCESS' } + ) + end + end + + describe '#on_poll_failure' do + it 'increments task_poll_error_total with exception label' do + error = Timeout::Error.new('timed out') + event = Conductor::Worker::Events::PollFailure.new( + task_type: 'my_task', duration_ms: 100.0, cause: error + ) + collector.on_poll_failure(event) + expect(backend).to have_received(:increment).with( + 'task_poll_error_total', + labels: { taskType: 'my_task', exception: 'Timeout::Error' } + ) + end + + it 'observes task_poll_time_seconds with status=FAILURE' do + error = StandardError.new('nope') + event = Conductor::Worker::Events::PollFailure.new( + task_type: 'my_task', duration_ms: 80.0, cause: error + ) + collector.on_poll_failure(event) + expect(backend).to have_received(:observe).with( + 'task_poll_time_seconds', 0.08, + labels: { taskType: 'my_task', status: 'FAILURE' } + ) + end + end + + describe '#on_task_execution_started' do + it 'increments task_execution_started_total' do + event = Conductor::Worker::Events::TaskExecutionStarted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', workflow_instance_id: 'wf1' + ) + collector.on_task_execution_started(event) + expect(backend).to have_received(:increment).with( + 'task_execution_started_total', labels: { taskType: 'my_task' } + ) + end + end + + describe '#on_task_execution_completed' do + it 'observes task_execute_time_seconds with status=SUCCESS' do + event = Conductor::Worker::Events::TaskExecutionCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 1200.0, output_size_bytes: 4096 + ) + collector.on_task_execution_completed(event) + expect(backend).to have_received(:observe).with( + 'task_execute_time_seconds', 1.2, + labels: { taskType: 'my_task', status: 'SUCCESS' } + ) + end + + it 'observes task_result_size_bytes as histogram with taskType label' do + event = Conductor::Worker::Events::TaskExecutionCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 500.0, output_size_bytes: 4096 + ) + collector.on_task_execution_completed(event) + expect(backend).to have_received(:observe).with( + 'task_result_size_bytes', 4096, labels: { taskType: 'my_task' } + ) + end + + it 'skips task_result_size_bytes when output_size_bytes is nil' do + event = Conductor::Worker::Events::TaskExecutionCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 500.0 + ) + collector.on_task_execution_completed(event) + expect(backend).to have_received(:observe).once + end + end + + describe '#on_task_execution_failure' do + it 'increments task_execute_error_total and observes time with FAILURE' do + error = ArgumentError.new('bad input') + event = Conductor::Worker::Events::TaskExecutionFailure.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 300.0, cause: error, is_retryable: true + ) + collector.on_task_execution_failure(event) + + expect(backend).to have_received(:increment).with( + 'task_execute_error_total', + labels: { taskType: 'my_task', exception: 'ArgumentError' } + ) + expect(backend).to have_received(:observe).with( + 'task_execute_time_seconds', 0.3, + labels: { taskType: 'my_task', status: 'FAILURE' } + ) + end + end + + describe '#on_task_update_completed' do + it 'observes task_update_time_seconds with status=SUCCESS' do + event = Conductor::Worker::Events::TaskUpdateCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 75.0 + ) + collector.on_task_update_completed(event) + expect(backend).to have_received(:observe).with( + 'task_update_time_seconds', 0.075, + labels: { taskType: 'my_task', status: 'SUCCESS' } + ) + end + end + + describe '#on_task_update_failure' do + it 'increments task_update_error_total and observes time with FAILURE' do + error = StandardError.new('net err') + task_result = Conductor::Http::Models::TaskResult.complete + event = Conductor::Worker::Events::TaskUpdateFailure.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', cause: error, retry_count: 4, + task_result: task_result, duration_ms: 120.0 + ) + collector.on_task_update_failure(event) + + expect(backend).to have_received(:increment).with( + 'task_update_error_total', + labels: { taskType: 'my_task', exception: 'StandardError' } + ) + expect(backend).to have_received(:observe).with( + 'task_update_time_seconds', 0.12, + labels: { taskType: 'my_task', status: 'FAILURE' } + ) + end + + it 'skips time observation when duration_ms is nil' do + error = StandardError.new('err') + task_result = Conductor::Http::Models::TaskResult.complete + event = Conductor::Worker::Events::TaskUpdateFailure.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', cause: error, retry_count: 4, task_result: task_result + ) + collector.on_task_update_failure(event) + expect(backend).not_to have_received(:observe) + end + end + + describe '#on_task_paused' do + it 'increments task_paused_total' do + event = Conductor::Worker::Events::TaskPaused.new(task_type: 'my_task') + collector.on_task_paused(event) + expect(backend).to have_received(:increment).with( + 'task_paused_total', labels: { taskType: 'my_task' } + ) + end + end + + describe '#on_thread_uncaught_exception' do + it 'increments thread_uncaught_exceptions_total with exception label' do + event = Conductor::Worker::Events::ThreadUncaughtException.new( + cause: RuntimeError.new('boom') + ) + collector.on_thread_uncaught_exception(event) + expect(backend).to have_received(:increment).with( + 'thread_uncaught_exceptions_total', labels: { exception: 'RuntimeError' } + ) + end + end + + describe '#on_active_workers_changed' do + it 'sets active_workers gauge' do + event = Conductor::Worker::Events::ActiveWorkersChanged.new( + task_type: 'my_task', count: 7 + ) + collector.on_active_workers_changed(event) + expect(backend).to have_received(:set).with( + 'active_workers', 7, labels: { taskType: 'my_task' } + ) + end + end + + # --- Workflow Events --- + + describe '#on_workflow_start_error' do + it 'increments workflow_start_error_total' do + event = Conductor::Worker::Events::WorkflowStartError.new( + workflow_type: 'my_wf', cause: RuntimeError.new('fail') + ) + collector.on_workflow_start_error(event) + expect(backend).to have_received(:increment).with( + 'workflow_start_error_total', + labels: { workflowType: 'my_wf', exception: 'RuntimeError' } + ) + end + end + + describe '#on_workflow_input_size' do + it 'observes workflow_input_size_bytes' do + event = Conductor::Worker::Events::WorkflowInputSize.new( + workflow_type: 'my_wf', size_bytes: 8192, version: 2 + ) + collector.on_workflow_input_size(event) + expect(backend).to have_received(:observe).with( + 'workflow_input_size_bytes', 8192, + labels: { workflowType: 'my_wf', version: '2' } + ) + end + end + + # --- HTTP Events --- + + describe '#on_http_api_request' do + it 'observes http_api_client_request_seconds' do + event = Conductor::Worker::Events::HttpApiRequest.new( + method: 'POST', uri: '/api/tasks/poll/batch/my_task', status: '200', duration_ms: 45.0 + ) + collector.on_http_api_request(event) + expect(backend).to have_received(:observe).with( + 'http_api_client_request_seconds', 0.045, + labels: { method: 'POST', uri: '/api/tasks/poll/batch/my_task', status: '200' } + ) + end + end +end diff --git a/spec/conductor/worker/canonical_prometheus_backend_spec.rb b/spec/conductor/worker/canonical_prometheus_backend_spec.rb new file mode 100644 index 0000000..668d0ac --- /dev/null +++ b/spec/conductor/worker/canonical_prometheus_backend_spec.rb @@ -0,0 +1,104 @@ +# frozen_string_literal: true + +require 'spec_helper' + +CANONICAL_PROMETHEUS_AVAILABLE = begin + require 'prometheus/client' + true +rescue LoadError + false +end + +CANONICAL_BACKEND_LOADED = begin + if CANONICAL_PROMETHEUS_AVAILABLE + require_relative '../../../lib/conductor/worker/telemetry/canonical_prometheus_backend' + true + else + false + end +rescue Conductor::ConfigurationError + false +end + +if CANONICAL_BACKEND_LOADED + RSpec.describe Conductor::Worker::Telemetry::CanonicalPrometheusBackend do + let(:registry) { Prometheus::Client::Registry.new } + let(:backend) { described_class.new(registry: registry) } + + describe '#initialize' do + it 'registers canonical counters' do + backend # force lazy initialization + %i[task_poll_total task_execution_started_total task_poll_error_total + task_execute_error_total task_update_error_total task_paused_total + thread_uncaught_exceptions_total workflow_start_error_total].each do |name| + expect(registry.exist?(name)).to be(true), "Expected counter #{name} to be registered" + end + end + + it 'registers canonical histograms' do + backend + %i[task_poll_time_seconds task_execute_time_seconds task_update_time_seconds + http_api_client_request_seconds task_result_size_bytes + workflow_input_size_bytes].each do |name| + expect(registry.exist?(name)).to be(true), "Expected histogram #{name} to be registered" + end + end + + it 'registers canonical gauges' do + backend + expect(registry.exist?(:active_workers)).to be true + end + end + + describe '#increment' do + it 'increments a counter with camelCase labels' do + expect do + backend.increment('task_poll_total', labels: { taskType: 'my_task' }) + end.not_to raise_error + end + end + + describe '#observe' do + it 'observes a time histogram with status label' do + expect do + backend.observe('task_poll_time_seconds', 0.25, + labels: { taskType: 'my_task', status: 'SUCCESS' }) + end.not_to raise_error + end + + it 'observes a size histogram' do + expect do + backend.observe('task_result_size_bytes', 5000, labels: { taskType: 'my_task' }) + end.not_to raise_error + end + end + + describe '#set' do + it 'sets a gauge value' do + expect do + backend.set('active_workers', 3, labels: { taskType: 'my_task' }) + end.not_to raise_error + end + end + + describe 'label normalization' do + it 'fills missing declared labels with empty strings' do + expect do + backend.increment('task_poll_error_total', labels: { taskType: 'my_task' }) + end.not_to raise_error + end + + it 'drops undeclared labels' do + expect do + backend.increment('task_poll_total', labels: { taskType: 'my_task', extra: 'nope' }) + end.not_to raise_error + end + end + end +else + RSpec.describe 'CanonicalPrometheusBackend (prometheus-client gem unavailable)' do + it 'documents that prometheus-client gem is not installed' do + expect(CANONICAL_PROMETHEUS_AVAILABLE).to be false + end + end +end diff --git a/spec/conductor/worker/fiber_executor_spec.rb b/spec/conductor/worker/fiber_executor_spec.rb index 98b5eeb..ca3fbfd 100644 --- a/spec/conductor/worker/fiber_executor_spec.rb +++ b/spec/conductor/worker/fiber_executor_spec.rb @@ -85,6 +85,103 @@ expect { runner.shutdown }.not_to raise_error end end + + describe 'event publishing helpers' do + let(:event_dispatcher) { Conductor::Worker::Events::SyncEventDispatcher.new } + let(:logger) { Logger.new(File::NULL) } + + let(:runner) do + described_class.new(worker, configuration: configuration, + event_dispatcher: event_dispatcher, logger: logger) + end + + let(:task_result) do + Conductor::Http::Models::TaskResult.new.tap do |r| + r.task_id = 'task-123' + r.workflow_instance_id = 'workflow-456' + end + end + + before do + runner.instance_variable_set(:@worker_id, 'fiber-worker-0') + end + + describe '#publish_task_update_completed' do + it 'publishes a TaskUpdateCompleted event with duration_ms' do + received = [] + event_dispatcher.register(Conductor::Worker::Events::TaskUpdateCompleted, + ->(event) { received << event }) + + runner.send(:publish_task_update_completed, task_result, 55.0) + + expect(received.size).to eq(1) + expect(received.first.task_type).to eq('fiber_task') + expect(received.first.duration_ms).to eq(55.0) + end + end + + describe '#publish_task_update_failure' do + it 'publishes a TaskUpdateFailure event with duration_ms' do + received = [] + event_dispatcher.register(Conductor::Worker::Events::TaskUpdateFailure, + ->(event) { received << event }) + + error = StandardError.new('update failed') + runner.send(:publish_task_update_failure, task_result, error, 88.0) + + expect(received.size).to eq(1) + expect(received.first.cause).to eq(error) + expect(received.first.duration_ms).to eq(88.0) + end + end + + describe '#publish_uncaught_exception' do + it 'publishes a ThreadUncaughtException event' do + received = [] + event_dispatcher.register(Conductor::Worker::Events::ThreadUncaughtException, + ->(event) { received << event }) + + runner.send(:publish_uncaught_exception, RuntimeError.new('boom')) + + expect(received.size).to eq(1) + expect(received.first.cause).to be_a(RuntimeError) + end + + it 'does not raise when dispatch fails' do + broken = Conductor::Worker::Events::SyncEventDispatcher.new + broken.register(Conductor::Worker::Events::ThreadUncaughtException, + ->(_e) { raise 'listener error' }) + r = described_class.new(worker, configuration: configuration, + event_dispatcher: broken, logger: logger) + r.instance_variable_set(:@worker_id, 'fiber-worker-0') + + expect { r.send(:publish_uncaught_exception, RuntimeError.new('test')) }.not_to raise_error + end + end + + describe 'TaskPaused when worker is paused' do + let(:paused_worker) do + Conductor::Worker::Worker.new('fiber_task', poll_interval: 100, paused: true) do |task| + { result: task.input_data['value'] } + end + end + + it 'publishes TaskPaused and returns empty array' do + paused_runner = described_class.new(paused_worker, configuration: configuration, + event_dispatcher: event_dispatcher, + logger: logger) + received = [] + event_dispatcher.register(Conductor::Worker::Events::TaskPaused, + ->(event) { received << event }) + + result = paused_runner.send(:batch_poll, 1) + + expect(result).to eq([]) + expect(received.size).to eq(1) + expect(received.first.task_type).to eq('fiber_task') + end + end + end end else # Test when async is not available diff --git a/spec/conductor/worker/legacy_metrics_collector_spec.rb b/spec/conductor/worker/legacy_metrics_collector_spec.rb new file mode 100644 index 0000000..418fdc9 --- /dev/null +++ b/spec/conductor/worker/legacy_metrics_collector_spec.rb @@ -0,0 +1,192 @@ +# frozen_string_literal: true + +require 'spec_helper' +require_relative '../../../lib/conductor/worker/telemetry/metrics_collector' + +RSpec.describe Conductor::Worker::Telemetry::LegacyMetricsCollector do + let(:backend) { double('backend') } + let(:collector) { described_class.new(backend: backend) } + + before do + allow(backend).to receive(:increment) + allow(backend).to receive(:observe) + allow(backend).to receive(:set) + end + + describe '#initialize' do + it 'uses NullBackend by default' do + collector = described_class.new + expect(collector.backend).to be_a(Conductor::Worker::Telemetry::NullBackend) + end + + it 'accepts a custom backend instance' do + custom = Object.new + collector = described_class.new(backend: custom) + expect(collector.backend).to eq(custom) + end + end + + describe '#collector_name' do + it 'returns "legacy"' do + expect(collector.collector_name).to eq('legacy') + end + end + + describe '#on_poll_started' do + it 'increments task_poll_total with snake_case task_type' do + event = Conductor::Worker::Events::PollStarted.new( + task_type: 'my_task', worker_id: 'w1', poll_count: 1 + ) + collector.on_poll_started(event) + expect(backend).to have_received(:increment).with( + 'task_poll_total', labels: { task_type: 'my_task' } + ) + end + end + + describe '#on_poll_completed' do + it 'observes task_poll_time_seconds without status label' do + event = Conductor::Worker::Events::PollCompleted.new( + task_type: 'my_task', duration_ms: 150.0, tasks_received: 3 + ) + collector.on_poll_completed(event) + expect(backend).to have_received(:observe).with( + 'task_poll_time_seconds', 0.15, labels: { task_type: 'my_task' } + ) + end + end + + describe '#on_poll_failure' do + it 'increments task_poll_error_total with error label' do + error = StandardError.new('timeout') + event = Conductor::Worker::Events::PollFailure.new( + task_type: 'my_task', duration_ms: 100.0, cause: error + ) + collector.on_poll_failure(event) + expect(backend).to have_received(:increment).with( + 'task_poll_error_total', labels: { task_type: 'my_task', error: 'StandardError' } + ) + end + end + + describe '#on_task_execution_started' do + it 'is a no-op (no legacy metric)' do + event = Conductor::Worker::Events::TaskExecutionStarted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', workflow_instance_id: 'wf1' + ) + expect { collector.on_task_execution_started(event) }.not_to raise_error + expect(backend).not_to have_received(:increment) + end + end + + describe '#on_task_execution_completed' do + it 'observes task_execute_time_seconds' do + event = Conductor::Worker::Events::TaskExecutionCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 500.0, output_size_bytes: 2048 + ) + collector.on_task_execution_completed(event) + expect(backend).to have_received(:observe).with( + 'task_execute_time_seconds', 0.5, labels: { task_type: 'my_task' } + ) + end + + it 'observes task_result_size_bytes when output_size_bytes is present' do + event = Conductor::Worker::Events::TaskExecutionCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 500.0, output_size_bytes: 2048 + ) + collector.on_task_execution_completed(event) + expect(backend).to have_received(:observe).with( + 'task_result_size_bytes', 2048, labels: { task_type: 'my_task' } + ) + end + + it 'skips task_result_size_bytes when output_size_bytes is nil' do + event = Conductor::Worker::Events::TaskExecutionCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 500.0, output_size_bytes: nil + ) + collector.on_task_execution_completed(event) + expect(backend).to have_received(:observe).once + end + end + + describe '#on_task_execution_failure' do + it 'increments task_execute_error_total with exception and retryable labels' do + error = ArgumentError.new('bad') + event = Conductor::Worker::Events::TaskExecutionFailure.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 100.0, cause: error, is_retryable: true + ) + collector.on_task_execution_failure(event) + expect(backend).to have_received(:increment).with( + 'task_execute_error_total', + labels: { task_type: 'my_task', exception: 'ArgumentError', retryable: 'true' } + ) + end + end + + describe '#on_task_update_failure' do + it 'increments task_update_failed_total' do + error = StandardError.new('net error') + task_result = Conductor::Http::Models::TaskResult.complete + event = Conductor::Worker::Events::TaskUpdateFailure.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', cause: error, retry_count: 4, task_result: task_result + ) + collector.on_task_update_failure(event) + expect(backend).to have_received(:increment).with( + 'task_update_failed_total', labels: { task_type: 'my_task' } + ) + end + end + + # Canonical-only events should be no-ops + describe 'canonical-only stubs' do + it 'does not raise on on_task_update_completed' do + event = Conductor::Worker::Events::TaskUpdateCompleted.new( + task_type: 'my_task', task_id: 't1', worker_id: 'w1', + workflow_instance_id: 'wf1', duration_ms: 50.0 + ) + expect { collector.on_task_update_completed(event) }.not_to raise_error + expect(backend).not_to have_received(:observe) + end + + it 'does not raise on on_task_paused' do + event = Conductor::Worker::Events::TaskPaused.new(task_type: 'my_task') + expect { collector.on_task_paused(event) }.not_to raise_error + end + + it 'does not raise on on_thread_uncaught_exception' do + event = Conductor::Worker::Events::ThreadUncaughtException.new(cause: RuntimeError.new('boom')) + expect { collector.on_thread_uncaught_exception(event) }.not_to raise_error + end + + it 'does not raise on on_active_workers_changed' do + event = Conductor::Worker::Events::ActiveWorkersChanged.new(task_type: 'my_task', count: 3) + expect { collector.on_active_workers_changed(event) }.not_to raise_error + end + + it 'does not raise on on_workflow_start_error' do + event = Conductor::Worker::Events::WorkflowStartError.new( + workflow_type: 'my_wf', cause: RuntimeError.new('fail') + ) + expect { collector.on_workflow_start_error(event) }.not_to raise_error + end + + it 'does not raise on on_workflow_input_size' do + event = Conductor::Worker::Events::WorkflowInputSize.new( + workflow_type: 'my_wf', size_bytes: 1024 + ) + expect { collector.on_workflow_input_size(event) }.not_to raise_error + end + + it 'does not raise on on_http_api_request' do + event = Conductor::Worker::Events::HttpApiRequest.new( + method: 'GET', uri: '/api/tasks', status: '200', duration_ms: 50.0 + ) + expect { collector.on_http_api_request(event) }.not_to raise_error + end + end +end diff --git a/spec/conductor/worker/metrics_collector_spec.rb b/spec/conductor/worker/metrics_collector_spec.rb index 080cf25..37b8f06 100644 --- a/spec/conductor/worker/metrics_collector_spec.rb +++ b/spec/conductor/worker/metrics_collector_spec.rb @@ -4,225 +4,98 @@ require_relative '../../../lib/conductor/worker/telemetry/metrics_collector' RSpec.describe Conductor::Worker::Telemetry::MetricsCollector do - let(:backend) { double('backend') } - let(:collector) { described_class.new(backend: backend) } - - before do - allow(backend).to receive(:increment) - allow(backend).to receive(:observe) - allow(backend).to receive(:set) - end - - describe '#initialize' do - it 'uses NullBackend by default' do - collector = described_class.new - expect(collector.backend).to be_a(Conductor::Worker::Telemetry::NullBackend) + describe '.create' do + around do |example| + old_val = ENV.fetch('WORKER_CANONICAL_METRICS', nil) + example.run + ensure + if old_val.nil? + ENV.delete('WORKER_CANONICAL_METRICS') + else + ENV['WORKER_CANONICAL_METRICS'] = old_val + end end - it 'uses NullBackend when :null is specified' do - collector = described_class.new(backend: :null) - expect(collector.backend).to be_a(Conductor::Worker::Telemetry::NullBackend) + it 'returns a LegacyMetricsCollector by default' do + ENV.delete('WORKER_CANONICAL_METRICS') + collector = described_class.create + expect(collector).to be_a(Conductor::Worker::Telemetry::LegacyMetricsCollector) end - it 'accepts a custom backend instance' do - custom_backend = Object.new - collector = described_class.new(backend: custom_backend) - expect(collector.backend).to eq(custom_backend) + it 'returns a LegacyMetricsCollector when WORKER_CANONICAL_METRICS is false' do + ENV['WORKER_CANONICAL_METRICS'] = 'false' + collector = described_class.create + expect(collector).to be_a(Conductor::Worker::Telemetry::LegacyMetricsCollector) end - end - - describe '#on_poll_started' do - it 'increments task_poll_total counter' do - event = Conductor::Worker::Events::PollStarted.new( - task_type: 'my_task', - worker_id: 'worker-1', - poll_count: 5 - ) - - collector.on_poll_started(event) - expect(backend).to have_received(:increment).with( - 'task_poll_total', - labels: { task_type: 'my_task' } - ) + it 'returns a CanonicalMetricsCollector when WORKER_CANONICAL_METRICS is true' do + ENV['WORKER_CANONICAL_METRICS'] = 'true' + collector = described_class.create(subscribe_global_http: false) + expect(collector).to be_a(Conductor::Worker::Telemetry::CanonicalMetricsCollector) end - end - describe '#on_poll_completed' do - it 'observes task_poll_time_seconds histogram' do - event = Conductor::Worker::Events::PollCompleted.new( - task_type: 'my_task', - duration_ms: 150.0, - tasks_received: 3 - ) + it 'accepts "1" as truthy for WORKER_CANONICAL_METRICS' do + ENV['WORKER_CANONICAL_METRICS'] = '1' + collector = described_class.create(subscribe_global_http: false) + expect(collector).to be_a(Conductor::Worker::Telemetry::CanonicalMetricsCollector) + end - collector.on_poll_completed(event) + it 'accepts "yes" as truthy for WORKER_CANONICAL_METRICS' do + ENV['WORKER_CANONICAL_METRICS'] = 'yes' + collector = described_class.create(subscribe_global_http: false) + expect(collector).to be_a(Conductor::Worker::Telemetry::CanonicalMetricsCollector) + end - expect(backend).to have_received(:observe).with( - 'task_poll_time_seconds', - 0.15, - labels: { task_type: 'my_task' } - ) + it 'is case-insensitive for WORKER_CANONICAL_METRICS' do + ENV['WORKER_CANONICAL_METRICS'] = 'TRUE' + collector = described_class.create(subscribe_global_http: false) + expect(collector).to be_a(Conductor::Worker::Telemetry::CanonicalMetricsCollector) end - end - describe '#on_poll_failure' do - it 'increments task_poll_error_total counter with error class' do - error = StandardError.new('Connection refused') - event = Conductor::Worker::Events::PollFailure.new( - task_type: 'my_task', - duration_ms: 100.0, - cause: error - ) + it 'passes backend option through to the collector' do + ENV.delete('WORKER_CANONICAL_METRICS') + collector = described_class.create(backend: :null) + expect(collector.backend).to be_a(Conductor::Worker::Telemetry::NullBackend) + end - collector.on_poll_failure(event) + it 'legacy collector returns "legacy" from collector_name' do + ENV.delete('WORKER_CANONICAL_METRICS') + collector = described_class.create + expect(collector.collector_name).to eq('legacy') + end - expect(backend).to have_received(:increment).with( - 'task_poll_error_total', - labels: { task_type: 'my_task', error: 'StandardError' } - ) + it 'canonical collector returns "canonical" from collector_name' do + ENV['WORKER_CANONICAL_METRICS'] = 'true' + collector = described_class.create(subscribe_global_http: false) + expect(collector.collector_name).to eq('canonical') end end - describe '#on_task_execution_started' do - it 'can be called without error' do - event = Conductor::Worker::Events::TaskExecutionStarted.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456' - ) - - # Currently a no-op but should not raise - expect { collector.on_task_execution_started(event) }.not_to raise_error + describe '.canonical_metrics_enabled?' do + around do |example| + old_val = ENV.fetch('WORKER_CANONICAL_METRICS', nil) + example.run + ensure + if old_val.nil? + ENV.delete('WORKER_CANONICAL_METRICS') + else + ENV['WORKER_CANONICAL_METRICS'] = old_val + end end - end - describe '#on_task_execution_completed' do - it 'observes task_execute_time_seconds histogram' do - event = Conductor::Worker::Events::TaskExecutionCompleted.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456', - duration_ms: 500.0, - output_size_bytes: 2048 - ) - - collector.on_task_execution_completed(event) - - expect(backend).to have_received(:observe).with( - 'task_execute_time_seconds', - 0.5, - labels: { task_type: 'my_task' } - ) - end - - it 'observes task_result_size_bytes when output_size_bytes is present' do - event = Conductor::Worker::Events::TaskExecutionCompleted.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456', - duration_ms: 500.0, - output_size_bytes: 2048 - ) - - collector.on_task_execution_completed(event) - - expect(backend).to have_received(:observe).with( - 'task_result_size_bytes', - 2048, - labels: { task_type: 'my_task' } - ) - end - - it 'skips task_result_size_bytes when output_size_bytes is nil' do - event = Conductor::Worker::Events::TaskExecutionCompleted.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456', - duration_ms: 500.0, - output_size_bytes: nil - ) - - collector.on_task_execution_completed(event) - - expect(backend).to have_received(:observe).once # Only duration, not size + it 'returns false by default' do + ENV.delete('WORKER_CANONICAL_METRICS') + expect(described_class.canonical_metrics_enabled?).to be false end - end - describe '#on_task_execution_failure' do - it 'increments task_execute_error_total counter with error details' do - error = ArgumentError.new('Invalid input') - event = Conductor::Worker::Events::TaskExecutionFailure.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456', - duration_ms: 100.0, - cause: error, - is_retryable: true - ) - - collector.on_task_execution_failure(event) - - expect(backend).to have_received(:increment).with( - 'task_execute_error_total', - labels: { - task_type: 'my_task', - exception: 'ArgumentError', - retryable: 'true' - } - ) - end - - it 'tracks non-retryable errors' do - error = RuntimeError.new('Fatal error') - event = Conductor::Worker::Events::TaskExecutionFailure.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456', - duration_ms: 100.0, - cause: error, - is_retryable: false - ) - - collector.on_task_execution_failure(event) - - expect(backend).to have_received(:increment).with( - 'task_execute_error_total', - labels: { - task_type: 'my_task', - exception: 'RuntimeError', - retryable: 'false' - } - ) + it 'returns true when set to "true"' do + ENV['WORKER_CANONICAL_METRICS'] = 'true' + expect(described_class.canonical_metrics_enabled?).to be true end - end - describe '#on_task_update_failure' do - it 'increments task_update_failed_total counter' do - error = StandardError.new('Network error') - task_result = Conductor::Http::Models::TaskResult.complete - event = Conductor::Worker::Events::TaskUpdateFailure.new( - task_type: 'my_task', - task_id: 'task-123', - worker_id: 'worker-1', - workflow_instance_id: 'workflow-456', - cause: error, - retry_count: 4, - task_result: task_result - ) - - collector.on_task_update_failure(event) - - expect(backend).to have_received(:increment).with( - 'task_update_failed_total', - labels: { task_type: 'my_task' } - ) + it 'returns false for arbitrary strings' do + ENV['WORKER_CANONICAL_METRICS'] = 'maybe' + expect(described_class.canonical_metrics_enabled?).to be false end end end diff --git a/spec/conductor/worker/ractor_task_runner_spec.rb b/spec/conductor/worker/ractor_task_runner_spec.rb index 111c8bc..3dd28c4 100644 --- a/spec/conductor/worker/ractor_task_runner_spec.rb +++ b/spec/conductor/worker/ractor_task_runner_spec.rb @@ -47,6 +47,95 @@ expect { runner.shutdown }.not_to raise_error end end + + describe 'event publishing helpers' do + let(:runner) do + described_class.new(worker, configuration: configuration, ractor_id: 0) + end + + let(:task_result) do + Conductor::Http::Models::TaskResult.new.tap do |r| + r.task_id = 'task-123' + r.workflow_instance_id = 'workflow-456' + end + end + + before do + runner.instance_variable_set(:@worker_id, 'test-worker-ractor-0') + end + + describe '#publish_task_update_completed' do + it 'publishes a TaskUpdateCompleted event with duration_ms' do + published = nil + allow(runner).to receive(:publish_event) { |event| published = event } + + runner.send(:publish_task_update_completed, task_result, 42.5) + + expect(published).to be_a(Conductor::Worker::Events::TaskUpdateCompleted) + expect(published.task_type).to eq('test_task') + expect(published.task_id).to eq('task-123') + expect(published.duration_ms).to eq(42.5) + end + end + + describe '#publish_task_update_failure' do + it 'publishes a TaskUpdateFailure event with duration_ms' do + published = nil + allow(runner).to receive(:publish_event) { |event| published = event } + + error = StandardError.new('update failed') + runner.send(:publish_task_update_failure, task_result, error, 99.0) + + expect(published).to be_a(Conductor::Worker::Events::TaskUpdateFailure) + expect(published.task_type).to eq('test_task') + expect(published.cause).to eq(error) + expect(published.duration_ms).to eq(99.0) + expect(published.retry_count).to eq(Conductor::Worker::RactorTaskRunner::RETRY_BACKOFFS.size) + end + end + + describe '#publish_uncaught_exception' do + it 'publishes a ThreadUncaughtException event' do + published = nil + allow(runner).to receive(:publish_event) { |event| published = event } + + error = RuntimeError.new('crash') + runner.send(:publish_uncaught_exception, error) + + expect(published).to be_a(Conductor::Worker::Events::ThreadUncaughtException) + expect(published.cause).to eq(error) + expect(published.task_type).to eq('test_task') + end + + it 'does not raise when publish_event fails' do + allow(runner).to receive(:publish_event).and_raise(StandardError.new('dispatch error')) + + expect { runner.send(:publish_uncaught_exception, RuntimeError.new('test')) }.not_to raise_error + end + end + end + + describe 'poll_task when paused' do + let(:paused_worker) do + Conductor::Worker::Worker.new('test_task', poll_interval: 100, paused: true) do |task| + { result: task.input_data['value'] } + end + end + + it 'publishes TaskPaused and returns nil' do + runner = described_class.new(paused_worker, configuration: configuration, ractor_id: 0) + runner.instance_variable_set(:@worker_id, 'test-worker-ractor-0') + + published = nil + allow(runner).to receive(:publish_event) { |event| published = event } + + result = runner.send(:poll_task) + + expect(result).to be_nil + expect(published).to be_a(Conductor::Worker::Events::TaskPaused) + expect(published.task_type).to eq('test_task') + end + end end describe Conductor::Worker::RactorSupport do diff --git a/spec/conductor/worker/task_runner_spec.rb b/spec/conductor/worker/task_runner_spec.rb index dca9273..669d32d 100644 --- a/spec/conductor/worker/task_runner_spec.rb +++ b/spec/conductor/worker/task_runner_spec.rb @@ -283,6 +283,8 @@ end describe 'paused worker' do + let(:received_events) { [] } + let(:paused_worker) do Conductor::Worker::Worker.new('test_task', poll_interval: 100, paused: true) do |task| { result: task.input_data['value'] } @@ -298,10 +300,148 @@ ) end + before do + event_dispatcher.register(Conductor::Worker::Events::TaskPaused, + ->(event) { received_events << [:task_paused, event] }) + end + it 'does not poll when paused' do runner.run_once expect(task_client).not_to have_received(:batch_poll_tasks) end + + it 'publishes TaskPaused event' do + runner.run_once + + paused = received_events.find { |e| e[0] == :task_paused }&.last + expect(paused).not_to be_nil + expect(paused.task_type).to eq('test_task') + end + end + + describe 'task update completed events' do + let(:received_events) { [] } + + let(:task_data) do + Conductor::Http::Models::Task.new.tap do |t| + t.task_id = 'task-123' + t.workflow_instance_id = 'workflow-456' + t.task_def_name = 'test_task' + t.input_data = { 'value' => 21 } + end + end + + before do + allow(task_client).to receive(:batch_poll_tasks).and_return([task_data]) + + event_dispatcher.register(Conductor::Worker::Events::TaskUpdateCompleted, + ->(event) { received_events << [:update_completed, event] }) + end + + it 'publishes TaskUpdateCompleted with duration_ms on successful update' do + runner.run_once + sleep(0.5) + + completed = received_events.find { |e| e[0] == :update_completed }&.last + expect(completed).not_to be_nil + expect(completed.task_type).to eq('test_task') + expect(completed.duration_ms).to be_a(Numeric) + expect(completed.duration_ms).to be >= 0 + end + end + + describe 'task update failure duration_ms' do + let(:received_events) { [] } + + let(:task_data) do + Conductor::Http::Models::Task.new.tap do |t| + t.task_id = 'task-123' + t.workflow_instance_id = 'workflow-456' + t.task_def_name = 'test_task' + t.input_data = { 'value' => 21 } + end + end + + before do + allow(task_client).to receive(:batch_poll_tasks).and_return([task_data]) + allow(task_client).to receive(:update_task).and_raise(StandardError.new('Update failed')) + + event_dispatcher.register(Conductor::Worker::Events::TaskUpdateFailure, + ->(event) { received_events << [:update_failure, event] }) + + stub_const('Conductor::Worker::TaskRunner::RETRY_BACKOFFS', [0, 0, 0, 0].freeze) + end + + it 'includes duration_ms in TaskUpdateFailure event' do + runner.run_once + sleep(1.0) + + failure = received_events.find { |e| e[0] == :update_failure }&.last + expect(failure).not_to be_nil + expect(failure.duration_ms).to be_a(Numeric) + expect(failure.duration_ms).to be >= 0 + end + end + + describe 'active workers changed events' do + let(:received_events) { [] } + + let(:task_data) do + Conductor::Http::Models::Task.new.tap do |t| + t.task_id = 'task-123' + t.workflow_instance_id = 'workflow-456' + t.task_def_name = 'test_task' + t.input_data = { 'value' => 21 } + end + end + + before do + allow(task_client).to receive(:batch_poll_tasks).and_return([task_data]) + + event_dispatcher.register(Conductor::Worker::Events::ActiveWorkersChanged, + ->(event) { received_events << [:active_workers, event] }) + end + + it 'publishes ActiveWorkersChanged when a task is submitted' do + runner.run_once + sleep(0.5) + + events = received_events.select { |e| e[0] == :active_workers } + expect(events).not_to be_empty + last_event = events.last.last + expect(last_event.task_type).to eq('test_task') + expect(last_event.count).to be_a(Integer) + end + end + + describe 'uncaught exception events' do + it 'publishes ThreadUncaughtException via publish_uncaught_exception' do + received_events = [] + event_dispatcher.register(Conductor::Worker::Events::ThreadUncaughtException, + ->(event) { received_events << event }) + + error = RuntimeError.new('unexpected failure') + runner.send(:publish_uncaught_exception, error) + + event = received_events.first + expect(event).not_to be_nil + expect(event.cause).to eq(error) + expect(event.task_type).to eq('test_task') + end + + it 'does not raise when publish_uncaught_exception itself fails' do + broken_dispatcher = Conductor::Worker::Events::SyncEventDispatcher.new + broken_dispatcher.register(Conductor::Worker::Events::ThreadUncaughtException, + ->(_event) { raise 'listener boom' }) + broken_runner = described_class.new( + worker, + configuration: configuration, + event_dispatcher: broken_dispatcher, + logger: logger + ) + + expect { broken_runner.send(:publish_uncaught_exception, RuntimeError.new('test')) }.not_to raise_error + end end end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index d20e4b5..8d40283 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -14,6 +14,10 @@ c.syntax = :expect end + config.before do + Conductor::Worker::Events::GlobalDispatcher.reset! + end + # Run specs in random order config.order = :random Kernel.srand config.seed