https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How to add a start time jitter of sec...
# ask-marvin
r
@Marvin How to add a start time jitter of seconds offsets to a schedule using prefect.yaml and Prefect version 3.7.0 ? The jitter/offset will avoid kubernetes firing up many pods at the same time. Can this only be done with some RRULE wizardry or a concurrency limit? Or, ideally, is there some easy cron job or setting in prefect.yaml for this?
m
✅ thought for 108.8 seconds
Short answer: there’s no built-in 
jitter
or 
offset
field in 
prefect.yaml
schedules in Prefect 3.7.0. So if you want to stagger starts by a few seconds, the main options are: 1. Use 
interval
schedules with different 
anchor_date
values if interval scheduling works for your case 2. Use 
rrule
with 
BYSECOND
 if you need more precise recurring offsets 3. Use concurrency limits if your real goal is “don’t run too many at once” rather than “start at slightly different seconds” I checked Prefect’s schedule support for 3.x, and there isn’t a native 
jitter
, 
delay
, or 
randomize
setting in deployment schedule config. Useful docs: - Schedules - prefect.yaml deployment configuration - Concurrency limits Best practical answer for 
prefect.yaml
 If you’re using interval schedules, the easiest clean solution is to stagger with 
anchor_date
. Example:
Copy code
yaml
deployments:
  - name: my-flow-a
    entrypoint: flows/my_flow.py:my_flow
    schedules:
      - interval: 300
        anchor_date: "2025-01-01T00:00:00Z"

  - name: my-flow-b
    entrypoint: flows/my_flow.py:my_flow
    schedules:
      - interval: 300
        anchor_date: "2025-01-01T00:00:05Z"

  - name: my-flow-c
    entrypoint: flows/my_flow.py:my_flow
    schedules:
      - interval: 300
        anchor_date: "2025-01-01T00:00:10Z"
That gives you a 5-second and 10-second offset while keeping the same 5-minute cadence. If you need cron-like schedules Cron itself in Prefect does not have a simple jitter option. If you need second-level staggering, 
rrule
is the better fit. Example:
Copy code
yaml
deployments:
  - name: my-flow-a
    entrypoint: flows/my_flow.py:my_flow
    schedules:
      - rrule: "FREQ=MINUTELY;INTERVAL=5;BYSECOND=0"

  - name: my-flow-b
    entrypoint: flows/my_flow.py:my_flow
    schedules:
      - rrule: "FREQ=MINUTELY;INTERVAL=5;BYSECOND=5"

  - name: my-flow-c
    entrypoint: flows/my_flow.py:my_flow
    schedules:
      - rrule: "FREQ=MINUTELY;INTERVAL=5;BYSECOND=10"
About concurrency limits Concurrency limits help if the problem is resource contention, but they do not add schedule jitter. They won’t stop Prefect from creating runs at the same scheduled moment; they only control how many run concurrently. So: - Want different start timestamps → use 
interval + anchor_date
or 
rrule
- Want fewer simultaneous executions → use concurrency limits - Want randomized jitter → not currently a built-in schedule feature If all you need is “avoid many Kubernetes pods starting at exactly the same time,” I’d recommend: - 
interval
+ 
anchor_date
 if possible - otherwise 
rrule
with 
BYSECOND
 - optionally combine that with concurrency limits as extra protection If you want, I can help you write the exact 
prefect.yaml
for your current schedule pattern, e.g.: - “every 5 minutes” - “hourly” - “daily at X” - “N deployments each offset by Y seconds”
r
@Marvin Write a Prefect 3.7.0 automation to alert if any currently running flow run is taking more than a threshold, say 50%, longer than recent flow runs of the same flow have taken. Do ensure that the parameters are also the same. And account for changes in the use of cache, i.e. only compare flow runs with cached tasks to historical runs that used the cache, and if the cache is invalidated then compare to historical flow runs where the cache had to be recreated. The goal of those filters is to avoid false positive signals of the alert. I only want truly anomalous long running flows to be alerted.
m
✅ thought for 174.6 seconds
Short answer: not as a native Prefect automation alone. What you want is much more sophisticated than what Prefect 3.7.0 automations can express directly. In particular, a native automation cannot, by itself: - compare a currently running flow run to recent historical runs - require same flow parameters - distinguish cache-hit-ish runs from cache-rebuild-ish runs - compute “alert if current runtime > 1.5x recent baseline” So the right design is: 1. Create a scheduled monitoring flow in Prefect 2. That flow queries currently running flow runs via the Prefect API 3. It builds a filtered historical baseline: - same 
flow_id
- same 
parameters
- same “cache mode/cohort” 4. It alerts only when the running flow exceeds your anomaly threshold That gives you the fidelity you want and avoids false positives. Docs: - Automations - Event triggers - Prefect orchestration client - Caching Important limitation about cache detection Prefect does not provide a first-class built-in automation filter like “only compare runs where tasks used cache hits.” So if cache behavior matters, you should explicitly instrument it yourself. The most reliable approach is to add a flow-level label/parameter/metric that records the cache cohort, for example: - 
cache_mode="warm"
for runs expected to hit cache - 
cache_mode="cold"
for runs expected to rebuild cache - or more specific values like 
dataset_version
, 
cache_epoch
, 
cache_seed
Then your anomaly monitor can compare only within the same cohort. If you don’t explicitly record this, inferring cache behavior from task states is possible but brittle. --- Recommended architecture Use a deployment for a monitoring flow that runs every few minutes, for example every 2 minutes. It should: - fetch all 
RUNNING
flow runs - for each running flow run: - ignore monitoring flows themselves - compute elapsed runtime - fetch recent completed runs of the same flow - filter to: - exact same parameters - same cache cohort - terminal successful runs only - compute a baseline using median or p50/p75 instead of mean - require a minimum sample size - alert if: - elapsed runtime > baseline * 1.5 - and elapsed runtime exceeds some absolute minimum - and optionally only once per flow run I’d strongly recommend median over average, because it’s more robust to outliers. --- Best-practice alert rule Instead of only: - “current runtime > 1.5x historical” use: - minimum historical sample size >= 5 - baseline = median historical duration - alert if current elapsed > max(absolute_floor, baseline * 1.5) For example: - sample size at least 5 - baseline = median of last 20 matching runs - anomaly threshold = 
max(300 seconds, baseline * 1.5)
That avoids noisy alerts on very short jobs. --- How to make cache-aware comparison reliable The cleanest solution is to include a cache cohort in your flow parameters. Example:
Copy code
python
from prefect import flow

