Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 0 additions & 7 deletions Dockerfile.ci
Original file line number Diff line number Diff line change
Expand Up @@ -1270,13 +1270,6 @@ function environment_initialization() {
export AIRFLOW__SCHEDULER__STANDALONE_DAG_PROCESSOR=True
fi

if [[ ${GO_WORKER=} == "true" ]]; then
echo
echo "${COLOR_BLUE}Starting go worker${COLOR_RESET}"
echo
export AIRFLOW__SCHEDULER__GO_WORKER=True
fi

RUN_TESTS=${RUN_TESTS:="false"}
CI=${CI:="false"}

Expand Down
63 changes: 11 additions & 52 deletions airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
Original file line number Diff line number Diff line change
Expand Up @@ -52,22 +52,13 @@ Prerequisites
* The ``apache-airflow-task-sdk`` package (installed with Airflow) provides the coordinator; no additional
Python packages are needed.

Deployment modes
----------------

A packed bundle can run in two ways. The same binary works in both, and you pick one per deployment:

* **Coordinator (recommended).** A Python task runner launches the Go bundle directly, with no separate Go
worker process on the host. This is the same coordinator mechanism the Java SDK uses. Because the mature
Python supervisor handles the Airflow-facing concerns, this path inherits remote task logs (S3/GCS), the
full range of task states, and alternate XCom backends, rather than implementing them again in Go. Those are
exactly the features the Edge Worker path is still missing.
* **Edge Worker.** A long-running Go process (``airflow-go-edge-worker``) polls Airflow for work and runs
your bundle, with no Python in the data path. It runs end-to-end today but is missing the features listed
under :ref:`go-sdk/limitations`.
Execution architecture
----------------------

The rest of this guide covers the recommended coordinator path; see :ref:`go-sdk/edge-worker` for a summary
of the Edge Worker.
A Python task runner launches the Go bundle directly, with no separate Go worker process on the host. This
is the same coordinator mechanism the Java SDK uses. Because the mature Python supervisor handles the
Airflow-facing concerns, Go tasks inherit remote task logs (S3/GCS), the full range of task states, and
alternate XCom backends rather than implementing them again in Go.

Quick start
-----------
Expand Down Expand Up @@ -159,10 +150,6 @@ manifest can never drift from what the binary actually executes.

var _ v1.BundleProvider = (*myBundle)(nil)

func (m *myBundle) GetBundleVersion() v1.BundleInfo {
return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
}

func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
simpleDag := dagbag.AddDag("simple_dag") // must match the Python dag_id
simpleDag.AddTask(extract) // task_id is taken from the function name
Expand Down Expand Up @@ -327,12 +314,10 @@ surface as Go values when read back via ``GetXCom``.
.. note::

``GetXCom`` returns the value exactly as decoded from the transport; there is no typed XCom
deserialization layer yet. The concrete type of a *numeric* value therefore depends on the deployment
mode. Over the Execution API (the Edge Worker path) numbers are decoded with ``encoding/json``, so every
number - integer or not - arrives as ``float64``. In coordinator mode the Python supervisor re-encodes the
value as ``msgpack``, so a whole number arrives as a Go integer type (whose width depends on the value) and
only a non-integer as ``float64``. Do not assume a fixed numeric type: type-switch over the numeric types
you expect, or round-trip the value through ``json.Marshal`` / ``json.Unmarshal`` into a typed Go value.
deserialization layer yet. The Python supervisor encodes values as ``msgpack``, so a whole number arrives
as a Go integer type (whose width depends on the value) and only a non-integer arrives as ``float64``. Do
not assume a fixed integer width: type-switch over the numeric types you expect, or round-trip the value
through ``json.Marshal`` / ``json.Unmarshal`` into a typed Go value.

.. _go-sdk/build:

Expand Down Expand Up @@ -432,37 +417,11 @@ All ``kwargs`` in the ``coordinators`` config entry are passed to the
- Seconds to wait for the bundle subprocess to connect after launch. Increase this if your
bundle startup is slow (e.g. on constrained hardware).

.. _go-sdk/edge-worker:

Alternative: the Go Edge Worker
-------------------------------

The same bundle binary can also run **without a Python coordinator**, under the standalone Go Edge Worker.
Rather than the worker launching the bundle once per task, ``airflow-go-edge-worker`` is a long-running Go
process that registers with the scheduler, polls the Edge Executor API for workloads, and runs the bundle
directly over HashiCorp ``go-plugin`` (gRPC), with no Python in the data path. The bundle source and
``RegisterDags`` registration are identical; only the deployment differs, and the mode is selected
automatically at launch from the CLI flags, so you do not change any task code.

This path does not use the ``[sdk] coordinators`` configuration and is currently missing the features listed
under :ref:`go-sdk/limitations`. See the Go SDK's own repository documentation for Edge Worker setup
(``airflow-go-edge-worker`` configuration and ``go install``).

.. _go-sdk/limitations:

Limitations
-----------

* **A Python stub Dag is still required.** The Execution API does not yet carry Dag structure for non-Python
languages, so task names and dependencies are declared in Python with
:func:`@task.stub <airflow.sdk.task.stub>`. This applies to both deployment modes and is a documented
known limitation.

The following are a non-exhaustive list of features the **Edge Worker** path has yet to implement. They are
the main reason the coordinator path is recommended: in coordinator mode the Python supervisor handles these
concerns, so they are **not** limitations there.

* Putting tasks into states other than success or failed/up-for-retry (deferred, failed-without-retries,
etc.).
* Remote task logs (e.g. S3/GCS).
* XCom reading/writing through non-default XCom backends.
:func:`@task.stub <airflow.sdk.task.stub>`. This is a documented known limitation.
4 changes: 2 additions & 2 deletions contributing-docs/30_new_language_sdk.rst
Original file line number Diff line number Diff line change
Expand Up @@ -38,8 +38,8 @@ start the foreign runtime and *how* to communicate with it, and the language SDK
implements the other end of whatever protocol the coordinator chooses. There is
no single mandated communication mechanism.

For a diagram-driven overview of how a task runs across the Python, Coordinator,
and edge-worker paths, see `Task execution architecture <31_task_execution_architecture.rst>`__.
For a diagram-driven overview of how a task runs across the Python and Coordinator
paths, see `Task execution architecture <31_task_execution_architecture.rst>`__.

.. contents:: Table of Contents
:depth: 2
Expand Down
16 changes: 2 additions & 14 deletions contributing-docs/31_task_execution_architecture.rst
Original file line number Diff line number Diff line change
Expand Up @@ -64,9 +64,9 @@ a **compiled, non-Python language**. A Python Dag still declares the task with `
Python and non-Python tasks can be mixed in one Dag), but the actual work is delegated to the matching runtime.
The **first-class** integration is the **Coordinator** layer — the Python Supervisor drives the language runtime
as a subprocess and proxies every Execution-API call for it — and it is the shared direction for the Java, Go,
and upcoming language SDKs. The Go SDK additionally offers a second, standalone **edge-worker** mode.
and upcoming language SDKs.

**Coordinator (Java and Go) — recommended.** Both the
**Coordinator (Java and Go).** Both the
`Java SDK <../airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst>`__ and the
`Go SDK <../airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst>`__ *reuse* the existing Python
Supervisor through the **Coordinator** layer. ``CoordinatorManager`` resolves the task's ``queue`` to a
Expand All @@ -90,18 +90,6 @@ loopback TCP is drawn as arrows going back and forth to its neighbours:

.. image:: images/diagram_coordinator_execution_sequence.png

**Go SDK — standalone edge worker (alternative).** The Go SDK can also run with **no Python in the data
path**, but this mode is still missing features the Supervisor provides — non-default task states, remote task
logs, and alternate XCom backends (see the Go SDK
`limitations <../airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst#limitations>`__) — so the
Coordinator mode above stays the recommended path. When it is used, a long-running, compiled **edge worker**
(``airflow-go-edge-worker``) *pulls* work from the **Edge Executor API**, launches the user's compiled Dag
bundle as a **go-plugin (gRPC) subprocess**, and invokes the task over gRPC. The task then uses the **native
TEI client** to reach the **Execution API** directly over HTTPS — so, unlike the coordinator path, it holds the
task JWT itself:

.. image:: images/diagram_go_edge_worker_architecture.png

.. note::

Both the Go and Java SDKs are **experimental** and under active development. See the
Expand Down
2 changes: 1 addition & 1 deletion contributing-docs/README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -167,4 +167,4 @@ Task Execution Lifecycle

* `Task execution architecture <31_task_execution_architecture.rst>`__ explains what happens inside a
worker when a task runs: the Python Supervisor and task runner, the Coordinator layer shared by the Java
and Go SDKs, and the standalone Go edge worker.
and Go SDKs.
Original file line number Diff line number Diff line change
@@ -1 +1 @@
274a2f3f7e4d8ef8b7807e3f5234eabd
fbdd79faaf7bd844a15bac9d08944aa9
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,7 @@
"""
Architecture diagram for the Coordinator execution path (Java and Go language SDKs).

Both plug into the *same* Python Supervisor via the **Coordinator** layer, rather
than running a standalone worker:
Both plug into the *same* Python Supervisor via the **Coordinator** layer:

* ``CoordinatorManager.for_queue()`` resolves the task's ``queue`` to a
``BaseCoordinator`` via ``[sdk] queue_to_coordinator`` / ``[sdk] coordinators``
Expand Down Expand Up @@ -277,11 +276,11 @@ def generate_coordinator_execution_architecture_diagram():
_node(
g,
"note",
"How it differs",
"Process boundary",
"vs Python task: same Supervisor, but a separate language subprocess over<br/>"
"loopback TCP instead of a forked Python process over a UNIX socketpair<br/>"
"vs the Go edge worker: the coordinator task does <b>not</b> hold the JWT and does<br/>"
"<b>not</b> call the Execution API directly — the Python Supervisor proxies every call",
"loopback TCP instead of a forked Python process over a UNIX socketpair.<br/>"
"The language task does <b>not</b> hold the JWT or call the Execution API directly;<br/>"
"the Python Supervisor proxies every call.",
shape="note",
theme=NOTE,
)
Expand Down

This file was deleted.

Binary file not shown.
Loading
Loading