start_client_metrics_server()

PREFECT_CLIENT_METRICS_ENABLED

PREFECT_CLIENT_METRICS_ENABLED

start_client_metrics_server()

python
from prefect import flow
from prefect.utilities.services import start_client_metrics_server

# Start the metrics server
start_client_metrics_server()

@flow
def my_flow():
    # Your flow code here
    pass

PREFECT_CLIENT_METRICS_ENABLED=true

PREFECT_CLIENT_METRICS_PORT

http://<pod-ip>:<metrics-port>/metrics

PREFECT_CLIENT_METRICS_ENABLED

PREFECT_CLIENT_METRICS_ENABLED=true

start_client_metrics_server()

yaml
apiVersion: <http://prefect.io/v1alpha1|prefect.io/v1alpha1>
kind: PrefectDeployment
metadata:
  name: my-flow
spec:
  flow:
    name: my-flow
  env:
    - name: PREFECT_CLIENT_METRICS_ENABLED
      value: "true"
    - name: PREFECT_CLIENT_METRICS_PORT
      value: "9090"  # optional, defaults to 9090

PREFECT_CLIENT_METRICS_PORT

PREFECT_CLIENT_METRICS_PORT

PREFECT_LOGGING_ORION_ENABLED

PREFECT_LOGGING_TO_API_ENABLED

PREFECT_CLIENT_METRICS_ENABLED=true

start_client_metrics_server()

PREFECT_CLIENT_METRICS_PORT

yaml
# NetworkPolicy (if you're using network policies)
apiVersion: <http://networking.k8s.io/v1|networking.k8s.io/v1>
kind: NetworkPolicy
metadata:
  name: allow-prometheus-scraping
spec:
  podSelector:
    matchLabels:
      # match your Prefect flow pod labels
      app: your-prefect-flow
  policyTypes:
  - Ingress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          # adjust based on where your Prometheus is running
          <http://kubernetes.io/metadata.name|kubernetes.io/metadata.name>: monitoring
    ports:
    - protocol: TCP
      port: 4201

---
# PodMonitor
apiVersion: <http://monitoring.coreos.com/v1|monitoring.coreos.com/v1>
kind: PodMonitor
metadata:
  name: prefect-flow-monitor
spec:
  selector:
    matchLabels:
      # match your Prefect flow pod labels
      app: your-prefect-flow
  podMetricsEndpoints:
  - port: metrics  # or the name you've given to the port in your pod spec
    path: /metrics

yaml
spec:
  containers:
  - name: flow
    ports:
    - name: metrics
      containerPort: 4201

start_client_metrics_server()

python
from prefect import flow
from prefect.utilities.services import start_client_metrics_server

start_client_metrics_server()

@flow
def my_flow():
    # your flow code here
    pass

yaml
apiVersion: <http://monitoring.coreos.com/v1|monitoring.coreos.com/v1>
kind: PodMonitor
metadata:
  name: prefect-flow-monitor
spec:
  selector:
    matchLabels:
      # whatever labels identify your Prefect flow run pods
      <http://prefect.io/flow-run-name|prefect.io/flow-run-name>: ""
  namespaceSelector:
    matchNames:
      - your-prefect-jobs-namespace
  podMetricsEndpoints:
  - port: metrics  # or the actual port number 4201
    path: /metrics

late_runs_count

start_client_metrics_server()

prefect_info_work_queues

late_runs_count

start_client_metrics_server()

end_time

None

prefect_info_flow_runs

start_time

end_time

start_time

None

end_time

/flow-runs/{id}

bash
curl http://<your-exporter-host>:9090/metrics | grep prefect_info_flow_runs

prefect_info_flow_runs

bash
# First find the pod name
kubectl get pods -n <namespace> | grep prometheus-prefect-exporter

# Then curl the metrics endpoint from the pod
kubectl exec -n <namespace> <exporter-pod-name> -- curl localhost:9090/metrics | grep prefect_info_flow_runs

prefect_info_flow_runs{deployment_id="...",flow_id="...",flow_name="...",flow_run_id="...",flow_run_name="...",start_time="...",end_time="...",state="..."} 1

end_time

None

end_time