@flow
def my_flow(customer_id: str, business_date: str, cache_cohort: str = "warm"):
    ...
Then your deployment or caller sets: - 
cache_cohort="warm"
when cache should already exist - 
cache_cohort="cold"
when cache was invalidated / must be rebuilt Then the monitor simply requires:
Copy code
python
historical.parameters == running.parameters
because the cache cohort is part of the parameters. This is by far the simplest and most robust solution. If you cannot put cache state in parameters, a second-best option is to emit an artifact/event/log at runtime declaring whether the run is warm or cold, then have the monitor read that signal. That is more work. --- Reference implementation
Below is a practical Prefect 3-style monitoring flow. It assumes: - you include 
cache_cohort
in flow parameters - you want to compare only completed runs - you want median baseline - you want Slack/webhook/email/etc. to be plugged in via 
send_alert()
```python from future import annotations from collections import defaultdict from datetime import datetime, timezone, timedelta from statistics import median from typing import Any from prefect import flow, task, get_run_logger from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowFilter, FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, ) from prefect.client.schemas.sorting import FlowRunSort def normalize_parameters(parameters: dict[str, Any] | None) -> tuple: parameters = parameters or {} return tuple(sorted(parameters.items())) async def send_alert(message: str) -> None: # Replace with Slack webhook, email, PagerDuty, etc. print(message) @task async def check_running_flow_runs( lookback_runs: int = 50, min_sample_size: int = 5, multiplier: float = 1.5, absolute_floor_seconds: int = 300, exclude_flow_names: list[str] | None = None, ) -> None: logger = get_run_logger() exclude_flow_names = set(exclude_flow_names or []) now = datetime.now(timezone.utc) async with get_client() as client: running_runs = await client.read_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Running"]) ) ), limit=200, sort=FlowRunSort.START_TIME_ASC, ) logger.info(f"Found {len(running_runs)} running flow runs") # Group running runs by flow_id so we can query history efficiently running_by_flow: dict[str, list] = defaultdict(list) for run in running_runs: if run.flow_id: running_by_flow[str(run.flow_id)].append(run) for flow_id, runs_for_flow in running_by_flow.items(): historical_runs = await client.read_flow_runs( flow_filter=FlowFilter(id={"any_": [flow_id]}), flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Completed"]) ) ), limit=lookback_runs, sort=FlowRunSort.END_TIME_DESC, ) for running in runs_for_flow: if running.start_time is None: continue if getattr(running, "flow_name", None) in exclude_flow_names: continue elapsed_seconds = (now - running.start_time).total_seconds() running_params_norm = normalize_parameters(running.parameters) matching_history = [] for hist in historical_runs: if hist.id == running.id: continue if hist.start_time is None or hist.end_time is None: continue hist_params_norm = normalize_parameters(hist.parameters) # Exact parameter match, including cache_cohort if you add it if hist_params_norm != running_params_norm: continue duration = (hist.end_time - hist.start_time).total_seconds() if duration <= 0: continue matching_history.append(duration) if len(matching_history) < min_sample_size: logger.info( f"Skipping {running.id}: only {len(matching_history)} matching historical runs" ) continue baseline = median(matching_history)
threshold = max(absolute_floor_seconds, baseline * multiplier) logger.info( f"Run {running.id}: elapsed={elapsed_seconds:.1f}s " f"baseline={baseline:.1f}s threshold={threshold:.1f}s " f"matches={len(matching_history)}" ) if elapsed_seconds > threshold: await send_alert( "\n".join( [ f"Anomalous long-running flow detected", f"flow_run_id={running.id}", f"flow_id={running.flow_id}", f"flow_name={getattr(running, 'flow_name', 'unknown')}", f"elapsed_seconds={elapsed_seconds:.1f}", f"baseline_median_seconds={baseline:.1f}", f"threshold_seconds={threshold:.1f}", f"sample_size={len(matching_history)}", f"parameters={running.parameters}", ] ) ) @flow(name="flow-runtime-anomaly-monitor") async def flow_runtime_anomaly_monitor(): await check_running_flow_runs( lookback_runs=50, min_sample_size=5, multiplier=1.5, absolute_floor_seconds=300, exclude_flow_names=["flow-runtime-anomaly-monitor"], ) if name == "__main__": import asyncio asyncio.run(flow_runtime_anomaly_monitor())
Copy code
---

*How to deploy it*

In `prefect.yaml`, you can run this monitor every couple minutes:

```yaml
deployments:
  - name: flow-runtime-anomaly-monitor
    entrypoint: flows/monitor.py:flow_runtime_anomaly_monitor
    schedules:
      - interval: 120
