Run the dispatcher

Flow's Kubernetes dispatcher, flow-service-dispatcher, claims task runs from core and runs each one as a pod in your cluster. This page is for operators deploying it. It covers executors, task types, Kubernetes permissions, task pods, AI tasks and running more than one dispatcher.

Before you start

• Core is running, and the dispatcher can reach its port 7700. See Configuring.
• A namespace for task pods, run workspace volumes and script ConfigMaps.
• A dispatcher token: a global machine token with an actor kind. See Give the dispatcher its token.
• Tekton Pipelines, only for the tekton executor.
• The same release tag as core, such as 5.0.0-beta.39. Core and the dispatcher move together.

What does the dispatcher do?

At start, the dispatcher registers with core for the task types it runs, then long-polls core for work. For each task it claims, it:

1. creates a Kubernetes Job or a Tekton TaskRun;
2. watches it;
3. reports the end and the results to core.

It also creates each run's run workspace volume before the run starts, and deletes the volume once the run has ended. Core calls the dispatcher only to read a running task's log, at the address the dispatcher registers (see Task logs). The dispatcher needs no ingress.

Choose an executor

dispatcher.executor (DISPATCHER_EXECUTOR) sets how each task runs. Both executors build the same volumes, environment and resources, and report the same failure reasons.

• The defaults differ. The image defaults to tekton. bmrg-flow 6.0.1, the unreleased v5 chart, defaults to kube-jobs.
• No minimum version of Kubernetes or Tekton is stated for Flow.
• On kube-jobs, a script task's body is mounted at /scripts/script and must start with a shebang, such as #!/bin/sh. Tekton runs it as the step's script.
• On Tekton, raise tekton.results.maxBytes only to the cluster's max-result-size, and only when Tekton runs with sidecar logs.
• Jobs only: kube.task.backOffLimit (0) and kube.task.restartPolicy (Never) set the Job's backoff limit and the pod's restart policy.

With tekton on a cluster without Tekton Pipelines, the dispatcher can't create a TaskRun, and every task fails with reason DispatchError. Install Tekton Pipelines, or set DISPATCHER_EXECUTOR=kube-jobs and restart the dispatcher.

Connect the dispatcher to Flow

For example, on the dispatcher's container:

env:
  - name: FLOW_ENGINE_URL
    value: http://flow-core.flow.svc:7700
  - name: FLOW_ENGINE_DISPATCHER_TOKEN
    valueFrom:
      secretKeyRef: { name: flow-dispatcher, key: token }
  - name: FLOW_DISPATCHER_NAME
    value: flow-kube-jobs
  - name: DISPATCHER_EXECUTOR
    value: kube-jobs
  - name: KUBE_NAMESPACE
    value: flow-tasks
  - name: FLOW_DISPATCHER_LOGS_URL
    value: http://flow-dispatcher.flow.svc:7702/api/v1/logs/stream

Like core, the image has no shell. Put JVM flags in JDK_JAVA_OPTIONS, and probe it with HTTP GET /health on port 7702 (see Configure health probes).

The dispatcher is connected when its log shows:

Dispatcher flow-kube-jobs (<host>) registered as <id> for task types [template, custom, script, ai, uploadartifact, downloadartifact].

Grant the Kubernetes permissions

Bind a namespaced Role in kube.namespace to the dispatcher's service account. No Role ships with Flow itself.

The dispatcher cancels, cleans up and releases volumes by label, with one collection delete each. That needs deletecollection; delete isn't enough. As a manifest:

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: flow-dispatcher
  namespace: flow-tasks