flow_run_name="dashing-hummingbird",
start_time="2025-07-01T04:36:57.346805Z",
end_time="2025-07-01T04:38:08.870199Z",
state_name="Completed"

yaml
apiVersion: <http://monitoring.coreos.com/v1|monitoring.coreos.com/v1>
kind: PodMonitor
metadata:
  name: prefect-flow-monitor
spec:
  podMetricsEndpoints:
  - interval: 5s  # More frequent scraping
    port: metrics
    path: /metrics

bash
   # Get the exporter logs
   kubectl logs <prometheus-prefect-exporter-pod> -n <namespace>

python
   from prefect.client import get_client
   
   async with get_client() as client:
       flow_run = await client.read_flow_run("<flow-run-id>")
       print(flow_run.end_time)

yaml
   # In your prometheus-prefect-exporter deployment
   env:
     - name: PREFECT_API_POLL_INTERVAL
       value: "10"  # Decrease the interval (in seconds) to catch more updates

yaml
   apiVersion: <http://monitoring.coreos.com/v1|monitoring.coreos.com/v1>
   kind: PodMonitor
   spec:
     podMetricsEndpoints:
     - interval: 5s  # More frequent scraping
       port: metrics
       path: /metrics

bash
   kubectl logs -f <prometheus-prefect-exporter-pod> -n <namespace>

PREFECT_API_POLL_INTERVAL

after_data

%Y-%m-%dT%H:%M:%S.%fZ

after_data

python
offset_minutes = 60
# At 2:00 PM, this would mean:
after_data = 1:00 PM
# Your query window would be: 1:00 PM to 2:00 PM

python
# If current time is 2:00 PM
after_data = 2:00 PM  # (now - 0 minutes)
# Query window: "Get all flow runs from 2:00 PM onwards"

python
# If current time is 2:00 PM
after_data = 12:00 PM  # (now - 120 minutes)
# Query window: "Get all flow runs from 12:00 PM onwards"

2:00 PM: Flow A starts
2:01 PM: Prometheus scrapes (offset=0)
         → Sees Flow A in "Running" state
2:02 PM: Flow A completes
2:02 PM: Prometheus scrapes (offset=0)
         → Sees Flow A in "Completed" state

SCRAPE_INTERVAL_SECONDS=120

2:00 PM: Scrape happens
         Query window: "Get flows from 2:00 PM onwards"
2:01 PM: Flow A runs and completes
2:02 PM: Flow B runs and completes
2:02 PM: Scrape happens
         Query window: "Get flows from 2:02 PM onwards"
         → MISSES both Flow A and Flow B!

SCRAPE_INTERVAL_SECONDS=120

2:00 PM: Scrape happens
         Query window: "Get flows from 1:57 PM onwards"
2:01 PM: Flow A runs and completes
2:02 PM: Flow B runs and completes
2:02 PM: Scrape happens
         Query window: "Get flows from 1:59 PM onwards"
         → CAPTURES both Flow A and Flow B

OFFSET_MINUTES should be > (SCRAPE_INTERVAL_SECONDS / 60)

SCRAPE_INTERVAL_SECONDS = 5
OFFSET_MINUTES = 1

SCRAPE_INTERVAL_SECONDS = 15  # or 30
OFFSET_MINUTES = 1

scrape_duration_seconds

while True

time.sleep()

python
while True:
    time.sleep(scrape_interval_seconds)
    # Scrape starts here

Time 0:00 - Scrape A starts
Time 0:05 - New scrape B starts (even if A is still running!)
Time 0:10 - New scrape C starts (even if A or B are still running!)

SCRAPE_INTERVAL_SECONDS = 30  # or higher
   OFFSET_MINUTES = 2  # provides adequate overlap

while True

time.sleep()

while True

python
   while True:
       time.sleep(scrape_interval_seconds)  # Thread blocks here
       # Scrape code runs here
       # When done, loops back to sleep

time.sleep()

0:00:00 - Loop starts
0:00:00 - Hits time.sleep(5) → Thread blocks for 5 seconds
0:00:05 - Sleep ends
0:00:05 - Scrape starts
0:00:07 - Scrape finishes (took 2 seconds)
0:00:07 - Hits time.sleep(5) again → Blocks for 5 seconds
0:00:12 - Sleep ends
0:00:12 - Next scrape starts