That checks every 2 minutes for overlong running flows. --- Strong recommendation: include cache cohort in parameters Here is an example of the monitored flow:
Copy code
python
from prefect import flow

@flow
def etl_flow(
    customer_id: str,
    business_date: str,
    cache_cohort: str = "warm",
):
    ...
Examples: - normal cached run:
Copy code
json
{
  "customer_id": "acme",
  "business_date": "2026-01-15",
  "cache_cohort": "warm"
}
- cache invalidated / rebuild run:
Copy code
json
{
  "customer_id": "acme",
  "business_date": "2026-01-15",
  "cache_cohort": "cold"
}
Then exact parameter matching already enforces your desired comparison set. --- Why not a native automation? Prefect automations are good for: - state change alerts - event-based notifications - simple threshold triggers They are not good for: - dynamic baselines from matched historical runs - exact parameter-based peer grouping - cache-aware historical cohort analysis So the monitoring flow approach is the correct Prefect-native pattern here. --- If you want even fewer false positives You can tighten the logic further: - compare only to the last N successful runs in the past 14 days - require at least 2 distinct execution dates - use p75 instead of median for more tolerance - suppress duplicate alerts for the same flow run - ignore manually triggered/ad hoc runs - include deployment id/name in the match criteria too For example, many teams compare within: - same flow - same deployment - same parameters - same cache cohort rather than only same flow. --- If you want, I can turn this into a complete production-ready example with: - a Slack webhook action - duplicate-alert suppression - deployment-name matching - exact Prefect 3.7-compatible imports - a full 
prefect.yaml
deployment block
r
@Marvin The cache_cohort as parameter idea sounds good, however when the flow run kicks off it won't know if it's time to invalidate the cache or not. Said another way, the parameter would have to be set dynamically. Not sure if that's even doable and, if so, is probably too complex. Is there an easier way to get the cache chohort comparison goal? Perhaps with the flow run adding a tag at run time if it can detect cache state programmatically?
m
✅ thought for 147.0 seconds
Yes — runtime tagging is not the easiest or best path here. A flow run generally should not depend on mutating its own tags during execution for this kind of monitoring. Even if you update tags through the API, it’s awkward, late, and not great for evaluating a run while it is still running. The easier and more reliable pattern is: infer the cache cohort from the task run states of the current flow run and of the historical candidate runs. In Prefect, a cached task run gets a 
Cached
state; a recomputed one will usually be 
Completed
. That gives you a practical basis for cohorting runs without needing a kickoff parameter. Docs: - Caching - Orchestration client Recommended approach For each flow run, classify it into a cache cohort such as: - 
warm
= enough key tasks already show 
Cached
- 
cold
= those same key tasks are 
Completed
instead of 
Cached
- 
mixed
= mixture of both - 
unknown
= not enough signal yet Then compare the currently running flow only against historical runs in the same cohort. That avoids: - dynamic kickoff params - runtime tag mutation - brittle external bookkeeping --- Important caveat You usually cannot know the cohort immediately at flow start. Why? Because the flow has to reach the relevant cached tasks first. So the anomaly monitor should: - classify a running flow using the task runs observed so far - skip alerting until enough signal exists - only compare once the cohort is confidently determined That is actually a good thing — it reduces false positives. --- Best practice: cohort on a small set of sentinel tasks Don’t try to infer cache behavior from every task. Instead, pick 1–3 important cached tasks that strongly determine total runtime, for example: - expensive extract task - expensive lookup/build task - expensive preprocessing task Then classify a run based on those sentinel tasks only. Example rule: - if all sentinel tasks seen so far are 
Cached
→ 
warm
- if all sentinel tasks seen so far are 
Completed
→ 
cold
- if mixed → 
mixed
- if none have run yet → 
unknown
This is much simpler and usually good enough. --- So: should you use runtime tags? You can update flow run tags later via the API, but I would not recommend using tags as the primary mechanism for this alerting pattern. Why: - tags are not naturally a live execution signal - updating them replaces the tag list, so you must merge carefully - they’re awkward for “currently running” classification - task states already contain the signal you actually care about So the simpler answer is: No, don’t use runtime tags for this. Use observed task run states to derive the cohort. --- Practical design Your monitoring flow should do this: 1. Find currently running flow runs 2. For each running flow run: - fetch its task runs - inspect only sentinel cached tasks - derive current cohort: 
warm
, 
cold
, 
mixed
, or 
unknown
3. If cohort is 
unknown
, skip for now 4. Fetch recent historical completed runs of the same flow 5. For each historical run: - same parameters - same derived cohort from its sentinel tasks 6. Build baseline from those matching runs 7. Alert if current elapsed time > threshold --- Suggested classification logic A simple robust rule: - `warm`: - at least 2 sentinel task runs found - all are 
Cached
- `cold`: - at least 2 sentinel task runs found - all are 
Completed
- `mixed`: - both 
Cached
and 
Completed
among sentinel tasks - `unknown`: - fewer than required sentinel observations You can tune the minimum number of observed sentinel tasks. --- Example implementation Below is a more complete sketch of that approach. ```python from future import annotations from datetime import datetime, timezone from statistics import median from prefect import flow, task, get_run_logger
from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowFilter, FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, TaskRunFilter, ) from prefect.client.schemas.sorting import FlowRunSort SENTINEL_TASK_NAMES = { "load_reference_data", "build_dimension_cache", "expensive_precompute", } def normalize_parameters(parameters): return tuple(sorted((parameters or {}).items())) def classify_cache_cohort(task_runs, sentinel_task_names, min_observations=2): relevant = [ tr for tr in task_runs if tr.name in sentinel_task_names and tr.state and tr.state.name in {"Cached", "Completed"} ] if len(relevant) < min_observations: return "unknown" state_names = {tr.state.name for tr in relevant} if state_names == {"Cached"}: return "warm" if state_names == {"Completed"}: return "cold" if "Cached" in state_names and "Completed" in state_names: return "mixed" return "unknown" async def send_alert(message: str): print(message) @task async def check_running_flow_runs(): logger = get_run_logger() now = datetime.now(timezone.utc) async with get_client() as client: running_runs = await client.read_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Running"]) ) ), limit=100, sort=FlowRunSort.START_TIME_ASC, ) for running in running_runs: if not running.start_time or not running.flow_id: continue running_task_runs = await client.read_task_runs( flow_run_filter=FlowRunFilter(id={"any_": [running.id]}), limit=200, ) running_cohort = classify_cache_cohort( running_task_runs, SENTINEL_TASK_NAMES, min_observations=2, ) if running_cohort == "unknown": logger.info(f"Skipping {running.id}: cache cohort not known yet") continue elapsed_seconds = (now - running.start_time).total_seconds() historical_runs = await client.read_flow_runs( flow_filter=FlowFilter(id={"any_": [running.flow_id]}), flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Completed"]) ) ), limit=50, sort=FlowRunSort.END_TIME_DESC, ) baseline_durations = [] running_params = normalize_parameters(running.parameters) for hist in historical_runs: if hist.id == running.id: continue if not hist.start_time or not hist.end_time: continue if normalize_parameters(hist.parameters) != running_params: continue hist_task_runs = await client.read_task_runs( flow_run_filter=FlowRunFilter(id={"any_": [hist.id]}), limit=200, ) hist_cohort = classify_cache_cohort( hist_task_runs, SENTINEL_TASK_NAMES, min_observations=2, ) if hist_cohort != running_cohort: continue duration = (hist.end_time - hist.start_time).total_seconds() if duration > 0: baseline_durations.append(duration) if len(baseline_durations) < 5: logger.info( f"Skipping {running.id}: insufficient matching history for cohort={running_cohort}" ) continue
baseline = median(baseline_durations) threshold = max(300, baseline * 1.5) if elapsed_seconds > threshold: await send_alert( f"Long-running anomalous flow detected\n" f"flow_run_id={running.id}\n" f"cohort={running_cohort}\n" f"elapsed_seconds={elapsed_seconds:.1f}\n" f"baseline_seconds={baseline:.1f}\n" f"threshold_seconds={threshold:.1f}\n" f"sample_size={len(baseline_durations)}\n" f"parameters={running.parameters}" ) @flow async def flow_runtime_anomaly_monitor(): await check_running_flow_runs() ``` --- Why this is easier than tags Because it avoids needing to: - predict cache mode before kickoff - mutate deployment parameters dynamically - update tags mid-run - manage tag consistency across retries and reruns Instead, it uses what actually happened. That’s usually what you want for anomaly detection. --- One even simpler option If full cache-cohort matching is too much, an easier compromise is: - only compare runs after the sentinel cached tasks have completed - baseline only on historical runs with identical parameters - use a more forgiving threshold, e.g. 
2.0x
instead of 
1.5x
That may already eliminate most false positives without explicitly assigning cohorts. --- My recommendation Use this order of preference: 1. Best: derive cache cohort from sentinel task states 2. Okay: add an explicit cache cohort artifact/log/result during execution and read that later 3. Least recommended: mutate flow run tags dynamically If you want, I can turn this into a production-ready monitor flow with: - correct Prefect 3.7 client imports - Slack webhook alerting - duplicate alert suppression - configurable sentinel task names - a matching 
prefect.yaml
deployment block
r
@Marvin In most flows there is probably only one task that gets cached. So if that is the sentinel task, it defeats the propose as the alert will have to wait until that task completes. i.e. the cached task will be the main contributor to long running flow runs so waiting for it to finish to determine cache state means the alert comes too late.
m
✅ thought for 31.9 seconds
Yes — that’s exactly the problem, and it means post hoc cache detection from task states is too late for your use case if the cached task is itself the long pole. So I’d revise the recommendation: If you need an early-running anomaly alert, you need an early-running cache-cohort signal. If the only strong cache signal appears after the expensive task finishes, then using task states to infer cohort won’t work well for alerting while the run is still in progress. That leaves three realistic options: --- Option 1: determine cache cohort before the expensive task starts This is the best solution if possible. Instead of trying to infer “was cache used?” after the fact, add a cheap preflight task at the beginning of the flow that determines whether the cache is expected to be warm or cold. Examples: - check whether the persisted result artifact/object/file exists - check whether the cache key for the expensive task already exists - check whether the upstream data version changed, implying cache invalidation - check whether the cache expiration window has passed Then store that as a runtime variable in the flow and emit it somewhere the monitor can read. This is much better than waiting for the expensive task to finish. Conceptually:
Copy code
python
from prefect import flow, task