rules:
  - apiGroups: [""]
    resources: [pods]
    verbs: [get, list]
  - apiGroups: [""]
    resources: [pods/log]
    verbs: [get]
  - apiGroups: [""]
    resources: [persistentvolumeclaims]
    verbs: [get, list, watch, create, deletecollection]
  # kube-jobs executor
  - apiGroups: [batch]
    resources: [jobs]
    verbs: [get, list, watch, create, deletecollection]
  - apiGroups: [""]
    resources: [configmaps]
    verbs: [create, deletecollection]
  # tekton executor
  - apiGroups: [tekton.dev]
    resources: [taskruns]
    verbs: [get, list, watch, create, delete, deletecollection]
  - apiGroups: [tekton.dev]
    resources: [taskruns/status]
    verbs: [update]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: flow-dispatcher
  namespace: flow-tasks
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: flow-dispatcher
subjects:
  - kind: ServiceAccount
    name: flow-dispatcher
    namespace: flow

kubectl auth can-i deletecollection jobs.batch -n flow-tasks \
  --as=system:serviceaccount:flow:flow-dispatcher

A missing rule shows as a Kubernetes forbidden error in the dispatcher's log, and as:

Choose the task types

flow.dispatcher.task-types (FLOW_DISPATCHER_TASK_TYPES) lists the task types this dispatcher registers. Core hands it claims for those types only. Every other type, such as approvals, sleeps and child workflows, runs inside core.

To register generic too:

flow.dispatcher.task-types=template,custom,script,ai,uploadartifact,downloadartifact,generic

A task whose type no connected dispatcher registers waits, at status ready, until one does. Core drops a type it doesn't know from a registration.

flow.dispatcher.task.max-in-flight (FLOW_DISPATCHER_TASK_MAX_IN_FLIGHT, 25; 0 is no cap) is the most tasks one dispatcher holds at once. A slot is held from the claim until core takes the task's end, including while a pod is Pending. While every slot is full, the dispatcher still receives cancels.

How leases and heartbeats work

A claim carries a lease. The dispatcher's heartbeat renews the lease of every task it holds, so core can tell a slow task from a lost one.

• A dispatcher silent for 60 seconds, with no poll or heartbeat, is treated as gone, and core requeues its claims.
• On stop, the dispatcher waits up to 10 seconds for the tasks in flight. It hands back each claim it never started, and core requeues it at once. A task still running reports nothing, so its lease lapses and core hands it out again.
• After a restart, both executors adopt a Job or TaskRun still running under the task's labels rather than create a second one.
• Registrations expire. Core deletes a dispatcher's registration after 7 days without a connection, so replaced pods don't pile up. A dispatcher that comes back registers again with a new ID.
• To stop handing out work, set flow.queue.enabled=false on core. See Connect the dispatcher.

Size and place task pods

Sizing, placement and isolation are set per dispatcher deployment, never per task. A workload that needs another size or tier runs a second dispatcher with its own task types (see Run several dispatchers).

Six properties size every task container, on both executors:

A blank value sets nothing, and with all six blank the container has no resources block. CPU ships blank because a CPU limit throttles a task rather than failing it.

Every task also gets /data, an emptyDir. It's memory-backed when kube.task.storage.data.memory=true and the task's worker.storage.data.memory parameter is true. Memory-backed data counts against the memory limit, so size memory to cover it.

When a task can't start

Starting is bounded apart from running. A task that never started is handed back to core rather than failed, unless the task itself is wrong.

Core requeues a task up to three times, with backoff, then fails it. A full cluster therefore delays a task, and an image mistake fails it in about a minute. A pod lost after it started, by eviction or preemption, is an ordinary failure.

A task with no timeout gets kube.task.timeout (60 minutes) as a per-pod backstop. The Maximum task duration setting, in Admin › Settings › Tasks, caps every task; its default, 0, lets a task inherit its run's timeout.

Configure run workspaces

For a run that declares a run workspace, the dispatcher creates one PersistentVolumeClaim before the run starts, labelled boomerang.io/workspace-ref=<run id>. Every task in the run mounts it. Every flow.dispatcher.workspace.reconcile-ms (60 seconds), the dispatcher deletes the claims of finished runs.

A finished task's pod still holds the claim. Under the Deletion Policy Never, the claim stays Terminating until those pods are removed (see Clean up finished tasks).

