Short answer: In Prefect 3.x, the “Late” minutes shown for a flow run are derived from a computed field (

estimated_start_time_delta

) that measures how far past the run’s expected start time it is. What’s counted and when it starts - For a run that has NOT started yet (e.g., Scheduled/Pending): - Late = now() − expected_start_time - Counting begins exactly at

expected_start_time

(UTC). The value increases over time until the run starts. - For a run that HAS started: - Late = start_time − expected_start_time - This becomes a fixed delay once the run starts. - For terminal runs (Completed/Failed/etc.): - Late = 0 Important distinctions - The “Late” minutes display is based on

expected_start_time

(the time the run was expected to start). - The background “late runs” service that marks runs with a “Late” scheduled state uses

next_scheduled_start_time

and a small tolerance (default ~15s). You can see “Late” minutes even if the state has not been flipped to “Late” (e.g., if services are disabled). - All timestamps are handled in UTC. Code references (Prefect 3.6.5) - Computation of lateness (hybrid property on FlowRun): server/database/orm_models.py - Late run marking service: server/services/late_runs.py - Service settings (tolerances/intervals): server/settings/models/server/services.py - UI flow run endpoints that surface the value: server/api/ui/flow_runs.py Example - Scheduled for 100000Z, still not started at 100300Z → Late ≈ 3 minutes. - Scheduled for 100000Z, actually started at 100500Z → Late ≈ 5 minutes (fixed once started).

Yes—two good options: 1) From Prefect (recommended if the agent could reach your API) - UI: Open the flow run’s page → Logs tab. - CLI:

prefect flow-run logs <FLOW_RUN_ID>

Handy helpers:

# Find recent runs, then copy the ID
  prefect flow-run ls --limit 20
  # Tail just the latest 50
  prefect flow-run logs <FLOW_RUN_ID> --tail -n 50

- API / Python client:

from prefect.client.orchestration import get_client
  from prefect.client.schemas.filters import LogFilter, LogFilterFlowRunId
  from prefect.client.schemas.sorting import LogSort

  flow_run_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

  async with get_client() as client:
      logs = await client.read_logs(
          log_filter=LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id])),
          sort=LogSort.TIMESTAMP_ASC,
          limit=200,
      )
      for log in logs:
          print(f"{log.timestamp} [{log.level}] {log.message}")

References: - Logs API: server/api/logs.py - Client methods: client/orchestration/_logs/client.py 2) Directly from Kubernetes (raw container logs) - The Kubernetes worker labels pods/jobs with the flow run ID. You can fetch logs by label:

# Show logs for all containers in the pod(s) created for this flow run
  kubectl logs -n <NAMESPACE> -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> --all-containers=true

  # Stream logs
  kubectl logs -n <NAMESPACE> -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -f --all-containers=true

  # If you prefer targeting the Job:
  kubectl logs -n <NAMESPACE> job/$(kubectl get job -n <NAMESPACE> -l <http://prefect.io/flow-run-id=<FLOW_RUN_ID|prefect.io/flow-run-id=<FLOW_RUN_ID>> -o jsonpath='{.items[0].metadata.name}') --all-containers=true

Labeling reference: prefect-kubernetes/worker.py Notes - If the pod can’t reach your Prefect API (env, network, or credentials), Prefect-side logs may be missing—use kubectl logs. - If your cluster cleans up completed pods (TTL/jobs GC), Kubernetes logs may be gone—Prefect logs (via API/CLI/UI) will still be available unless you’ve configured retention to purge them.

@Marvin I can see some logs in Prefect, but partially. Do you know why? The pods should be able to reach the Prefect API

Likely causes (and quick fixes) for “partial” logs in K8s, even when the pod can reach your Prefect API: - Final logs lost at shutdown (most common) - Prefect batches logs and flushes them asynchronously; on process exit it only waits ~5s to flush. If your run ends quickly or there’s any API slowness, the last few logs may never be sent. - Try: - Set

PREFECT_LOGGING_TO_API_BATCH_INTERVAL=1.0

(or 0.5) so batches flush more often - Reduce batch size to keep requests small:

PREFECT_LOGGING_TO_API_BATCH_SIZE=2000000

- Increase pod

terminationGracePeriodSeconds

(e.g., 60) - Explicitly flush at the end of your flow:

import logging
      from prefect import flow, get_run_logger
      from prefect.logging.handlers import APILogHandler

      @flow
      def my_flow():
          log = get_run_logger()
          try:
              ...
          finally:
              for h in logging.getLogger("prefect.flow_runs").handlers:
                  if isinstance(h, APILogHandler):
                      h.flush()
              for h in logging.getLogger("prefect.task_runs").handlers:
                  if isinstance(h, APILogHandler):
                      h.flush()

- Proxy/LB dropping log batches - Prefect’s default batch can be ~4MB; some ingress/proxies default to 1–2MB bodies or have short timeouts. When a batch fails, it’s dropped (no retry), which looks like “missing” chunks. - Fix: lower

PREFECT_LOGGING_TO_API_BATCH_SIZE

(e.g., 1–2MB), and raise proxy body limits/timeouts if you control the LB (e.g., Nginx

client_max_body_size

). - Logs emitted outside flow/task context - Only logs with a `flow_run_id`/`task_run_id` are written to the API.

print()

or stdlib logging in a subprocess won’t have context and are dropped. - Fix: always use

get_run_logger()

inside flows/tasks; for subprocesses, emit logs via a small helper that includes

flow_run_id

, or send them to stdout and use kubectl to view. - Log level filtering - By default Prefect logs at INFO; if “missing” logs are DEBUG, they won't appear unless

PREFECT_LOGGING_LEVEL=DEBUG

(or you raise the level on your logger). - Oversized messages truncated - Very large single log lines are truncated to

PREFECT_LOGGING_TO_API_MAX_LOG_SIZE

(default ~1MB). That can look “partial” if your messages are huge. Quick diagnostics - Are the missing lines only the last N lines? If yes, it’s almost certainly batching/flush timing; apply the first set of fixes. - Enable internal logging to see batching behavior/errors:

PREFECT_LOGGING_INTERNAL_LEVEL=DEBUG

Look for lines like “processing batch of size …” or “— Error logging to API —”. - Confirm env in the pod:

echo $PREFECT_LOGGING_TO_API_ENABLED
  echo $PREFECT_LOGGING_TO_API_BATCH_INTERVAL
  echo $PREFECT_LOGGING_TO_API_BATCH_SIZE

- If you’re behind a proxy/LB, do you have a body size limit or short upstream timeout? Relevant code (Prefect 3.6.5) - API log handler and batching (flush timeout, no-retry on failure): prefect/logging/handlers.py - Logging settings (batch size/interval, max log size): settings/models/logging.py - Logs API endpoint: server/api/logs.py If you can share: - Whether the “partial” gap is only at the end - Whether you’re using any prints/subprocess logging - Whether there’s an ingress/proxy in front of your 25 API instances …I can tailor the exact settings and steps for your setup.

Short answer: Not out of the box with the Kubernetes work pool. Unlike Cloud Run V2, the Kubernetes worker does not surface a clickable “logs URL” in the UI. It does set the flow run’s

infrastructure_pid

(formatted like

namespace:job_name

), which you can use to construct your own link. Ways to get a link today - Build a link from

infrastructure_pid

- Read

flow_run.infrastructure_pid

and format a URL to your cluster’s dashboard/logs view (e.g., Lens, Kubernetes Dashboard, Log explorer like Loki/Elastic). - Example pattern:

ns, job = flow_run.infrastructure_pid.split(":", 1)
    logs_url = f"<https://your-dashboard.example.com/namespaces/{ns}/jobs/{job}>"

- Put the URL in the state message - When you create the job (or when your flow starts), include a message with the prebuilt logs URL so it’s visible in the run’s UI:

from prefect import get_client
    from prefect.states import Running

    async with get_client() as client:
        await client.set_flow_run_state(
            flow_run_id,
            state=Running(message=f"View K8s job logs: https://.../{namespace}/{job_name}")
        )

- Emit an event with related metadata - Advanced: emit a Prefect event that includes a logs URL so you can reference it via your observability tooling. - Customize the worker - Subclass the Kubernetes worker to construct a logs URL at job creation time and either: - set a Running state message with the URL, or - encode extra info into

infrastructure_pid

(e.g.,

namespace:job_name:https://...

), which you can parse later. References - Flow run infrastructure identifier is persisted as

infrastructure_pid

and is set by the Kubernetes worker. You can consume it via the API/SDK to generate links: prefect-kubernetes/worker.py If you tell me which K8s log UI you use (Kubernetes Dashboard, Lens, Grafana Loki, Elastic/Kibana, Datadog, etc.), I can provide a ready-to-drop snippet that builds the exact link format and writes it into the flow’s state so it shows up in the UI.

Short answer: It means the flow’s execution process received SIGTERM, and Prefect chose to reschedule the run instead of letting it crash. In Kubernetes, SIGTERM commonly comes from pod eviction (node drain/scale-down), job deletion, or a manual terminate—so yes, an underlying node going down or being drained is a likely cause. What actually happens - When the flow-run process gets SIGTERM, Prefect can switch the run to an AwaitingRetry (Scheduled) state and exit, allowing the work pool to submit it again. - The Kubernetes work pool auto-configures this behavior: - If your Job’s

backoffLimit=0

, the worker sets

PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR=reschedule

inside the pod so a SIGTERM triggers resubmission. - If

backoffLimit>0

, Kubernetes will retry the Job itself; Prefect disables the “reschedule on SIGTERM” to avoid duplicate execution. - If reschedule-on-SIGTERM is not enabled, a SIGTERM generally leads to the run being marked Crashed instead. Useful checks - Inspect your Job spec: -

backoffLimit=0

→ Prefect likely enabled reschedule-on-SIGTERM -

backoffLimit>0

→ Kubernetes is handling retries; Prefect won’t reschedule on SIGTERM - Confirm the env var inside the flow pod:

echo $PREFECT_FLOW_RUN_EXECUTE_SIGTERM_BEHAVIOR

Expect

reschedule

if Prefect is handling resubmission. - Look at the flow run’s state transitions: you should see it move to AwaitingRetry (Scheduled), then queued/running again. Code references (Prefect 3.x) - Runner reschedule log and logic: prefect/runner/runner.py - SIGTERM reschedule flag wiring in CLI execute: prefect/cli/flow_run.py - Kubernetes worker eviction handling and SIGTERM behavior based on backoffLimit: prefect-kubernetes/worker.py If you share your Job template (particularly

backoffLimit

) and whether you saw any node drain/eviction events at the time, I can confirm which path triggered the reschedule.

Yes. When you call run_deployment you get a FlowRun back, which includes the run’s ID. You can turn that into a clickable UI link and include it anywhere (logs, state messages, Slack, etc.). How to do it - Capture the flow run immediately (don’t block):

from prefect.deployments import run_deployment
  from prefect.utilities.urls import url_for

  fr = run_deployment("my-flow/my-deployment", timeout=0)  # returns right away
  link = url_for(fr)  # e.g., <https://your-ui/runs/flow-run/<id>>
  print(f"Child flow: {link}")

Notes: -

url_for(fr)

auto-uses

PREFECT_UI_URL

to build the right base. - If you prefer manual construction:

f"{PREFECT_UI_URL.value()}/runs/flow-run/{fr.id}"

- Drop the link into a state message (shows in the UI):

from prefect.states import Running
  from prefect.client.orchestration import get_client

  async with get_client() as client:
      await client.set_flow_run_state(
          fr.id,
          Running(message=f"Monitor logs: {url_for(fr)}")
      )

- Or send to Slack/notifications:

from prefect_slack import SlackWebhook
  slack = SlackWebhook.load("my-slack")
  slack.notify(body=f"Triggered: {url_for(fr)}")

Handy details - Import path:

from prefect.deployments import run_deployment

(and

arun_deployment

if you’re async) - Return value: a

FlowRun

object; ID is

fr.id

- UI link format (self-hosted):

{PREFECT_UI_URL}/runs/flow-run/<flow_run_id>

- If you’re running inside another flow, this pattern works great for parent->child run links. If you share your base UI URL (or how you want the message formatted), I can give you a drop-in snippet tailored to your environment.

Yes—use artifacts to surface a clickable link in the Prefect UI. That’s the supported way to show custom URLs (like your Kubernetes job/pod logs) on a flow/task run page. Recommended approach: create a Link artifact - Use

create_link_artifact

during the run to publish the K8s logs URL. - It will show up on the run’s Artifacts tab; if you provide a key, it’s also listed on the global Artifacts page with version history. Example (build a K8s logs URL from the flow’s namespace/job name and publish it):

from prefect import flow, task
from prefect.artifacts import create_link_artifact
from prefect.runtime import flow_run

@task
def publish_k8s_logs_link(namespace: str, job_name: str):
    # Example URL patterns – adjust for your dashboard/log viewer
    logs_url = f"<https://k8s-dashboard.example.com/#/job/{namespace}/{job_name}>"
    create_link_artifact(
        link=logs_url,
        link_text=f"View Kubernetes logs for {job_name}",
        key="k8s-job-logs",                # optional but useful for surfacing globally
        description=f"Kubernetes logs for job `{job_name}` in `{namespace}`"
    )

@flow
def my_flow(namespace: str, job_name: str):
    publish_k8s_logs_link(namespace, job_name)

# Run with your known namespace/job_name (you can also infer job/pod from infrastructure_pid)
my_flow("default", "prefect-job-abc123")

Tip: infer namespace/job from the flow run’s infrastructure_pid - For Kubernetes work pools,

flow_run.infrastructure_pid

is typically

namespace:job_name

. - From within the flow:

from prefect.runtime import flow_run
  ns, job = flow_run.infrastructure_pid.split(":", 1)

- Then construct your URL and publish it with

create_link_artifact

. Alternative: use a Markdown artifact for richer formatting

from prefect.artifacts import create_markdown_artifact

create_markdown_artifact(
    markdown=f"<{logs_url}|Open Kubernetes logs>",
    key="k8s-logs-markdown",
    description="Direct link to the job logs"
)

Also possible: place a link in the state message - Less structured, but quick:

from prefect.states import Running
  from prefect.client.orchestration import get_client

  async with get_client() as client:
      await client.set_flow_run_state(
          flow_run_id=<id>,
          state=Running(message=f"Logs: {logs_url}")
      )

Notes - There’s no built-in “extra links” field on runs; artifacts are the supported mechanism for persistent, clickable links in the UI. - Make sure your URL points to something reachable by your users (K8s Dashboard, Lens proxy, Datadog, Kibana, Loki, etc.). If you tell me which viewer you use, I can give you an exact URL template.

Great — since you’re on GKE, the cleanest way is to publish a Cloud Logging (Log Explorer) link as a Prefect artifact for each run. The Kubernetes worker labels pods with

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

, so you can build a precise Log Explorer query that shows exactly the pod logs for that flow run. What to do - Add project/cluster/location info to your K8s work pool (Job Variables → env): -

GKE_PROJECT

-

GKE_CLUSTER

-

GKE_LOCATION

(zone or region) - Derive namespace and job name from the flow run’s

infrastructure_pid

(format

namespace:job_name

). - Create a Link artifact with a Log Explorer URL that filters by: -

resource.type="k8s_container"

-

resource.labels.cluster_name="<cluster>"

-

resource.labels.location="<location>"

-

resource.labels.namespace_name="<namespace>"

-

labels."k8s-pod/prefect.io/flow-run-id"="<flow_run_id>"

- Optionally also filter

labels."k8s-pod/job-name"="<job_name>"

Drop-in snippet

from urllib.parse import quote_plus

from prefect import flow, task
from prefect.artifacts import create_link_artifact
from prefect.runtime import flow_run

def gke_logs_url(*, project: str, cluster: str, location: str,
                 namespace: str, flow_run_id: str, job_name: str | None = None,
                 hours: int = 6) -> str:
    # Build an advanced filter for Cloud Logging
    parts = [
        'resource.type="k8s_container"',
        f'resource.labels.cluster_name="{cluster}"',
        f'resource.labels.location="{location}"',
        f'resource.labels.namespace_name="{namespace}"',
        f'labels."k8s-pod/prefect.io/flow-run-id"="{flow_run_id}"',
    ]
    if job_name:
        parts.append(f'labels."k8s-pod/job-name"="{job_name}"')

    query = "\n".join(parts)
    return (
        f"<https://console.cloud.google.com/logs/query;>"
        f"query={quote_plus(query)};"
        f"timeRange=PT{hours}H?project={project}"
    )

@task
def publish_gke_logs_link():
    # Pull ns/job from infrastructure_pid (namespace:job_name)
    infra = flow_run.infrastructure_pid or ""
    namespace, job_name = infra.split(":", 1) if ":" in infra else ("default", "")

    # Get your GKE env (set these in your K8s work pool job variables)
    import os
    project = os.getenv("GKE_PROJECT", "")
    cluster = os.getenv("GKE_CLUSTER", "")
    location = os.getenv("GKE_LOCATION", "")

    url = gke_logs_url(
        project=project,
        cluster=cluster,
        location=location,
        namespace=namespace,
        flow_run_id=str(flow_run.id),
        job_name=job_name or None,
        hours=6,
    )

    create_link_artifact(
        link=url,
        link_text=f"View GKE logs for {job_name or flow_run.name}",
        key="gke-logs",  # optional, helps surface globally
        description=f"GKE pod logs filtered by flow-run label in namespace `{namespace}`"
    )

@flow
def my_flow():
    publish_gke_logs_link()

Notes - The artifact appears on the run’s Artifacts tab (and on the global Artifacts page if you use a

key

). - If you prefer, you can also place the same URL in the run’s state message so it’s visible at a glance. - The K8s worker applies

<http://prefect.io/flow-run-id|prefect.io/flow-run-id>

and

<http://prefect.io/flow-run-name|prefect.io/flow-run-name>

labels to pods/jobs, so the filter remains stable even if a job creates multiple pods.

Yes—if by “all” you mean “everything my code and libraries emit,” you can get very close by enabling these: - Capture print statements - Per-flow:

@flow(log_prints=True)

- Or globally:

PREFECT_LOGGING_LOG_PRINTS=true

- Capture third‑party library logs (requests, boto3, etc.) - Add their logger names to:

PREFECT_LOGGING_EXTRA_LOGGERS=requests,urllib3,boto3

- You can include any Python logger names your code uses. - Ensure Prefect sends logs to the API - Confirm:

PREFECT_LOGGING_TO_API_ENABLED=true

- If you were seeing partial logs, also consider:

PREFECT_LOGGING_TO_API_BATCH_INTERVAL=1.0
    PREFECT_LOGGING_TO_API_BATCH_SIZE=2000000

- Set an appropriate log level

PREFECT_LOGGING_LEVEL=INFO   # or DEBUG for more detail

Limits and notes - Python warnings are not auto-captured by a dedicated setting; if you need them, add:

import logging, warnings
  logging.captureWarnings(True)

- Anything that does not use Python logging (e.g., raw stdout from a subprocess) won’t be captured unless you: - route it through

print()

with

log_prints=True

, or - add explicit logging/collection in your code, or - view raw container logs via your K8s/GKE viewer. With these enabled, the Prefect UI flow run Logs tab will show: - Prefect flow/task logs - print() output - third‑party logger output you listed - all delivered to the API (subject to batch/flush timing and size limits)