@task
def detect_cache_cohort(...) -> str:
    # cheap logic
    # return "warm" or "cold"
    ...

@task
def expensive_cached_task(...):
    ...

@flow
def my_flow(...):
    cache_cohort = detect_cache_cohort(...)
    expensive_cached_task(...)
This gives you the classification up front. --- Option 2: make cache intent explicit in your own cache key logic Also good, if you control the cache design. If you are using custom caching rules, the easiest reliable approach is often: - define the cache key or invalidation logic yourself - separately define a cheap function that answers: - “would this run hit existing cache?” - or “does this run require rebuilding cache?” That preflight check can run at the top of the flow in milliseconds/seconds, long before the expensive task would finish. So instead of observing whether Prefect did return 
Cached
, you ask ahead of time whether it should. That gives you the cohort early enough for alerting. --- Option 3: maintain your own side-channel run classification If preflight cache detection is not possible from Prefect’s native cache internals, then add your own small metadata side-channel. For example, near the start of the flow: - inspect the underlying storage location that backs the cache - determine 
warm
vs 
cold
- write that result to: - an artifact - a log line with a structured marker - an external store - flow run name/labels are less ideal Then your monitor reads that side-channel. This is still better than runtime tag mutation. --- Why tags still aren’t the best answer Even here, tags are not ideal because: - they’re not a natural “live status” field - changing them during execution is awkward - the monitor would need to rely on successful API updates from the running flow - tags are coarse and not designed as streaming telemetry A small artifact or external metadata record is a cleaner fit. --- Best concrete design for your case I would structure it like this: 1. First task in the flow: - compute 
cache_cohort
cheaply: 
warm
or 
cold
2. Persist that classification immediately: - artifact, log marker, or external metadata store 3. Monitoring flow: - only alerts for running flows once cache cohort is known - compares against historical runs with same parameters and same cohort 4. Alert threshold: - current elapsed > 1.5x median of matched historical runs That gets you early signal without waiting for the expensive cached task to finish. --- What does “cheap preflight cache detection” look like? That depends on how your cache is persisted. If the cached task result is ultimately stored in something like:
- S3 / GCS / Azure blob - local filesystem / shared volume - a result store - a database table / materialized dataset then the preflight task just checks for the existence/freshness of the expected cache object. For example:
Copy code
python
from prefect import task, flow