Clean up finished tasks

The Deletion Policy setting, in Admin › Settings › Tasks, decides when a finished task's Job or TaskRun is deleted:

• Whatever the policy, a finished Job or TaskRun is removed kube.task.ttlDays (7) days after it completes. Jobs use ttlSecondsAfterFinished. For TaskRuns, the dispatcher sweeps its own every flow.dispatcher.task.reconcile-ms (hourly), or never with flow.dispatcher.task.reconcile.enabled=false.
• The policy is taken when a run is queued, so a change applies to runs queued after it.
• kube.task.deletion never applies, because core always sends the policy.
• The pod log goes with the Job or TaskRun. With an artifact store, Flow keeps its own copy: see Store finished logs.

Run AI tasks

AI Prompt and AI Agent are ai tasks. Their catalogue revision declares the image, boomerangio/task-ai, and the dispatcher runs it like any other task.

The endpoint and token reach a pod only when set. A task's own Endpoint and API token win over them (see Use AI tasks).

• An image outside the prefixes fails the task with reason ImageNotAllowed before anything is created. To run your own AI worker image, add its prefix: flow.dispatcher.ai.image-prefixes=boomerangio/task-ai,registry.example.com/ai/.
• The token is visible in the pod spec, like any secured parameter.

Give AI tasks their own network zone

1. Run a second dispatcher deployment with flow.dispatcher.task-types=ai and its own flow.dispatcher.name.
2. Put it in a namespace whose egress policy allows only the model endpoint.
3. Set the general dispatcher's flow.dispatcher.task-types to template,custom,script,uploadartifact,downloadartifact, without ai.

No other task type can then reach the model endpoint, and no AI task runs outside the zone. One dispatcher can't hold two zones.

Export traces from tasks

Flow stores no traces, and core emits no run or task spans. Each task gets TRACEPARENT, derived from its run and task run, so tasks that export their own spans land in one trace per run.

For ai pods, the dispatcher can set the exporter:

Each ai pod also gets OTEL_SERVICE_NAME=flow-task-<task> and OTEL_RESOURCE_ATTRIBUTES with flow.workflow.run.id, flow.task.run.id and flow.task.name. The AI worker doesn't support grpc: it logs one warning and exports nothing. Langfuse takes OTLP over HTTP at <host>/api/public/otel; Tempo listens for it on port 4318.

Run several dispatchers

Flow's Kubernetes dispatcher splits work by task type only. Run one deployment per split, each with its own flow.dispatcher.name and flow.dispatcher.task-types:

• An AI zone: see Give AI tasks their own network zone.
• An isolation tier or a size: a deployment with its own dispatcher.tasks.runtimeClassName or kube.resource.*, registered for the types that need it.

Two deployments registered for the same type both claim it, so remove a type from one when another should own it. Flow's Kubernetes dispatcher can't split by workflow label, by task or by workspace. Only a dispatcher built with the dispatcher SDK can, through its poll filters: see Filters in the dispatcher protocol reference. Even then, a dispatcher without filters still claims work a filtered one would take.

Replicas of one deployment each register under the pod's host name and claim work. Each dispatcher process also accepts live log reads only with the bearer it generated at its own start. If flow.dispatcher.logs.url names a Service in front of several replicas, a live log read can land on the wrong one and be refused. Run one replica, or register each pod's own address:

env:
  - name: POD_IP
    valueFrom:
      fieldRef: { fieldPath: status.podIP }
  - name: FLOW_DISPATCHER_LOGS_URL
    value: http://$(POD_IP):7702/api/v1/logs/stream

Next steps

• Task logs: stored logs, live logs and Loki.
• Machine tokens: the dispatcher's token and its rotation.
• Configuring: core's properties, routes and probes.
• Use AI tasks: what workflow authors set on AI tasks.
• Write a dispatcher: run tasks somewhere other than Kubernetes, or split work by workflow label.