Streaming Logs and Channels

Purpose and Scope

This page documents the streaming-oriented surfaces that connect Airflow task execution, language SDK coordinators, task logs, and the task-instance log UI. In this context, a channel is a runtime communication path between an SDK worker process and the Airflow side that coordinates execution. A log channel carries structured log records, while a communication channel carries request and response frames for task-time operations such as reading Variables, Connections, and XCom. Runtime signals, such as SIGTERM and SIGINT, are treated as part of the same contract because they decide how a running handler is asked to stop and how long it may clean up before the process exits.

Sources: ts-sdk/README.md, ts-sdk/tests/coordinator/log-channel.test.ts, ts-sdk/tests/coordinator/comm-channel.test.ts, ts-sdk/tests/coordinator/runtime-signal.test.ts, airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.test.tsx

The strongest source-backed example in this repository is the alpha TypeScript SDK. The README describes public TypeScript task handlers and the coordinator runtime used to execute registered handlers from Airflow. It also explains that Airflow still declares the DAG in Python, while a TypeScript module registers handlers with matching dagId and taskId values and starts a coordinator. The channel tests then define the behavioral contract that makes this split practical: the handler process can write log records, ask the coordinator for Airflow metadata, return task state, and react to termination signals in a controlled way.

Sources: ts-sdk/README.md, ts-sdk/tests/coordinator/log-channel.test.ts, ts-sdk/tests/coordinator/comm-channel.test.ts, ts-sdk/tests/coordinator/runtime-signal.test.ts

Relevant Source Files

ts-sdk/README.md - Introduces @apache-airflow/ts-sdk, registerTask, TaskHandlerArgs, TaskClient, startCoordinator, Python stub tasks, and the coordinator configuration shape used by TypeScript bundles.
ts-sdk/tests/coordinator/log-channel.test.ts - Specifies how LogChannel connects to a TCP endpoint, formats newline-delimited JSON records, assigns logger names, creates child loggers, and shares socket ownership.
ts-sdk/tests/coordinator/comm-channel.test.ts - Specifies request/response timeout behavior, socket destruction on response write timeout, default coordinator request timeout behavior, and write-error handling for CommChannel.
ts-sdk/tests/coordinator/runtime-signal.test.ts - Specifies runtime abort behavior for SIGTERM and SIGINT, including abort reasons, listener cleanup, and force-exit grace-period handling.
airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.test.tsx - Exercises task log UI behavior including virtualized rendering, source and location toggling, and grouped task log sections.

Core Primitives

The core SDK primitive is the task handler: an async TypeScript function registered with registerTask({ dagId, taskId }, handler). Handler arguments include ctx and client; ctx identifies the running task, while TaskClient exposes task-time Airflow data access. The README examples show handlers calling client.getVariable, client.getConnection, and client.getXCom, then returning serializable values. Non-undefined returns are pushed to XCom under the return_value key by the active runtime, matching Python @task behavior. This means the streaming runtime is not a replacement for Airflow scheduling; it is an execution-side bridge for non-Python handlers.

Sources: ts-sdk/README.md

The second primitive is the coordinator. Airflow runs TypeScript bundles through the Python-side airflow.sdk.coordinators.node.NodeCoordinator, configured under the [sdk] section with a coordinators mapping and queue_to_coordinator. The Python DAG declares stub tasks, usually with a queue such as typescript, and the TypeScript bundle contains bundle.mjs, airflow-metadata.yaml, registered handlers, and await startCoordinator(). This separation gives the scheduler a normal Airflow DAG graph while allowing the actual task implementation to run in a Node 22+, ESM-only TypeScript SDK process.

Sources: ts-sdk/README.md

Execution Flow

A typical execution starts with a Python DAG that declares the dependency graph using stub tasks. When a stub task is assigned to the queue mapped to the TypeScript coordinator, Airflow launches the configured Node coordinator against a bundle directory. The TypeScript entrypoint imports all modules that call registerTask, then starts the coordinator. At runtime, the coordinator matches the Airflow task identity to the registered handler. During handler execution, the communication channel handles structured coordinator requests, while the log channel emits records that can be rendered as task logs. The handler returns a value, throws an error, or observes an abort signal, and the coordinator reports the resulting task state back to Airflow.

Sources: ts-sdk/README.md, ts-sdk/tests/coordinator/comm-channel.test.ts, ts-sdk/tests/coordinator/log-channel.test.ts

The communication channel tests emphasize that runtime requests must fail predictably rather than hang indefinitely. CommChannel.request can be called with a timeout override, and when no override is supplied it uses COORDINATOR_REQUEST_TIMEOUT_MS. If a request for GetVariable does not receive a response before the timeout, the promise rejects with a message naming the request type and elapsed milliseconds. Sending responses has a separate timeout path: if writing a task-state response does not complete in time, the channel destroys the socket with an error. If the write completes or throws synchronously, the timeout is cleared so a later timer does not destroy an already-resolved channel operation.

Sources: ts-sdk/tests/coordinator/comm-channel.test.ts

Log Channel Semantics