@task
def detect_cache_cohort(cache_key: str) -> str:
    exists = check_if_cache_object_exists(cache_key)
    return "warm" if exists else "cold"

@task
def expensive_task(...):
    ...

@flow
def my_flow(...):
    cache_cohort = detect_cache_cohort("some-derived-key")
    # optionally emit cache_cohort here
    expensive_task(...)
That’s often much simpler than trying to interrogate Prefect internals. --- So what should you do in practice? Given your constraint, I’d recommend: - Do not rely on detecting 
Cached
after task completion - Do not rely on runtime tag mutation as the primary mechanism - Add a tiny preflight task that predicts cache hit vs rebuild before the expensive task starts That is the only approach that gives you: - early enough classification - accurate cohorting - low false positives - simpler monitoring logic --- If you want, I can help you design that preflight check for your actual cache backend, for example if your cached task results live in: - Prefect result storage - S3 - local disk / PVC - Redis - database tables If you tell me where the cache/result is stored, I can sketch the exact Prefect 3.7 pattern.
n
not sure if im parsing the ask totally, marvin seems to be annoyingly verbose here
can you explain why an automation looking for something beyond a duration doesn't work? are you saying bc sometimes you miss cache, changing the expected duration?
r
@Marvin The preflight task is a good idea. Seems to me you can combine that with your first idea of setting cache_cohort as a parameter. So the preflight task might be the main flow and then send a dynamic preflight informed parameter to the sub flow -- with the subflow being the business logic flow. Can you write that? @Nate the issue is that parameters change for the same flow. Sometimes send a list of dates to process, sometimes it's just today. Or a toggle to say this particular flow run is the longer start of day process and the same business logic can be reused for intraday and those should be quicker -- this can be done with 2 schedules. The cache state is important to compare apples with apples. Sometimes you cache expensive reference data for say, 2 hours, but then that's getting too long for comfort, so happy to re fetch that data several times a day. So you get an up/down step pattern in your flow run times for the same flow.
m
✅ thought for 181.2 seconds
Yes — that’s a good design. A parent/preflight flow can: 1. receive the original business inputs 2. perform a cheap cache-state check up front 3. derive 
cache_cohort
4. call the business logic flow as a subflow, passing: - the original inputs - the derived 
cache_cohort
That gives you exactly what you want: - the business flow run has 
cache_cohort
as a real parameter - your anomaly monitor can compare: - same flow - same business parameters - same 
cache_cohort
- and you don’t need to know the cohort at external trigger time That’s much cleaner than tags. Docs: - Flows - Caching - Run flows from flows Important modeling suggestion Since you mentioned parameters vary a lot — list of dates, today-only, start-of-day toggle, intraday vs longer run — I’d strongly recommend the business flow take explicit normalized parameters plus 
cache_cohort
. For example: - 
processing_dates: list[str]
- 
run_mode: str
such as 
intraday
or 
start_of_day
- 
cache_cohort: str
such as 
warm
or 
cold
Then your monitor can compare exact parameter equality and get good “apples to apples” grouping. --- Example pattern Here is a full example using: - parent preflight flow - subflow for business logic - dynamic 
cache_cohort
- parameters that reflect your business distinctions ```python from future import annotations from datetime import date, datetime, timezone from typing import Literal from prefect import flow, task, get_run_logger RunMode = Literal["intraday", "start_of_day"] CacheCohort = Literal["warm", "cold"] @task def determine_cache_cohort( processing_dates: list[str], run_mode: RunMode, ) -> CacheCohort: """ Cheap preflight check. Replace this logic with your real cache existence/freshness check: - check object exists in S3 / GCS / filesystem - check result storage key exists - check last refresh timestamp - check if cache TTL has expired """ logger = get_run_logger() # Example placeholder logic: # start_of_day runs rebuild cache, intraday reuses it if run_mode == "start_of_day": cohort: CacheCohort = "cold" else: cohort = "warm" logger.info( f"Preflight determined cache_cohort={cohort} " f"for run_mode={run_mode}, processing_dates={processing_dates}" ) return cohort @task def load_reference_data(cache_cohort: CacheCohort) -> dict: """ Example expensive cached task. """ return {"cache_cohort_used": cache_cohort} @task def process_business_data( processing_dates: list[str], run_mode: RunMode, reference_data: dict, ) -> dict: return { "processing_dates": processing_dates, "run_mode": run_mode, "reference_data": reference_data, "status": "ok", } @flow(name="business-logic-flow") def business_logic_flow( processing_dates: list[str], run_mode: RunMode, cache_cohort: CacheCohort, ) -> dict: """ This is the flow your anomaly monitor should compare historically. Important: cache_cohort is now part of the flow run parameters. """ logger = get_run_logger() logger.info( f"Starting business flow with processing_dates={processing_dates}, " f"run_mode={run_mode}, cache_cohort={cache_cohort}" ) reference_data = load_reference_data(cache_cohort) result = process_business_data(processing_dates, run_mode, reference_data) return { "cache_cohort": cache_cohort, "result": result, } @flow(name="preflight-wrapper-flow") def preflight_wrapper_flow( processing_dates: list[str], run_mode: RunMode, ) -> dict: """ Parent flow: 1. determines cache cohort cheaply 2. passes it into the business subflow """ logger = get_run_logger()
cache_cohort = determine_cache_cohort( processing_dates=processing_dates, run_mode=run_mode, ) logger.info(f"Calling subflow with derived cache_cohort={cache_cohort}") return business_logic_flow( processing_dates=processing_dates, run_mode=run_mode, cache_cohort=cache_cohort, ) if name == "__main__": preflight_wrapper_flow( processing_dates=[str(date.today())], run_mode="intraday", )
Copy code
---

*Why this works well for your anomaly use case*

The *business flow run* will have parameters like:

```json
{
  "processing_dates": ["2026-05-13"],
  "run_mode": "intraday",
  "cache_cohort": "warm"
}
or
Copy code
json
{
  "processing_dates": ["2026-05-13"],
  "run_mode": "start_of_day",
  "cache_cohort": "cold"
}
Now your monitor can compare only historical runs with exact same parameters, including cache cohort. That handles your “step pattern” problem much better: - intraday warm-cache runs get compared to intraday warm-cache runs - start-of-day cold-cache runs get compared to start-of-day cold-cache runs - today-only runs don’t get mixed with list-of-dates runs --- How to think about parameter normalization Because you said parameters vary, it’s a good idea to normalize them to avoid accidental mismatch. For example: - always pass 
processing_dates
as a sorted list of ISO strings - always pass 
run_mode
explicitly - maybe also pass 
reference_data_policy
if relevant Example normalization task:
Copy code
python
@task
def normalize_inputs(
    processing_dates: list[str] | None,
    run_mode: RunMode,
) -> tuple[list[str], RunMode]:
    if not processing_dates:
        processing_dates = [date.today().isoformat()]

    normalized_dates = sorted(processing_dates)
    return normalized_dates, run_mode
