diff --git a/Dockerfile.ci b/Dockerfile.ci index 4023e37c9fb51..0a16a57175550 100644 --- a/Dockerfile.ci +++ b/Dockerfile.ci @@ -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"} diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst index 158fff4ddeb2f..067d2d0bc6ac3 100644 --- a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst +++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst @@ -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 ----------- @@ -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 @@ -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: @@ -432,22 +417,6 @@ 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 @@ -455,14 +424,4 @@ 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 `. 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 `. This is a documented known limitation. diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst index 69ebccc23babb..3cb2f60e4609a 100644 --- a/contributing-docs/30_new_language_sdk.rst +++ b/contributing-docs/30_new_language_sdk.rst @@ -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 diff --git a/contributing-docs/31_task_execution_architecture.rst b/contributing-docs/31_task_execution_architecture.rst index 6c9419accca48..6b12ddd0a1418 100644 --- a/contributing-docs/31_task_execution_architecture.rst +++ b/contributing-docs/31_task_execution_architecture.rst @@ -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 @@ -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 diff --git a/contributing-docs/README.rst b/contributing-docs/README.rst index 823ed47d54583..36e7f69b532d5 100644 --- a/contributing-docs/README.rst +++ b/contributing-docs/README.rst @@ -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. diff --git a/contributing-docs/images/diagram_coordinator_execution_architecture.md5sum b/contributing-docs/images/diagram_coordinator_execution_architecture.md5sum index c19d1feaf7f34..06c161604fbfa 100644 --- a/contributing-docs/images/diagram_coordinator_execution_architecture.md5sum +++ b/contributing-docs/images/diagram_coordinator_execution_architecture.md5sum @@ -1 +1 @@ -274a2f3f7e4d8ef8b7807e3f5234eabd +fbdd79faaf7bd844a15bac9d08944aa9 diff --git a/contributing-docs/images/diagram_coordinator_execution_architecture.png b/contributing-docs/images/diagram_coordinator_execution_architecture.png index 22668ad8d1e75..11b5f8a445f43 100644 Binary files a/contributing-docs/images/diagram_coordinator_execution_architecture.png and b/contributing-docs/images/diagram_coordinator_execution_architecture.png differ diff --git a/contributing-docs/images/diagram_coordinator_execution_architecture.py b/contributing-docs/images/diagram_coordinator_execution_architecture.py index aa7a158f6d378..6de9b735dac37 100644 --- a/contributing-docs/images/diagram_coordinator_execution_architecture.py +++ b/contributing-docs/images/diagram_coordinator_execution_architecture.py @@ -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`` @@ -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
" - "loopback TCP instead of a forked Python process over a UNIX socketpair
" - "vs the Go edge worker: the coordinator task does not hold the JWT and does
" - "not call the Execution API directly — the Python Supervisor proxies every call", + "loopback TCP instead of a forked Python process over a UNIX socketpair.
" + "The language task does not hold the JWT or call the Execution API directly;
" + "the Python Supervisor proxies every call.", shape="note", theme=NOTE, ) diff --git a/contributing-docs/images/diagram_go_edge_worker_architecture.md5sum b/contributing-docs/images/diagram_go_edge_worker_architecture.md5sum deleted file mode 100644 index 8ea4f8852048e..0000000000000 --- a/contributing-docs/images/diagram_go_edge_worker_architecture.md5sum +++ /dev/null @@ -1 +0,0 @@ -6f508dad8e802534ed02155b810c7ad8 diff --git a/contributing-docs/images/diagram_go_edge_worker_architecture.png b/contributing-docs/images/diagram_go_edge_worker_architecture.png deleted file mode 100644 index b7e0a71e06db5..0000000000000 Binary files a/contributing-docs/images/diagram_go_edge_worker_architecture.png and /dev/null differ diff --git a/contributing-docs/images/diagram_go_edge_worker_architecture.py b/contributing-docs/images/diagram_go_edge_worker_architecture.py deleted file mode 100644 index 0d8c5ed5e84b2..0000000000000 --- a/contributing-docs/images/diagram_go_edge_worker_architecture.py +++ /dev/null @@ -1,261 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. -# /// script -# requires-python = ">=3.10" -# dependencies = [ -# "rich>=13.6.0", -# "graphviz>=0.20.1", -# ] -# /// -""" -Architecture diagram for the Go SDK's standalone **edge-worker** deployment mode. - -This is the Go SDK's alternative to the Coordinator path: instead of a Python -Supervisor driving the bundle (the approach Java and the Go Coordinator mode both -take), there is **no Python Supervisor and no msgpack socket**: - -* a long-running compiled **edge worker** (``airflow-go-edge-worker``) *pulls* - work from the **Edge Executor API** over HTTP; -* it launches the user's compiled **Dag bundle** as a **go-plugin (gRPC) - subprocess** and invokes the task over gRPC; and -* the task uses the **native Task Execution Interface client** (AIP-72) to reach - the **Execution API** directly over HTTPS — so, unlike the Python task, it holds - the task JWT itself. - -This mode is missing features the Supervisor provides (see the Go SDK "known -limitations"), which is why the Coordinator path is recommended. - -Rendered with graphviz directly so every label sits inside a sized shape: -3-D box = native OS process, rounded box = an object inside a process, -component = a server app, cylinder = the database, note = a caption. -""" - -from __future__ import annotations - -from pathlib import Path - -import graphviz -from rich.console import Console - -MY_DIR = Path(__file__).parent -MY_FILENAME = Path(__file__).with_suffix("").name - -console = Console(width=400, color_system="standard") - -# Soft palette: (fill, border) per role. Kept consistent with the Python Task SDK -# diagrams: user code = green, Execution API = red, DB = grey. -GO = ("#e0f7fa", "#00838f") # compiled edge worker / native clients (Go cyan) -USER = ("#e8f5e9", "#2e7d32") # user task code -EDGE = ("#fff3e0", "#ef6c00") # Edge Executor API (amber) -API = ("#fdecea", "#c62828") # Execution API (red) -NEUTRAL = ("#eceff1", "#455a64") # database -NOTE = ("#fffde7", "#f9a825") # caption - - -def _label(title: str, sub: str | None = None) -> str: - html = f"<{title}" - if sub: - html += f'
{sub}' - return html + ">" - - -def _node(g, node_id: str, title: str, sub: str, *, shape: str, theme: tuple[str, str]) -> None: - fill, border = theme - style = "filled" if shape in ("box3d", "cylinder", "component", "note") else "rounded,filled" - g.node( - node_id, - label=_label(title, sub), - shape=shape, - style=style, - fillcolor=fill, - color=border, - penwidth="2", - margin="0.18,0.12", - ) - - -def generate_go_edge_worker_architecture_diagram(): - image_file = MY_DIR / f"{MY_FILENAME}.png" - console.print(f"[bright_blue]Generating architecture image {image_file}") - - g = graphviz.Digraph("go_edge_worker_architecture") - g.attr( - rankdir="TB", - splines="spline", - nodesep="0.6", - ranksep="0.9", - pad="0.5", - bgcolor="white", - fontname="Helvetica", - newrank="true", - compound="true", - ) - g.attr("node", fontname="Helvetica", fontsize="13", fontcolor="#102027") - g.attr("edge", fontname="Helvetica", fontsize="10", penwidth="1.8", color="#546e7a") - - # ------------------------------------------------------------------ # - # Worker host — compiled native OS processes, no Python runtime. - # ------------------------------------------------------------------ # - with g.subgraph(name="cluster_host") as host: - host.attr( - label="Worker host — native OS processes (compiled binary · no Python runtime)", - labelloc="t", - style="rounded,filled", - fillcolor="#f4fdff", - color="#90a4ae", - penwidth="1.5", - fontsize="16", - fontname="Helvetica-Bold", - margin="18", - ) - - with host.subgraph(name="cluster_worker") as worker: - worker.attr( - label="Edge-worker process · long-running compiled binary", - labelloc="t", - style="rounded,filled", - fillcolor="#e7fbff", - color=GO[1], - penwidth="1.5", - fontsize="13", - fontname="Helvetica-Bold", - margin="14", - ) - _node( - worker, - "worker", - "airflow-go-edge-worker", - "register · heartbeat · fetch workloads
report task state · spawn bundle plugins", - shape="box3d", - theme=GO, - ) - _node( - worker, - "edge_client", - "edgeapi.Client", - "Edge Executor API client", - shape="box", - theme=GO, - ) - worker.edge("worker", "edge_client", style="dotted", color=GO[1], arrowhead="none") - - with host.subgraph(name="cluster_bundle") as bundle: - bundle.attr( - label="Dag bundle plugin process · go-plugin (gRPC) subprocess · runs USER CODE", - labelloc="t", - style="rounded,filled", - fillcolor="#edf7ee", - color=USER[1], - penwidth="1.5", - fontsize="13", - fontname="Helvetica-Bold", - margin="14", - ) - _node( - bundle, - "gotask", - "Go task function", - "your compiled task code, registered in the bundle
holds the task JWT", - shape="box3d", - theme=USER, - ) - _node( - bundle, - "tei_client", - "sdk client → pkg/api", - "native Task Execution Interface client (AIP-72)
" - "VariableClient · ConnectionClient · XComClient", - shape="box", - theme=GO, - ) - bundle.edge("gotask", "tei_client", style="dotted", color=GO[1], arrowhead="none") - - # Worker launches the bundle plugin and invokes the task over gRPC. - g.edge( - "worker", - "gotask", - label="go-plugin gRPC\nExecute(ExecuteTaskWorkload)", - color=GO[1], - fontcolor=GO[1], - penwidth="2.2", - ) - - # ------------------------------------------------------------------ # - # Airflow API server — separate process / host. - # ------------------------------------------------------------------ # - with g.subgraph(name="cluster_server") as server: - server.attr( - label="Airflow API server · separate process / host", - labelloc="t", - style="rounded,filled", - fillcolor="#fdf6ec", - color="#b0846a", - penwidth="1.5", - fontsize="14", - fontname="Helvetica-Bold", - margin="16", - ) - _node(server, "edge_api", "Edge Executor API", "(edge provider)", shape="component", theme=EDGE) - _node( - server, - "execution_api", - "Execution API", - "FastAPI · TEI / AIP-72", - shape="component", - theme=API, - ) - _node(server, "metadata_db", "Metadata DB", "", shape="cylinder", theme=NEUTRAL) - server.edge("edge_api", "metadata_db", style="dotted", color=EDGE[1], dir="both") - server.edge("execution_api", "metadata_db", style="dotted", color=API[1], dir="both") - - g.edge( - "edge_client", - "edge_api", - label="HTTP + Edge API JWT\nregister · heartbeat · poll for\nworkloads · report state", - color=EDGE[1], - fontcolor=EDGE[1], - penwidth="2.2", - ) - g.edge( - "tei_client", - "execution_api", - label="HTTPS + task JWT\nVariables · Connections · XComs · state", - color=API[1], - fontcolor=API[1], - penwidth="2.2", - ) - - # Contrast caption. - _node( - g, - "note", - "Contrast with the Python Task SDK", - "no Python Supervisor · no msgpack stdin socket
" - "work is pulled via the Edge API (not pushed by an executor)
" - "the task runs as a compiled gRPC plugin and calls the
" - "Execution API directly — so it holds the JWT itself", - shape="note", - theme=NOTE, - ) - g.edge("worker", "note", style="invis") - - g.render(outfile=str(image_file), format="png", cleanup=True) - console.print(f"[green]Generated architecture image {image_file}") - - -if __name__ == "__main__": - generate_go_edge_worker_architecture_diagram() diff --git a/dev/breeze/doc/images/output_shell.svg b/dev/breeze/doc/images/output_shell.svg index 3d183fe886dd7..f2adc36e62ea4 100644 --- a/dev/breeze/doc/images/output_shell.svg +++ b/dev/breeze/doc/images/output_shell.svg @@ -1,4 +1,4 @@ - +