Great question. Assuming you’re on Prefect 3.x self-hosted, you’ve got a few good options to monitor “running/failed runs/average runtime”: 1) Use the built-in UI - If you started your server with the UI enabled (default), the dashboard already shows state counts and trends. Make sure the server is up with UI enabled and point your browser to your UI URL. - Docs: Self-hosted Server overview 2) Query the REST API (or Python client) for metrics - State counts and durations are available via the API. Two particularly useful endpoints: - Count flow runs by state: count_flow_runs - Aggregated durations and counts over time: flow_run_history - Example: quick Python script to get: - current running count - failed runs in the last 24 hours - average runtime of completed runs in the last 24 hours

import os
from datetime import datetime, timedelta, timezone
from prefect.client.orchestration import SyncPrefectClient
from prefect.server.schemas.filters import (
    FlowRunFilter,
    FlowRunFilterState,
    FlowRunFilterStateType,
    FlowRunFilterStartTime,
)

api_url = os.environ.get("PREFECT_API_URL", "<http://localhost:4200/api>")

now = datetime.now(timezone.utc)
since = now - timedelta(days=1)

client = SyncPrefectClient(api=api_url)

# 1) Currently running
running_count = client.count_flow_runs(
    flow_runs=FlowRunFilter(
        state=FlowRunFilterState(
            type=FlowRunFilterStateType(any_=["RUNNING"])
        )
    )
)

# 2) Failed in last 24h
failed_24h = client.count_flow_runs(
    flow_runs=FlowRunFilter(
        state=FlowRunFilterState(
            type=FlowRunFilterStateType(any_=["FAILED"])
        ),
        start_time=FlowRunFilterStartTime(after_=since)
    )
)

# 3) Average runtime (seconds) for completed runs in last 24h
# Use history to get summed run_time and counts
bucket_seconds = (now - since).total_seconds()  # single bucket for 24h
history = client.flow_run_history(
    history_start=since,
    history_end=now,
    history_interval=bucket_seconds,
    flow_runs=FlowRunFilter(
        state=FlowRunFilterState(
            type=FlowRunFilterStateType(any_=["COMPLETED"])
        )
    ),
)

total_runtime = sum(h.sum_estimated_run_time for h in history)
completed_count = sum(h.count_runs for h in history)
avg_runtime_sec = (total_runtime / completed_count) if completed_count else 0

print("Running now:", running_count)
print("Failed last 24h:", failed_24h)
print("Avg runtime (completed, last 24h):", avg_runtime_sec)

Notes: - The filter classes mirror the REST payload. See: count_flow_runs and flow_run_history - History responses include count_runs and sum_estimated_run_time so you can compute averages. 3) Alerts and SLAs (Automations) - If you want proactive alerts (e.g., “failure rate > X%” or “runs taking > Y seconds”), set up Automations with Metric Triggers. - Docs: Automations and Metric Triggers 4) External dashboards (Grafana/Datadog/etc.) - Prefect does not expose a built-in Prometheus metrics endpoint today. Common pattern: run a lightweight exporter that queries the Prefect API (like the script above) on an interval and pushes those values to your telemetry stack (Prometheus Pushgateway, Datadog API, etc.). I’m happy to share a sample exporter if you tell me your target system. 5) Health checks - Server health/readiness endpoints: /api/health and /api/readiness

- Worker health checks (useful if you want to also watch your executors): Worker health checks A couple of questions to tailor this: - How do you want to visualize the metrics (Grafana, Datadog, something else)? - What time windows matter (e.g., last 1h vs 24h)? - Since you use Vertex AI work pools, do you also want queue backlog/throughput metrics by work pool or worker health alongside run stats?