Then use normalized values in the wrapper before calling the subflow. That helps your monitor do reliable equality matching. --- Recommended deployment shape I would usually deploy the wrapper flow, not the business subflow directly. Why: - wrapper computes the cohort - subflow receives the final parameter set - you only need one scheduled entry point In 
prefect.yaml
that looks like:
Copy code
yaml
deployments:
  - name: business-with-preflight
    entrypoint: flows/my_flow.py:preflight_wrapper_flow
    schedules:
      - cron: "0 */2 * * *"
        parameters:
          processing_dates: []
          run_mode: "intraday"

      - cron: "0 6 * * *"
        parameters:
          processing_dates: []
          run_mode: "start_of_day"
That gives you separate schedules for intraday vs start-of-day while reusing the same business logic. --- One subtle point If your anomaly monitor is looking at flow runs, decide which flow you want to monitor: - the wrapper flow - or the business subflow For your use case, I’d monitor the business subflow, because that’s where the real runtime variance is. The wrapper should stay very short and predictable. --- Optional improvement: derive more than just warm/cold If your cache behavior has more nuance, you can make 
cache_cohort
richer, for example: - 
warm
- 
cold
- 
partial_refresh
- 
expired_2h
- 
expired_manual_invalidation
But I’d start simple with 
warm
/ 
cold
unless you truly need more. --- Best final shape I’d recommend this parameter set on the business flow: - 
processing_dates: list[str]
- 
run_mode: "intraday" | "start_of_day"
- 
cache_cohort: "warm" | "cold"
That should be enough to support robust anomaly comparisons without overcomplicating things. If you want, I can next write: 1. a production-ready version of the wrapper + business flow 2. the monitoring flow that alerts when a running business subflow exceeds 
1.5x
the median of recent matching runs 3. the matching 
prefect.yaml
for both schedules and deployment
n
quick question: i assume you're using open source and not cloud (which has metric triggers for automations)?
r
@Marvin Not really enthused with determine_cache_cohort. The start of day, intraday was just one pattern. Across an enterprise there'll be all sorts of patterns. The code should self discover. Maybe come at the goal from a different direction. Let's only alert for flows that have no caching, i.e. limit the scope of the alert. So the code only has to self discover which flows have code it in that could use a cache. Then we can come back and solve the caching problem at a later time. Can you write that and hopefully it's a dramatically simplified bit of code?
m
✅ thought for 172.5 seconds
r
self hosting
👍 1
m
Yes — that is dramatically simpler, and it’s a good scope cut. Instead of solving “compare warm-cache vs cold-cache runs correctly,” you can do this: Only alert on anomalous long-running flows that show no evidence of task caching. That avoids the hardest part of the problem. --- Key idea Use historical task run metadata to self-discover whether a flow appears to use caching at all. A practical rule is: - if any recent task run for that flow has: - a non-null 
cache_key
, or - state name 
Cached
- then treat that flow as cache-using - and exclude it from this alerting policy Then the alert only applies to flows with no observed caching evidence. This is much simpler and doesn’t require: - preflight wrapper flows - dynamic parameters - runtime tag mutation - cache cohort inference Docs: - Caching - Orchestration client --- Important caveat This is a heuristic, not a perfect static code analysis of “does this flow contain cached tasks.” Prefect’s API exposes enough task-run metadata to detect evidence of caching in observed runs, but not a perfect “this flow definition includes cached task decorators” flag. So the clean rule becomes: - alert only on flows with no recent observed caching evidence That’s operationally good enough in most cases. --- Recommended policy For each currently running flow run: 1. get recent completed runs of the same flow with the same parameters 2. build a duration baseline 3. separately determine whether this flow has any recent caching evidence 4. if caching evidence exists, skip alerting 5. if no caching evidence exists, alert when current elapsed > threshold You can determine caching evidence from recent historical task runs for that flow. --- Simplified implementation Here’s a monitoring flow that does that. ```python from future import annotations from datetime import datetime, timezone from statistics import median from typing import Any from prefect import flow, task, get_run_logger from prefect.client.orchestration import get_client from prefect.client.schemas.filters import ( FlowFilter, FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, ) from prefect.client.schemas.sorting import FlowRunSort def normalize_parameters(parameters: dict[str, Any] | None) -> tuple: return tuple(sorted((parameters or {}).items())) async def send_alert(message: str) -> None: # Replace with Slack webhook / email / PagerDuty / etc. print(message) async def flow_has_recent_caching_evidence(client, flow_id: str, recent_flow_run_ids: list[str]) -> bool: """ Return True if any recent task run for these flow runs shows evidence of caching. Evidence: - task_run.cache_key is not None - task_run.state_name == 'Cached' """ for flow_run_id in recent_flow_run_ids: task_runs = await client.read_task_runs( flow_run_filter={"id": {"any_": [flow_run_id]}}, limit=200, ) for tr in task_runs: state_name = getattr(tr, "state_name", None) cache_key = getattr(tr, "cache_key", None) if cache_key is not None or state_name == "Cached": return True return False @task async def check_running_flows_without_caching( lookback_runs: int = 20, min_sample_size: int = 5, multiplier: float = 1.5, absolute_floor_seconds: int = 300, exclude_flow_names: list[str] | None = None, ) -> None: logger = get_run_logger() exclude_flow_names = set(exclude_flow_names or []) now = datetime.now(timezone.utc) async with get_client() as client: running_runs = await client.read_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Running"]) )
), limit=100, sort=FlowRunSort.START_TIME_ASC, ) logger.info(f"Found {len(running_runs)} running flow runs") for running in running_runs: if not running.flow_id or not running.start_time: continue if getattr(running, "flow_name", None) in exclude_flow_names: continue historical_runs = await client.read_flow_runs( flow_filter=FlowFilter(id={"any_": [str(running.flow_id)]}), flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Completed"]) ) ), limit=lookback_runs, sort=FlowRunSort.END_TIME_DESC, ) matching_history = [] recent_flow_run_ids = [] running_params = normalize_parameters(running.parameters) for hist in historical_runs: recent_flow_run_ids.append(str(hist.id)) if hist.id == running.id: continue if not hist.start_time or not hist.end_time: continue if normalize_parameters(hist.parameters) != running_params: continue duration = (hist.end_time - hist.start_time).total_seconds() if duration > 0: matching_history.append(duration) if len(matching_history) < min_sample_size: logger.info( f"Skipping {running.id}: only {len(matching_history)} matching historical runs" ) continue has_caching = await flow_has_recent_caching_evidence( client=client, flow_id=str(running.flow_id), recent_flow_run_ids=recent_flow_run_ids, ) if has_caching: logger.info( f"Skipping {running.id}: flow has recent caching evidence" ) continue elapsed_seconds = (now - running.start_time).total_seconds() baseline = median(matching_history) threshold = max(absolute_floor_seconds, baseline * multiplier) logger.info( f"Run {running.id}: elapsed={elapsed_seconds:.1f}s " f"baseline={baseline:.1f}s threshold={threshold:.1f}s " f"matches={len(matching_history)} caching={has_caching}" ) if elapsed_seconds > threshold: await send_alert( "\n".join( [ "Anomalous long-running non-cached flow detected", f"flow_run_id={running.id}", f"flow_id={running.flow_id}", f"flow_name={getattr(running, 'flow_name', 'unknown')}", f"elapsed_seconds={elapsed_seconds:.1f}", f"baseline_median_seconds={baseline:.1f}", f"threshold_seconds={threshold:.1f}", f"sample_size={len(matching_history)}", f"parameters={running.parameters}", ] ) ) @flow(name="non-cached-flow-runtime-anomaly-monitor") async def non_cached_flow_runtime_anomaly_monitor(): await check_running_flows_without_caching( lookback_runs=20, min_sample_size=5, multiplier=1.5, absolute_floor_seconds=300, exclude_flow_names=["non-cached-flow-runtime-anomaly-monitor"], ) if name == "__main__": import asyncio asyncio.run(non_cached_flow_runtime_anomaly_monitor()) ``` --- How this works This monitor: - finds currently running flows - finds recent completed runs of the same flow
- filters to exact parameter matches - computes a median duration baseline - checks whether the flow has any recent evidence of caching - skips alerting if it does - alerts only for non-cached flows that are running unusually long So it gives you a much smaller and cleaner first version of the alert. --- Why this is simpler Because it replaces: - “detect cache cohort correctly for each run” - “compare only warm to warm / cold to cold” - “handle enterprise-specific invalidation patterns” with: - “if the flow appears cache-aware at all, don’t include it in this policy” That’s a very reasonable first slice. --- One suggested refinement You might want to require multiple pieces of caching evidence before excluding a flow, for example: - at least 2 recent task runs with 
cache_key
- or at least 1 
Cached
state That helps avoid weird edge cases. For example, you could replace:
Copy code
python
if cache_key is not None or state_name == "Cached":
    return True