asyncio

apscheduler

scrape_interval_seconds

prometheus_client

python
REGISTRY.register(metrics)

metrics

PrefectMetrics

python
start_http_server(metrics_port, metrics_addr)

while True

python
   # This starts a web server in a separate thread
   start_http_server(metrics_port, metrics_addr)

prometheus_client

collect()

while True

while True

python
   while True:
       time.sleep(scrape_interval_seconds)

signal.pause()

1. Program starts
2. Metrics collector is registered
3. HTTP server starts in background thread
4. Main thread enters infinite sleep loop

When Prometheus scrapes:
→ HTTP request comes to background server
→ Server triggers collect() method
→ Fresh metrics are gathered and returned

(All this happens independently of the sleep loop!)

scrape_interval_seconds

collect()

PrefectMetrics

SCRAPE_INTERVAL_SECONDS

python
import signal

def metrics():
    # ... same setup code ...
    
    # Start the HTTP server
    start_http_server(metrics_port, metrics_addr)
    <http://logger.info|logger.info>(f"Exporter listening on {metrics_addr}:{metrics_port}")
    
    # Wait for signals (more proper than while True/sleep)
    signal.pause()

while True

Clarify purpose of SCRAPE_INTERVAL_SECONDS and its relationship to Prometheus scraping

## Background
I noticed some historical changes regarding scrape intervals in the exporter:

1. PR <https://github.com/PrefectHQ/prometheus-prefect-exporter/commit/7a3d105235c9ba6020df8f111bc2cf88905a7ffa|7a3d105> removed `polling_interval_seconds` from PrefectMetrics
2. PR <https://github.com/PrefectHQ/prometheus-prefect-exporter/pull/57|#57> later added `SCRAPE_INTERVAL_SECONDS`

## Current Implementation
The current code in main.py uses `SCRAPE_INTERVAL_SECONDS` in a while True loop:

```python
while True:
    time.sleep(scrape_interval_seconds)

polling_interval_seconds

SCRAPE_INTERVAL_SECONDS

SCRAPE_INTERVAL_SECONDS

SCRAPE_INTERVAL_SECONDS

prefect_info_flow_runs

OFFSET_MINUTES

python
   from prefect.client import get_client
   
   async with get_client() as client:
       flow_run = await client.read_flow_run(flow_run_id)
       print(f"State: {flow_run.state.type}")
       print(f"End Time: {flow_run.end_time}")

OFFSET_MINUTES

OFFSET_MINUTES

SCRAPE_INTERVAL_SECONDS=5

OFFSET_MINUTES=1

[now - 1 minute] <-------- 1 minute window --------> [now]
         ^                           ^                    ^
   Window start              Flow completes           Scrape time

Time 0s: Flow reaches "Completed" state
   Time 0.1s: State recorded
   Time 0.2s: Prometheus scrapes (sees Completed but no end_time yet)
   Time 0.3s: end_time recorded
   Time 5s: Next scrape (might miss it if flow is now outside window)

OFFSET_MINUTES=2
   SCRAPE_INTERVAL_SECONDS=15

OFFSET_MINUTES=1
   SCRAPE_INTERVAL_SECONDS=30

SCRAPE_INTERVAL_SECONDS

python
# From main.py
while True:
    time.sleep(scrape_interval_seconds)

SCRAPE_INTERVAL_SECONDS

yaml
scrape_configs:
  - job_name: 'prefect'
    scrape_interval: 15s  # THIS controls when Prometheus collects metrics
    static_configs:
      - targets: ['localhost:8000']

[now - OFFSET_MINUTES]

[now]

OFFSET_MINUTES

SCRAPE_INTERVAL_SECONDS

SCRAPE_INTERVAL_SECONDS

OFFSET_MINUTES=60:
- 1 flow per minute = ~60 flows per query
- Each flow has state, metadata, etc.

OFFSET_MINUTES=10:
- 1 flow per minute = ~10 flows per query
- 6x less data per query

prometheus/prefect_info_flow_runs/gauge

[now - OFFSET_MINUTES] <-------- window --------> [now]
                                                      |
                                              Only flows in this
                                              window are reported

10:00 AM: Flow A completes
   10:05 AM: You see Flow A in metrics (within window)
   10:11 AM: Flow A disappears (outside window)

yaml
   rules:
     - record: flow_runs_completed_total
       expr: sum(prefect_info_flow_runs)

prefect_info_flow_runs

Flow Run ID: ABC123
   Metrics show:
   - Row 1: state=Running, start_time=present, end_time=null
   - Row 2: state=Completed, start_time=present, end_time=present

Flow Run 1: Clean completion
   - State transition: Running → Completed
   - end_time properly recorded
   
   Flow Run 2: Irregular completion
   - State transition: Running → Completed (but interrupted)
   - end_time might not be recorded properly

python
   from prefect.client import get_client
   
   async with get_client() as client:
       # Get flow runs for your flows
       flow_runs = await client.read_flow_runs(
           flow_name={"any_": ["Echo Flow", "Nucleus Ingestion"]}
       )
       for run in flow_runs:
           print(f"Flow: {run.name}")
           print(f"State: {run.state.type}")
           print(f"End Time: {run.end_time}")
           print("---")

python
   # In your flow code
   @flow(
       name="Echo Flow",
       retries=0,  # Prevent automatic retries
       persist_result=True  # Ensure state transitions are recorded
   )

python
from prefect.client import get_client
from datetime import datetime, timezone, timedelta

async def check_flow_run_times(flow_name):
    async with get_client() as client:
        # Get recent flow runs for the specified flow
        flow_runs = await client.read_flow_runs(
            flow_name={"any_": [flow_name]},
            sort="-start_time",
            limit=5
        )
        
        for run in flow_runs:
            print(f"\nFlow Run ID: {run.id}")
            print(f"State: {run.state.type}")
            print(f"Start Time: {run.start_time}")
            print(f"End Time: {run.end_time}")
            if run.end_time and run.start_time:
                duration = run.end_time - run.start_time
                print(f"Duration: {duration}")

python
from prefect.client import get_client

async def show_actual_times(flow_run_id):
    async with get_client() as client:
        run = await client.read_flow_run(flow_run_id)
        print(f"Start Time: {run.start_time}")
        print(f"End Time: {run.end_time}")  # This exists in the API even though UI shows duration
        print(f"Duration shown in UI: {run.end_time - run.start_time if run.end_time and run.start_time else 'N/A'}")

json
{
  "start_time": "2025-07-01T19:01:46.665632Z",
  "end_time": "2025-07-01T19:03:04.016106Z"
}

Flow end time: 19:03:04
   Current time: [your current time]
   OFFSET_MINUTES window: [current time - OFFSET_MINUTES] to [current time]

prefect_info_flow_runs{
     flow_name="tangerine-anaconda",
     flow_run_id="5db158c6-cc71-442b-8645-ba8e5b81a69c",
     state="completed"
   }

bash
   curl -s <http://localhost:8000/metrics> | grep prefect_info_flow_runs

bash
curl -s <http://prefect.com/api/flow_runs/5db158c6-cc71-442b-8645-ba8e5b81a69c/states> | jq '.'

json
{
  "state_type": "COMPLETED",
  "start_time": "2025-07-01T19:01:46.665632Z",
  "end_time": "2025-07-01T19:03:04.016106Z",
  "state": {
    "type": "COMPLETED",
    "timestamp": "2025-07-01T19:03:04.016106Z",
    "state_details": {
      "untrackable_result": true,
      "transition_id": "88f3a170-4d10-43c3-987c-cabdbaa9d965"
    }
  }
}

end_time

timestamp

transition_id

"untrackable_result": true

bash
   # Get the current metrics
   curl -s <http://localhost:8000/metrics> | grep "prefect_info_flow_runs.*5db158c6-cc71-442b-8645-ba8e5b81a69c"

bash
   # Show exporter environment variables
   env | grep "OFFSET_MINUTES"

bash
   curl -s <http://localhost:8000/metrics> | grep "tangerine-anaconda"

untrackable_result: true

untrackable_result: true

untrackable_result: true