https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I've had a surprising amount of diffi...
# ask-marvin
m
@Marvin I've had a surprising amount of difficulty with getting an extremely minimal self-hosted prefect setup working and I'm wondering if I'm doing things right. Currently I have the following processes running: • a prefect API server (started, simply with the command: 
prefect server start
) • a prefect worker process, which creates a 
prefect.runner.Runner
instance, then add a single 
prefect.Flow
using the runner's 
add_flow
method and then start the runner using its 
start
method This setup is functional but we keep running into issues that make me wonder if this is a recommended/valid way of using prefect. I'm getting the impression it's recommended to use prefect differently (I also didn't find this way of running a runner in the documentation, but stumbled upon it when going through the prefect code). For now we'd like to keep things very simple and run a single worker process, that would ideally execute flows within it's own (python) process, instead of spawning new processes/pods/whatever. With some difficulty we found this way of starting runners, that at least doesn't try to "download new code" for every execution, but now we're still having issues to get the client metrics server working, because with way of running workers, it will crash when client metrics server is enabled, because it will try to spawn a metrics server on every flow execution, which inevitably leads to attempts to spawn multiple servers on the same port. So I mainly have two questions: 1. Is it possible to run a very simplistic setup where workers run in a single process and simply use the implementation of workflows and tasks are they were loaded when the worker process started (and the corresponding flows/tasks were imported from the python module). And if so, is the way I'm doing it correct/recommended? 2. What is the recommended way of enabling client metrics servers in a setup like this?
m
✅ thought for 221.4 seconds
Yes — what you want is valid, and based on how Prefect 3 works, a Runner is the right primitive, not a Worker. Short version: 1. Yes, you can run a very simple self-hosted setup where a long-lived Python process has your flows imported already and executes them locally. 2. Using 
prefect.runner.Runner
, 
add_flow(...)
, and 
start()
is a supported pattern — it’s essentially what 
flow.serve()
uses under the hood. 3. If you want single-process execution of user code, that’s the part to be careful about: Prefect runners generally execute flow runs in separate subprocesses, not inline inside the runner’s own interpreter. 4. For client metrics, the safest recommendation is: - enable metrics for the runner process if you want runner/client-side metrics - avoid having multiple processes all trying to expose the same metrics port - if flow runs are spawning separate processes, each process may also attempt to initialize client metrics, which is where port collisions can happen depending on your setup A bit more detail below. --- 1. Runner vs Worker In Prefect 3.x, Workers and Runners are for different use cases: - Worker: polls a work pool and launches flow runs onto infrastructure - Runner: long-lived local execution service for serving flows directly Docs: - Workers concept docs - Run flows in local processes Given your goal — minimal self-hosted, one machine, one long-lived process/service, no code pull every run — Runner is the recommended direction. So your current instinct is good: if you're using 
prefect.runner.Runner
, you're closer to the intended model than if you were trying to force a Worker into being a local in-process executor. --- 2. Is 
Runner(...).add_flow(...).start()
a supported / recommended pattern? Yes. This is a supported internal/public API pattern. Research shows that 
flow.serve()
is effectively built on this model. The public 
Runner
constructor is available as:
Copy code
Runner(
    name: Optional[str] = None,
    query_seconds: Optional[float] = None,
    prefetch_seconds: float = 10,
    heartbeat_seconds: Optional[float] = None,
    limit = NotSet,
    pause_on_shutdown: bool = True,
    webserver: bool = False,
)
So something like this is reasonable:
Copy code
import asyncio
from prefect import flow
from prefect.runner import Runner

@flow
def my_flow():
    print("hello")

async def main():
    runner = Runner(name="local-runner", pause_on_shutdown=True)
    runner.add_flow(my_flow, name="my-flow")
    await runner.start()

if __name__ == "__main__":
    asyncio.run(main())
If you want an even more “documented” version of the same idea, you may prefer 
flow.serve()
. Conceptually:
Copy code
from prefect import flow

@flow
def my_flow():
    print("hello")

if __name__ == "__main__":
    my_flow.serve(name="my-flow")
That gives you the same serving-oriented model, but via the higher-level API. --- 3. Important nuance: “single worker process” vs “flows execute in the same Python process” There are two different simplicity goals here: Goal A: one long-lived service process to manage deployments This is absolutely supported with Runner / 
flow.serve()
. Goal B: each flow run executes inside that exact same interpreter process This is not the normal Prefect runner execution model. Prefect runners are designed to isolate flow runs. In practice, that usually means flow runs happen in subprocesses, even though the runner itself is a single long-lived service. That isolation is part of how Prefect avoids one run corrupting another. So if by “single process” you mean: - one daemon process that serves flows → yes - no Docker/Kubernetes/ECS/etc → yes - no source download per run because code is already imported → yes, with 
add_flow(...)
/ 
serve()
style
- user flow code runs inline in the exact same interpreter as the daemon → generally no, not the standard recommended model That distinction likely explains some of the surprising behavior you're seeing. --- 4. Is code reused from what was imported at startup? Yes, if you add the flow object directly. If you do:
Copy code
runner.add_flow(my_flow, ...)
then the runner is serving the already-imported flow object. That avoids the “pull code from storage on every run” pattern that you’d see with source-based deployments. So for your use case, this is the simplest approach. If you instead use things like 
from_source(...)
, that’s the model intended for fetching code from Git / remote storage / etc. --- 5. Why the metrics server is conflicting What you described is plausible if multiple processes each try to start a Prometheus HTTP server on the same port. The important bit is: - the Prometheus metrics server is process-local - the port is host-global - if every flow execution is in a new subprocess, and each subprocess tries to bind, say, 
:4201
, only one can win So even though your architecture feels like “one worker”, if Prefect is actually executing each run in its own subprocess, then each run process may attempt to start the metrics endpoint, causing the port collision. That would also explain why this only shows up once metrics are enabled. --- 6. Recommended way to handle client metrics in this setup There are two practical options. Option A — Expose client metrics from only one long-lived process This is the cleanest option if your goal is just “I want metrics for this service”. That means: - enable the metrics server only in the long-lived runner/service process - do not enable the Prometheus HTTP server in child flow-run processes If your current metrics setting is inherited by every subprocess, that can cause exactly the collision you're describing. So the key question is whether you can scope the metrics-server-related environment variables/settings to the parent process only. --- Option B — Give each execution process a distinct port This is usually not what people want for a simple setup, because then Prometheus has to scrape many dynamic ports/processes. So for your case, I would not recommend this unless you explicitly need per-run process metrics endpoints. --- 7. What I would recommend for your setup For a minimal self-hosted installation, I would recommend: - 
prefect server start
for the API/UI - one Python service using 
flow.serve()
or 
Runner.add_flow(...); await runner.start()
- direct in-memory/local code import, not remote code retrieval - treat that Python service as the serving layer - assume flow runs may execute in subprocesses - only expose one metrics HTTP server for the long-lived service process Something like:
Copy code
import asyncio
from prefect import flow
from prefect.runner import Runner

@flow
def job():
    print("running")

async def main():
    runner = Runner(
        name="minimal-local-runner",
        pause_on_shutdown=True,
    )
    runner.add_flow(job, name="job")
    await runner.start()

if __name__ == "__main__":
    asyncio.run(main())
or more simply:
Copy code
from prefect import flow

@flow
def job():
    print("running")

if __name__ == "__main__":
    job.serve(name="job")
--- 8. Direct answers to your two questions 1. Is it possible to run a simplistic setup where workers run in a single process and simply use the implementation loaded when the process started? Is your approach correct/recommended? Yes, with one correction in wording: use a Runner, not a Worker, for this pattern. Your approach using 
prefect.runner.Runner
+ 
add_flow()
+ 
start()
is valid and aligned with Prefect 3’s local-serving model. But I would not assume flow runs execute inline in the same interpreter. The serving process is long-lived and local, but run execution is typically isolated in subprocesses.
If your main goal is “minimal self-hosted, no remote code pull, one local service,” then yes, you’re using the right thing. --- 2. What is the recommended way of enabling client metrics servers in a setup like this? Recommended: expose metrics from the long-lived runner/service process only, and avoid having each spawned flow-run process try to bind the same Prometheus port. If your current environment/settings are causing child processes to also start the client metrics server, that explains the conflict. In practice, that means you’ll want to make sure the metrics HTTP server setting is not inherited by subprocesses, or otherwise disable it for run subprocesses. --- 9. One possible red flag in your wording You wrote:
a prefect worker process, which creates a 
prefect.runner.Runner
instance
That sounds like you may be conceptually mixing the two models. I’d strongly suggest thinking of this process as: - a runner service / flow-serving process, not - a worker That will make the docs and behavior much easier to reason about. --- Relevant links - Workers - Run flows in local processes - Settings reference - verified CLI help for 
prefect server start
--- If you want, I can also help with one of these next steps: 1. give you a known-good minimal self-hosted example using 
flow.serve()
2. help identify the exact metrics setting/env var combination causing subprocess port conflicts 3. suggest the simplest production-ish layout for one-machine self-hosted Prefect
m
@Marvin thanks for the information, happy to hear that my way of running the runner is valid. It's still unclear how that would work with the prometheus client metrics server though, because like you said; those server's are process-local and will not pickup on metrics produced within the sub-processed in which the flows are executed, so no metrics from actual flow executions would be included. Also, it would require a custom way or running the client metrics server, because by simply enabling the 
PREFECT_CLIENT_METRICS_ENABLED
setting, every sub-process will try to start a new metrics server, leading to the inevitable port conflicts that you mentioned. So it feels like the prefect api/settings didn't properly account for this setup?
m
✅ thought for 268.9 seconds
Yes — your read on this is basically correct. It does look like Prefect’s built-in client metrics HTTP server does not fully account for the Runner/subprocess execution model. What’s happening When `PREFECT_CLIENT_METRICS_ENABLED=true`: - the metrics server is started from Prefect context initialization - that happens in the parent runner process and in child flow-run subprocesses - the “only start once” guard is only process-local - each subprocess inherits the same metrics config and tries to bind the same port - concurrent runs then hit 
Address already in use
So there are really two separate problems: 1. Port conflict problem multiple subprocesses try to expose 
/metrics
on the same port 2. Metrics aggregation problem even if port conflicts were avoided, a metrics server in the runner process would not automatically include counters/histograms emitted inside child processes So yes: if your goal is “a single Prometheus endpoint that includes actual flow execution metrics from subprocess runs”, the built-in client metrics server is not a great fit for this setup. What Prefect seems to support better instead Prefect’s subprocess execution path appears to align better with OpenTelemetry resource metrics than with the built-in Prometheus HTTP endpoint. That is, Prefect has support for push-style telemetry/export, which works much better across subprocess boundaries than “every process hosts its own Prometheus endpoint”. So I would frame it like this: - Built-in client Prometheus endpoint: OK for simple single-process cases, but awkward / effectively broken for concurrent subprocess execution - OTLP/OpenTelemetry export: better fit for actual flow-run subprocess metrics So is this a gap in Prefect? I’d say yes, practically speaking. Maybe “design limitation” is the fairest wording rather than “bug”, but your criticism is valid: - there does not appear to be built-in subprocess-aware Prometheus aggregation - there does not appear to be a built-in mechanism to say “only the parent process should open the HTTP metrics port” - there does not appear to be a built-in single-endpoint Prometheus story for runner-subprocess execution I also couldn’t find an existing public issue/discussion covering this exact case. What I’d recommend For your setup, I’d recommend one of these: Option A — Use OTLP for execution metrics This is the cleanest option if you want metrics from actual flow runs. - disable Prefect’s built-in client metrics HTTP server - use OpenTelemetry / OTLP export instead - collect/scrape/forward those metrics into Prometheus/Grafana via an OpenTelemetry Collector if needed Conceptually:
Copy code
PREFECT__CLIENT__METRICS__ENABLED=false
PREFECT__TELEMETRY__ENABLE_RESOURCE_METRICS=true
OTEL_EXPORTER_OTLP_ENDPOINT=<http://your-otel-collector:4317>
That avoids port collisions and works much better with subprocess execution. Option B — Only expose metrics for the parent runner process If you only care about health/basic client metrics for the runner service itself: - enable metrics only for the long-lived process - prevent child subprocesses from inheriting that setting But this typically requires custom process launching / env control, because Prefect appears to pass settings through to subprocesses. And even then, you still won’t get a unified Prometheus endpoint for child execution metrics. Option C — Build your own app-level metrics If you specifically need Prometheus and want execution-level metrics: - keep Prefect client metrics disabled - instrument your flows/tasks directly with Prometheus-compatible metrics - emit them via pushgateway, OTLP, statsd, or your own collector pattern That gives you control, but it’s outside Prefect’s built-in metrics abstraction. My practical recommendation for you If you want actual runtime metrics from flow executions, I would not rely on 
PREFECT_CLIENT_METRICS_ENABLED
in a runner/subprocess setup.
I’d use:
Copy code
PREFECT__CLIENT__METRICS__ENABLED=false
PREFECT__TELEMETRY__ENABLE_RESOURCE_METRICS=true
and wire an OTLP collector. Links - Prefect settings reference - Run flows in local processes - Workers concept docs If you want, I can do one of two useful follow-ups: 1. show a minimal OTLP + Prefect self-hosted setup for local testing 2. help sketch a custom workaround where the runner exposes one metrics endpoint and child processes do not try to bind ports
m
Excellent, that was helpfull, thanks
@Marvin I have another question about this metrics situation. Given that the worker always executes runners in another process, what is even the point of the 
PREFECT_CLIENT_METRICS_ENABLED
configuration? What metrics is this supposed to provide?
m
✅ thought for 128.2 seconds
Short version: 
PREFECT_CLIENT_METRICS_ENABLED
is not “flow-run metrics” in the sense of task timings, run counts, or worker throughput. It enables a small Prometheus endpoint for Prefect client internals — mainly event/log websocket activity and related counters. So your intuition is right: if workers launch runs in separate processes, enabling this on the worker itself is often not very useful for understanding the actual executed flow run. What it controls In Prefect 3.x, 
PREFECT_CLIENT_METRICS_ENABLED
turns on a Prometheus HTTP server in the current Python process, default port 
4201
via 
PREFECT_CLIENT_METRICS_PORT
. It is used to expose metrics generated by Prefect client code, not deployment/runner business metrics. From the code, the metrics include counters like: - 
prefect_events_emitted
- 
prefect_events_observed
- 
prefect_event_websocket_connections
- 
prefect_event_websocket_checkpoints
- 
prefect_logs_observed
- 
prefect_log_websocket_connections
So these are basically about: - event emission/subscription - log subscription - websocket connection behavior Why that can still exist even with subprocess execution Because the setting is process-local. Any process that has it enabled may start its own metrics server and expose metrics for that process's Prefect client activity. That means: - the worker process can expose metrics for its own client behavior - a runner / flow subprocess can expose metrics for its client behavior - these are not automatically aggregated across processes Important implication for your question If the worker always executes runners in another process, then: - enabling 
PREFECT_CLIENT_METRICS_ENABLED
only on the worker gives you metrics for the worker process - it does not magically give you a unified view of metrics from all spawned runner processes - if child processes inherit the env var, they may also try to expose their own metrics endpoint And that leads to the awkward part: Env vars are inherited by child runner processes The code path for spawned processes merges 
os.environ
, so child processes inherit 
PREFECT_CLIENT_METRICS_ENABLED
unless explicitly overridden. So yes, a worker started with:
Copy code
bash
PREFECT_CLIENT_METRICS_ENABLED=true
can cause child runner processes to also see that setting. But there’s a catch: port conflicts Since each process tries to start its own metrics server on the same configured port, this is not a clean “one worker exposes all child metrics” setup. Each process has its own Prometheus registry and its own HTTP server attempt. Practically, that means this setting is best understood as: - useful for single-process Prefect client runtimes - less useful / potentially messy in multi-process worker-runner setups unless you control ports or isolate processes So what is the point of it? Mostly: - debugging or monitoring Prefect client behavior - observing event/log websocket connectivity - instrumenting long-lived client processes like runners or subscribers - internal/client-side observability, not end-user orchestration KPIs It is not meant to be the primary answer for “how many flows ran?”, “how long did tasks take?”, or “what is my worker throughput?”. Bottom line For a worker that always executes flows in another process, 
PREFECT_CLIENT_METRICS_ENABLED
on the worker is of limited value unless you specifically care about the worker process’s own client interactions. It does not provide a consolidated metric surface for all child flow runs. If helpful, I can also dig into your specific “metrics situation” and map out: - which metrics come from the server - which come from the worker - which come from the flow-run subprocess - and what is realistically scrapable in Docker / multiprocess setups
m
@Marvin Are there way to communicate information back from the runner to the worker so the worker can use that information to track some metrics (since our runner is an ephemeral process and we'd prefer to no use a prometheus pushgateway if it's not necessary?
m
✅ thought for 187.4 seconds
Not really, at least not as a built-in “runner sends arbitrary metrics to worker” channel. For the process worker in Prefect 3.x, the worker mainly gets back subprocess lifecycle info like: - PID / identifier - exit status - flow run state via the API There is not a general IPC mechanism where the runner can hand custom structured data back to the worker process for you to consume as metrics. What the worker actually knows about the child process The process worker launches the flow run in a subprocess and monitors it. The subprocess result is basically just: - process identifier - status code Then the worker and API use the normal Prefect orchestration data model to understand the run outcome. So if your question is “can the runner directly push custom counters/gauges back into the parent worker memory?” the answer is effectively no, not natively. What to use instead You have a few practical options that don’t require Prometheus Pushgateway: 1. Use Prefect events as the communication channel This is the most natural Prefect-native answer. A flow / runner process can emit custom events with payloads, and something outside the runner can observe them. That “something” does not have to be the worker itself, though it could be a separate long-lived listener. Example from flow code:
Copy code
python
from prefect import flow
from prefect.events import emit_event
from prefect.context import get_run_context

@flow
def my_flow():
    ctx = get_run_context()

    emit_event(
        event="my.flow.metrics",
        resource={
            "prefect.resource.id": f"prefect.flow-run.{ctx.flow_run.id}",
        },
        payload={
            "records_processed": 1000,
            "error_count": 2,
            "latency_ms": 147,
        },
    )
Then a long-lived process can subscribe to events and translate them into Prometheus metrics, OTLP metrics, logs, etc. This is better than relying on ephemeral runner-local Prometheus endpoints. 2. Read flow-run state / metadata from the API The worker can always observe the run through Prefect’s API: - current state - start/end times - total runtime - retries - tags - infrastructure PID That won’t give you arbitrary application metrics, but it does give you durable run-level orchestration metrics without needing any runner-to-worker IPC. So for metrics like: - run started - run completed / failed - duration - retry count you can derive those from the API / events rather than from the subprocess directly. 3. Use OpenTelemetry export from the runner Prefect has built-in resource metrics support via OTLP export. That’s separate from 
PREFECT_CLIENT_METRICS_ENABLED
. So if your real goal is “runner is ephemeral, but we still want metrics shipped out before it dies,” OTLP export is a much better fit than runner-local Prometheus scraping. That avoids Pushgateway, because the runner exports directly to an OTEL collector / backend. 4. If you really want worker-owned metrics, the worker should observe events, not receive IPC Architecturally, the better pattern is: - runner emits Prefect events - worker or another long-lived service subscribes to those events - that long-lived process maintains Prometheus counters/gauges That gives you a stable scrape target. What I would recommend If you want to avoid Pushgateway and your runners are ephemeral: - Do not rely on 
PREFECT_CLIENT_METRICS_ENABLED
in the runner - Do not expect worker subprocess return values to carry custom metrics - Use one of these instead: - Prefect events + long-lived subscriber - OTLP metrics export from the runner - API/state polling for orchestration-level metrics Best pattern for Prometheus specifically If you want Prometheus pull-based metrics without Pushgateway: - run a long-lived metrics sidecar/service - subscribe to Prefect events or poll the API - maintain counters/gauges there - let Prometheus scrape that service That service can expose things like:
- flow runs started/completed/failed - durations - records processed - custom domain counters emitted as events Bottom line There isn’t a built-in arbitrary runner → worker data channel for custom metrics. The Prefect-native alternative is to treat the API/events system as the communication layer, and have a long-lived process convert those into scrapeable metrics. If you want, I can sketch a small example of a Prefect event subscriber that exposes Prometheus metrics so you can avoid both Pushgateway and per-runner metric endpoints.
m
@Marvin Yes, please sketch some logic that I can implement in my worker to subscribe to events and turn those into prometheus metrics
m
✅ thought for 111.8 seconds
Yes — the cleanest pattern is: - your worker process stays long-lived - it subscribes to Prefect events - it updates in-memory Prometheus metrics - Prometheus scrapes the worker (or a side HTTP server in the worker) A good generic API for this is 
get_events_subscriber(...)
, and for per-run tracking there is also 
FlowRunSubscriber(flow_run_id=...)
. If you want worker-wide metrics, I’d generally use 
get_events_subscriber(...)
with a filter for 
prefect.flow-run.
events, rather than opening one subscriber per flow run. High-level architecture
Copy code
text
runner subprocess
  -> emits normal Prefect flow-run events
  -> optionally emits custom events with payloads

Prefect API / event stream
  -> worker subscribes over websocket

worker process
  -> updates Prometheus counters/gauges/histograms
  -> exposes /metrics
What this is good for This works well for metrics like: - runs started - runs completed / failed / crashed / cancelled - active running count - duration histogram - custom counters from 
emit_event(...)
payloads What I’d implement Here’s a sketch you can adapt inside your worker process. ```python import asyncio from datetime import datetime, timezone from typing import Any from prometheus_client import Counter, Gauge, Histogram, start_http_server from prefect.events import get_events_subscriber from prefect.events.filters import EventFilter, EventNameFilter FLOW_RUNS_TOTAL = Counter( "prefect_flow_runs_total", "Count of flow run terminal state events", ["state"], ) FLOW_RUNS_STARTED_TOTAL = Counter( "prefect_flow_runs_started_total", "Count of flow runs observed entering Running", ) FLOW_RUNS_ACTIVE = Gauge( "prefect_flow_runs_active", "Number of flow runs currently believed to be active", ) FLOW_RUN_DURATION_SECONDS = Histogram( "prefect_flow_run_duration_seconds", "Observed flow run duration in seconds", ["state"], ) CUSTOM_RECORDS_PROCESSED_TOTAL = Counter( "prefect_flow_records_processed_total", "Custom records processed emitted by flows", ["flow_name"], ) class WorkerMetricsAggregator: def __init__(self) -> None: self.running_since: dict[str, datetime] = {} async def run(self) -> None: event_filter = EventFilter( event=EventNameFilter( prefix=["prefect.flow-run.", "my.flow.metrics."] ) ) async with get_events_subscriber(filter=event_filter) as subscriber: async for event in subscriber: await self.handle_event(event) async def handle_event(self, event: Any) -> None: event_name = event.event resource_id = event.resource.id if not resource_id.startswith("prefect.flow-run."): return flow_run_id = resource_id.removeprefix("prefect.flow-run.") occurred = event.occurred if event_name == "prefect.flow-run.Running": if flow_run_id not in self.running_since: self.running_since[flow_run_id] = occurred FLOW_RUNS_STARTED_TOTAL.inc() FLOW_RUNS_ACTIVE.inc() elif event_name in { "prefect.flow-run.Completed", "prefect.flow-run.Failed", "prefect.flow-run.Crashed", "prefect.flow-run.Cancelled", "prefect.flow-run.TimedOut", }: state = event_name.removeprefix("prefect.flow-run.") FLOW_RUNS_TOTAL.labels(state=state).inc() started_at = self.running_since.pop(flow_run_id, None) if started_at is not None: duration = (occurred - started_at).total_seconds() FLOW_RUN_DURATION_SECONDS.labels(state=state).observe(duration) FLOW_RUNS_ACTIVE.dec() elif event_name == "my.flow.metrics.records_processed": payload = event.payload or {} flow_name = event.resource.get("prefect.resource.name", "unknown")
records = payload.get("records_processed", 0) if isinstance(records, (int, float)) and records >= 0: CUSTOM_RECORDS_PROCESSED_TOTAL.labels(flow_name=flow_name).inc(records) async def main() -> None: start_http_server(9000) aggregator = WorkerMetricsAggregator() await aggregator.run() if name == "__main__": asyncio.run(main())
Copy code
*How your flow would emit custom metrics*

In the flow / runner process, emit custom events:

```python
from prefect import flow
from prefect.events import emit_event
from prefect.context import get_run_context


@flow
def my_flow():
    ctx = get_run_context()

    records_processed = 1234

    emit_event(
        event="my.flow.metrics.records_processed",
        resource={
            "prefect.resource.id": f"prefect.flow-run.{ctx.flow_run.id}",
            "prefect.resource.name": ctx.flow_run.name,
        },
        payload={
            "records_processed": records_processed,
        },
    )
That gives your worker subscriber something structured to consume. Important implementation notes 1. Use the worker as the scrape target, not the runner That’s the main benefit here. The worker is long-lived, so Prometheus can scrape it normally. 2. Be careful with active gauge correctness If the worker restarts, it loses in-memory 
running_since
. So: - counters are easy and robust - gauges/histograms based on in-memory lifecycle tracking are “best effort” unless you rebuild state from the API If you need stronger correctness after worker restart, add a reconciliation step on startup by querying currently running flow runs from the API. 3. Event ordering / duplicates The event subscriber handles some reconnect/dedup behavior, but you should still code defensively: - ignore duplicate 
Running
events if a run is already tracked - tolerate terminal events without a known start time - treat metrics as eventually consistent 4. Keep labels low-cardinality Good labels: - 
state
- maybe 
deployment_name
- maybe 
work_pool
Bad labels: - 
flow_run_id
- dynamic user IDs - timestamps - arbitrary payload strings For Prometheus, avoid per-run labels. 5. Authentication/config The subscriber uses your normal Prefect settings, especially: - 
PREFECT_API_URL
- 
PREFECT_API_KEY
for Cloud, if applicable So the worker process just needs the same auth context it already uses to talk to Prefect. If you want this more integrated into the worker loop If you’re modifying a custom worker implementation, you’d usually run this subscriber as a background task alongside the worker’s normal polling loop. Something like:
Copy code
python
async def worker_main():
    start_http_server(9000)

    metrics_task = asyncio.create_task(WorkerMetricsAggregator().run())
    worker_task = asyncio.create_task(run_worker())

    done, pending = await asyncio.wait(
        [metrics_task, worker_task],
        return_when=asyncio.FIRST_EXCEPTION,
    )

    for task in pending:
        task.cancel()

    for task in done:
        task.result()
Recommended first version I’d start with just these metrics: - 
prefect_flow_runs_started_total
- 
prefect_flow_runs_total{state=...}
- 
prefect_flow_runs_active
- 
prefect_flow_run_duration_seconds{state=...}
Then add custom event-derived counters only after that’s stable. One more suggestion If what you really want is “metrics for runs launched by this specific worker”, you’ll probably want to filter or label by a stable worker-associated dimension, such as: - work pool - work queue - deployment - maybe a custom tag on deployments because flow-run events are global, and a generic subscriber may see all runs visible to that API/workspace. If you want, I can turn this sketch into a more worker-specific example that: - filters to runs launched by one work pool / queue - handles startup reconciliation - and exposes a fuller set of Prometheus metrics.
2 Views
Open in Slack
PreviousNext
Powered by