LogChannel is the SDK-side surface for sending structured log events over a socket. The tests show that LogChannel.connect accepts an address and optionally a custom root logger name. Without a custom name, records use ts-sdk. Each emitted record includes JSON fields such as event, level, logger, and an automatically stamped string timestamp. The event text is also prefixed with the logger name, for example [ts-sdk] hello, so the Airflow UI text renderer can surface the source even when it is presenting log content as text rather than as raw JSON fields.

Sources: ts-sdk/tests/coordinator/log-channel.test.ts

Child loggers make it possible to label subsystem logs without opening separate sockets. Calling child("comm") on the root creates a logger named ts-sdk.comm; another child such as client becomes ts-sdk.client. The tests verify that root, communication, and client log records can be emitted in order with their own levels, while sharing the same underlying connection. They also verify socket ownership: child channels must not close the shared socket. That distinction matters for long-lived coordinators because a helper subsystem should be able to finish logging without accidentally terminating the root runtime log stream.

Sources: ts-sdk/tests/coordinator/log-channel.test.ts

Task Log UI Behavior

The Airflow UI test for task instance logs shows how streamed or stored log content is presented to users. The component renders a virtualized-list, waits for virtualized items or group headers to appear, and responds to scrolling. Source details are hidden by default, then a keyboard S interaction toggles logger and location fields into view. The test expects a row containing a DagBag message to reveal source=airflow.models.dagbag.DagBag and loc=dagbag.py:593. This is an important contract for SDK logs too: preserving logger/source fields allows advanced users to correlate visible log messages with the subsystem that produced them.

Sources: airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.test.tsx

The same UI test also validates grouped log sections. Group headers use summary-style test identifiers, and groups such as Log message source details and Pre task execution logs are visible as collapsible summaries. Combined with virtualized rendering, grouping lets the UI handle large task logs without forcing every row into the DOM at once. For operators, SDK authors, and UI contributors, the practical implication is that log records should remain structured enough to support grouping, source display, and text rendering, but compact enough to stream and virtualize efficiently.

Sources: airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.test.tsx

Runtime Signals and Shutdown

Runtime signal handling defines how Airflow can stop a running SDK process. The createRuntimeAbort tests show that receiving SIGTERM aborts the runtime signal immediately, but does not force process exit until ABORT_GRACE_PERIOD_MS elapses. If the handler or coordinator does not finish within that grace period, the supplied exitProcess function is called with code 1. When an event does not pass a payload, the registered signal name is still used in the abort reason, producing a message such as Task aborted by SIGTERM. This gives handler code a standard abort signal while preserving diagnostic context.

Sources: ts-sdk/tests/coordinator/runtime-signal.test.ts

Shutdown also has a cleanup contract. The runtime abort object exposes dispose, and the tests verify that disposing after a signal removes listeners for both SIGINT and SIGTERM and clears the force-exit timer. That prevents a completed or intentionally torn-down runtime from exiting later because of an old timer. SDK code that opens log and communication channels should follow the same discipline: abort work promptly, close the root log channel only from the owner, let pending communication operations resolve or reject, and dispose signal listeners once the coordinator is no longer responsible for the task.

Sources: ts-sdk/tests/coordinator/runtime-signal.test.ts, ts-sdk/tests/coordinator/log-channel.test.ts, ts-sdk/tests/coordinator/comm-channel.test.ts

Compact Reference

Surface	Source-backed behavior
`registerTask({ dagId, taskId }, handler)`	Binds a TypeScript handler to a Python stub DAG task identity for coordinator mode.
`TaskHandlerArgs`	Provides handler context and a `TaskClient` for task-time Airflow access.
`TaskClient.getVariable`, `getConnection`, `getXCom`	Used by README examples for Variables, Connections, and XCom return values.
`startCoordinator()`	Starts the TypeScript coordinator entrypoint after handlers are registered.
`LogChannel.connect(address, loggerName?)`	Opens a log socket with default logger `ts-sdk` or a custom root name.
`LogChannel.child(name)`	Creates hierarchical child loggers such as `ts-sdk.comm` sharing the root socket.
`CommChannel.request(message, options?)`	Sends coordinator requests and rejects on response timeout.
`CommChannel.sendResponse(id, payload, error, options?)`	Writes responses and destroys the socket if the write times out.
`createRuntimeAbort(logs, options?)`	Produces an abort signal driven by process-style signals and a force-exit grace period.
Task log UI `S` shortcut	Toggles source and location metadata in the task-instance log view.

Next Steps

If you are authoring TypeScript task handlers, start by matching Python stub task IDs to registerTask calls, then use TaskClient for Airflow data instead of embedding scheduler-side assumptions in the Node process. If you are extending the runtime, treat log records, coordinator frames, and abort signals as one reliability boundary: logs must stay readable in the UI, requests must time out cleanly, and termination must give handlers a bounded cleanup window. For adjacent topics, read the Task SDK and Task SDK Sessions pages for the broader language-SDK contract, and the Task Logs page for operational log retrieval and troubleshooting.

Ask a question...