with a small evidence counter. --- One more caveat about “same parameters” If your parameters include lists/dicts, a simple tuple-sort normalization may fail if nested structures are present. If you want, I can improve that with JSON-stable normalization so this works reliably for: - lists of dates - nested dict configs - booleans/toggles - optional fields That’s probably worth doing for your use case. --- If you want, I can next give you a cleaned-up production version with: - robust parameter normalization - better caching-evidence scoring - duplicate alert suppression - Slack webhook notification - and a matching 
prefect.yaml
deployment block
r
@Marvin Cheers! I'll give it a go and see if it works. Find and play yourself an appropriate congratulatory song -- at least imagine it's playing even though you're not allowed to play songs. Don't tell me the name of the song -- just keep it for yourself as your little secret pleasure.
m
✅ thought for 4.6 seconds
Deal 😄 I’ve got one in mind and it is absolutely playing in my imaginary victory playlist. Good luck with the implementation — and if the API filters or task-run metadata get fiddly, come back with the error/output and I’ll help tighten it up.
r
@Marvin flow_has_recent_caching_evidence never uses the flow_id parameter. Is that an oversight and if so can you fix just that? Also probably need to come up with a less ambitious victory song.
😄 1
m
✅ thought for 14.2 seconds
Yes — that was an oversight. In that version, 
flow_id
was passed but never used. Since the function was actually checking only the supplied 
recent_flow_run_ids
, the cleanest fix is simply to remove the unused parameter. Here’s the corrected version:
Copy code
python
async def flow_has_recent_caching_evidence(client, recent_flow_run_ids: list[str]) -> bool:
    """
    Return True if any recent task run for these flow runs shows evidence of caching.
    Evidence:
    - task_run.cache_key is not None
    - task_run.state_name == 'Cached'
    """
    for flow_run_id in recent_flow_run_ids:
        task_runs = await client.read_task_runs(
            flow_run_filter={"id": {"any_": [flow_run_id]}},
            limit=200,
        )

        for tr in task_runs:
            state_name = getattr(tr, "state_name", None)
            cache_key = getattr(tr, "cache_key", None)

            if cache_key is not None or state_name == "Cached":
                return True

    return False
And update the call site to:
Copy code
python
has_caching = await flow_has_recent_caching_evidence(
    client=client,
    recent_flow_run_ids=recent_flow_run_ids,
)
If you want 
flow_id
used, then it should be part of a different implementation strategy, e.g. querying task runs across recent runs of that flow directly. But for the code I gave, removing it is the right fix. And yes, agreed — probably best to downgrade from “triumphant stadium encore” to something more like “quietly satisfied spreadsheet jazz.”
Open in Slack
PreviousNext
Powered by