diff --git a/airflow-core/docs/core-concepts/overview.rst b/airflow-core/docs/core-concepts/overview.rst index ab12b6afdfd6f..1f2b1ad8a11bf 100644 --- a/airflow-core/docs/core-concepts/overview.rst +++ b/airflow-core/docs/core-concepts/overview.rst @@ -192,6 +192,91 @@ code is never executed in the context of the *scheduler*. bundle version when dispatching each task. If needed, the cadence of sync and scan of the *Dag bundle* can be configured. +Task execution architecture +--------------------------- + +The diagrams above show how Airflow's components are *deployed*. The diagrams below instead show what happens +*inside a worker when a task actually runs* — how the Task SDK, the Supervisor and Coordinator processes, and +the language runtimes work together: which processes are involved, and the classes and protocols they use to +communicate. + +.. _overview-task-sdk-execution-architecture: + +Python Task SDK execution +......................... + +When a *worker* actually runs a task, it does not run the user's code directly. Instead it starts a +lightweight **Supervisor** that runs in its own **native operating-system process** and +*forks* a second native process in which the **Task SDK** runtime (``task_runner``) executes the user code. +The two processes talk over a socket, and the Supervisor is the only side that ever holds the short-lived +task JWT or talks to the *Execution API* — the user's code never sees the token and never touches the +database. + +The same runtime can also run *in-process* (a single Python process, no fork, no sockets, no HTTP) for +``dag.test()`` and local runs. The diagram below contrasts the two paths and marks where each Python process +lives: + +.. image:: ../img/diagram_task_sdk_execution_architecture.png + +The message flow of a supervised run — startup, running the user code, proxied Connection/Variable/XCom +lookups, heartbeats, and reporting the final state — is shown below as a sequence diagram, with each process +on its own lifeline. The **Supervisor** sits in the middle, so the Task ↔ Supervisor request/response +round-trip (the task asks for a Connection/Variable/XCom and gets the answer back) reads as arrows going back +and forth between neighboring lifelines. Each arrow is numbered, colored by its sender, and labeled with the +message class or protocol used: + +.. image:: ../img/diagram_task_sdk_execution_sequence.png + +.. _overview-non-python-language-sdks: + +Non-Python language SDKs (Go and Java) +...................................... + +The Task Execution Interface (TEI) introduced in AIP-72 is language-agnostic, so a task can also be written in +a **compiled, non-Python language**. A Python Dag still declares the task with ``@task.stub(queue=...)`` (so +Python and non-Python tasks can be mixed in one Dag), but the actual work is delegated to the matching runtime. +There are currently **two different integration styles** — the Go SDK runs a standalone worker, while the Java +SDK plugs into the existing Python Supervisor. + +.. _overview-go-sdk-architecture: + +**Go SDK — standalone edge worker.** The `Go Task SDK +`_ has **no Python Supervisor and no msgpack +stdin socket**. 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 Python task, it holds the task JWT itself: + +.. image:: ../img/diagram_native_language_sdk_architecture.png + +.. _overview-java-sdk-architecture: + +**Java (JVM) SDK — Coordinator plugged into the Supervisor.** The `Java Task SDK +`_ takes the opposite approach: it *reuses* the +existing Python Supervisor through a new **Coordinator** layer. ``CoordinatorManager`` resolves the task's +``queue`` to a ``BaseCoordinator`` — ``JavaCoordinator`` for the ``java`` queue, or the built-in +``_PythonCoordinator`` otherwise. ``JavaCoordinator`` opens two loopback-TCP servers, spawns a **JVM bundle +process** with ``subprocess.Popen``, and drives it with ``_JavaActivitySubprocess`` (a subclass of the shared +``ActivitySubprocess``). The JVM connects *back* over TCP and speaks the **same msgpack protocol** as a Python +task, so the Python side heartbeats, manages state, and **proxies every Execution-API call** — meaning the JVM +task, like a Python task, never holds the task JWT itself: + +.. image:: ../img/diagram_java_sdk_execution_architecture.png + +The end-to-end workflow of a Java task — from ``@task.stub`` through the coordinator, the JVM subprocess, the +proxied Connection/Variable/XCom lookups, and reporting the final state — is shown below as a sequence diagram. +As above, the **Supervisor** is the central lifeline, so the JVM ↔ Supervisor round-trip over loopback TCP is +drawn as arrows going back and forth to its neighbours: + +.. image:: ../img/diagram_java_sdk_execution_sequence.png + +.. note:: + + Both the Go and Java SDKs are **experimental** and under active development. See the `Go Task SDK + documentation `_ and the `Java Task SDK + documentation `_ for current status, + quick-starts, and known limitations. + .. _overview:workloads: Workloads diff --git a/airflow-core/docs/img/diagram_java_sdk_execution_architecture.md5sum b/airflow-core/docs/img/diagram_java_sdk_execution_architecture.md5sum new file mode 100644 index 0000000000000..377507bd88454 --- /dev/null +++ b/airflow-core/docs/img/diagram_java_sdk_execution_architecture.md5sum @@ -0,0 +1 @@ +061e71105ed4ce16e740a942073b7ace diff --git a/airflow-core/docs/img/diagram_java_sdk_execution_architecture.png b/airflow-core/docs/img/diagram_java_sdk_execution_architecture.png new file mode 100644 index 0000000000000..8fa1e5e6255b6 Binary files /dev/null and b/airflow-core/docs/img/diagram_java_sdk_execution_architecture.png differ diff --git a/airflow-core/docs/img/diagram_java_sdk_execution_architecture.py b/airflow-core/docs/img/diagram_java_sdk_execution_architecture.py new file mode 100644 index 0000000000000..25be42106221a --- /dev/null +++ b/airflow-core/docs/img/diagram_java_sdk_execution_architecture.py @@ -0,0 +1,287 @@ +# 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 Java (JVM) Task SDK. + +Unlike the Go SDK (a standalone edge worker), the Java SDK plugs into the *same* +Python Supervisor via a new **Coordinator** layer: + +* ``CoordinatorManager`` resolves the task's ``queue`` to a ``BaseCoordinator`` + (``JavaCoordinator`` for the ``java`` queue, ``_PythonCoordinator`` otherwise); +* ``JavaCoordinator.execute_task()`` opens two loopback-TCP servers, spawns a JVM + bundle process with ``subprocess.Popen``, and drives it with + ``_JavaActivitySubprocess`` — a subclass of the shared ``ActivitySubprocess``; +* the JVM process connects *back* over TCP and speaks the same msgpack protocol as + a Python task, so the Python side heartbeats, proxies every Execution-API call, + and manages state. The JVM task therefore **never holds the task JWT**. + +Rendered with graphviz directly so labels sit inside sized shapes: 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") + +# (fill, border) per role — consistent with the other Task SDK diagrams. +COORD = ("#ede7f6", "#5e35b1") # Coordinator layer (Python, deep purple) +SUP = ("#e3f2fd", "#1565c0") # Supervisor / ActivitySubprocess + Client (blue) +JVM = ("#fbe9e7", "#d84315") # JVM SDK runtime (deep orange) +USER = ("#e8f5e9", "#2e7d32") # user task code (green) +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_java_sdk_execution_architecture_diagram(): + image_file = MY_DIR / f"{MY_FILENAME}.png" + console.print(f"[bright_blue]Generating architecture image {image_file}") + + g = graphviz.Digraph("java_sdk_execution_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 — native OS processes. + # ------------------------------------------------------------------ # + with g.subgraph(name="cluster_host") as host: + host.attr( + label="Worker host — native OS processes", + labelloc="t", + style="rounded,filled", + fillcolor="#f7f7fb", + color="#90a4ae", + penwidth="1.5", + fontsize="16", + fontname="Helvetica-Bold", + margin="18", + ) + + with host.subgraph(name="cluster_sup") as sup: + sup.attr( + label="Supervisor process · native OS process (Python)", + labelloc="t", + style="rounded,filled", + fillcolor="#eef0fb", + color=SUP[1], + penwidth="1.5", + fontsize="13", + fontname="Helvetica-Bold", + margin="14", + ) + _node( + sup, + "coord_mgr", + "CoordinatorManager.for_queue()", + "resolves the task queue → a coordinator via
[sdk] queue_to_coordinator / [sdk] coordinators", + shape="box", + theme=COORD, + ) + _node( + sup, + "java_coord", + "JavaCoordinator (BaseCoordinator)", + "execute_task(client): open two loopback-TCP
servers, subprocess.Popen(java -jar bundle)", + shape="box", + theme=COORD, + ) + _node( + sup, + "supervisor", + "_JavaActivitySubprocess", + "subclass of the shared ActivitySubprocess
heartbeat · proxy every API call · manage state", + shape="box3d", + theme=SUP, + ) + _node( + sup, + "client", + "Client", + "authenticated Execution API client
holds the short-lived task JWT", + shape="box", + theme=SUP, + ) + sup.edge( + "coord_mgr", "java_coord", style="dotted", color=COORD[1], arrowhead="vee", label="picks" + ) + sup.edge( + "java_coord", "supervisor", style="dotted", color=SUP[1], arrowhead="vee", label="drives" + ) + sup.edge("supervisor", "client", style="dotted", color=SUP[1], arrowhead="none") + + with host.subgraph(name="cluster_jvm") as jvm: + jvm.attr( + label="JVM bundle subprocess · native OS process (JVM) · runs USER CODE", + labelloc="t", + style="rounded,filled", + fillcolor="#fdefe9", + color=JVM[1], + penwidth="1.5", + fontsize="13", + fontname="Helvetica-Bold", + margin="14", + ) + _node( + jvm, + "server", + "Server.serve(bundle)", + "bundle JAR entry point (Main-Class)
Kotlin runtime · connects back over TCP", + shape="box3d", + theme=JVM, + ) + _node( + jvm, + "task", + "Task.execute(Context, Client)", + "your Java / Kotlin task [user code]
Context = static run data · Client = API accessors", + shape="box3d", + theme=USER, + ) + _node( + jvm, + "comm", + "CoordinatorComm / Frame", + "msgpack framing (Kotlin)", + shape="box", + theme=JVM, + ) + jvm.edge("server", "task", style="dotted", color=JVM[1], arrowhead="vee", label="invokes") + jvm.edge("task", "comm", style="dotted", color=JVM[1], arrowhead="none") + + # Loopback-TCP comms between the two native OS processes (JVM connects back). + g.edge( + "comm", + "supervisor", + label="msgpack frames over loopback TCP (127.0.0.1)\nJVM connects back · ToSupervisor / ToTask", + color=JVM[1], + fontcolor=JVM[1], + dir="both", + penwidth="2.2", + ) + g.edge( + "comm", + "supervisor", + label="structured logs\n(second TCP channel)", + color=JVM[1], + fontcolor="#a1674f", + style="dashed", + ) + + # ------------------------------------------------------------------ # + # API server — separate process / host. + # ------------------------------------------------------------------ # + with g.subgraph(name="cluster_server") as server: + server.attr( + label="API server · separate process / host", + labelloc="t", + style="rounded,filled", + fillcolor="#fdeeec", + color=API[1], + penwidth="1.5", + fontsize="13", + fontname="Helvetica-Bold", + margin="14", + ) + _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("execution_api", "metadata_db", style="dotted", color=API[1], dir="both") + + g.edge( + "client", + "execution_api", + label="HTTPS + task JWT\n(proxied for the JVM task)", + color=API[1], + fontcolor=API[1], + penwidth="2.2", + ) + + # Contrast caption. + _node( + g, + "note", + "How it differs", + "vs Python task: same Supervisor, but a JVM subprocess over loopback TCP
" + "instead of a forked Python process over a UNIX socketpair
" + "vs Go SDK: the JVM task does not hold the JWT and does not call the
" + "Execution API directly — the Python Supervisor proxies every call", + shape="note", + theme=NOTE, + ) + g.edge("supervisor", "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_java_sdk_execution_architecture_diagram() diff --git a/airflow-core/docs/img/diagram_java_sdk_execution_sequence.md5sum b/airflow-core/docs/img/diagram_java_sdk_execution_sequence.md5sum new file mode 100644 index 0000000000000..590d397530f46 --- /dev/null +++ b/airflow-core/docs/img/diagram_java_sdk_execution_sequence.md5sum @@ -0,0 +1 @@ +5c927fe1f455ae4fc5bf77ebf7fc7ae7 diff --git a/airflow-core/docs/img/diagram_java_sdk_execution_sequence.png b/airflow-core/docs/img/diagram_java_sdk_execution_sequence.png new file mode 100644 index 0000000000000..71cd2ebbc4ae0 Binary files /dev/null and b/airflow-core/docs/img/diagram_java_sdk_execution_sequence.png differ diff --git a/airflow-core/docs/img/diagram_java_sdk_execution_sequence.py b/airflow-core/docs/img/diagram_java_sdk_execution_sequence.py new file mode 100644 index 0000000000000..3c9d4ae96587c --- /dev/null +++ b/airflow-core/docs/img/diagram_java_sdk_execution_sequence.py @@ -0,0 +1,327 @@ +# 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", +# ] +# /// +""" +UML-style sequence diagram for a Java (JVM) task run. + +Each participant gets its own vertical lifeline; messages are horizontal arrows +between lifelines, read top to bottom. The **Supervisor** (Python) sits in the +middle so the JVM <-> Supervisor round-trip (the JVM asks for a +Connection/Variable/XCom over loopback TCP and gets the answer back) and the +Supervisor <-> Execution API round-trip are both drawn as adjacent request/ +response pairs. The JVM task never talks to the Execution API — the Supervisor +proxies every call, so the JVM never holds the task JWT. + +Arrows are colored by sender: + +* teal — Scheduler +* purple — Coordinator layer (CoordinatorManager / JavaCoordinator) +* blue — Supervisor (_JavaActivitySubprocess) → JVM / Execution API +* orange — JVM SDK runtime / user code → Supervisor (over loopback TCP) +* red — Execution API → Supervisor (responses) + +Graphviz has no native sequence-diagram shape, so lifelines are drawn as dashed +vertical edges through invisible way-points, one row per message, with each +message as a ``constraint=false`` horizontal edge on that row. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +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") + +# (fill, border) per participant — consistent with the other Task SDK diagrams. +SCHED = ("#e0f2f1", "#00695c") # scheduler (teal) +COORD = ("#ede7f6", "#5e35b1") # coordinator layer (purple) +SUP = ("#e3f2fd", "#1565c0") # supervisor (blue) +JVM = ("#fbe9e7", "#d84315") # JVM runtime + user code (orange) +API = ("#fdecea", "#c62828") # execution API (red) + +SCHED_C, COORD_C, SUP_C, JVM_C, API_C = SCHED[1], COORD[1], SUP[1], JVM[1], API[1] + +LIFELINE = "#b0bec5" + +# Participants, left to right. Supervisor is central so both round-trips are +# drawn between neighboring lifelines. +PARTICIPANTS = [ + ("sched", "Scheduler", "creates the ExecuteTask workload", SCHED), + ("jvm", "JVM subprocess", "Server · Task · user code", JVM), + ("sup", "Supervisor (Python)", "Coordinator · _JavaActivitySubprocess", SUP), + ("api", "Execution API", "FastAPI · TEI / AIP-72", API), +] + +# Each step is one row. A "msg" is an arrow between two lifelines; a "self" is an +# activation box on a single lifeline (local processing, no message). +STEPS: list[dict[str, Any]] = [ + { + "kind": "msg", + "from": "sched", + "to": "sup", + "color": SCHED_C, + "label": 'ExecuteTask workload\n(carries queue="java")', + }, + { + "kind": "self", + "actor": "sup", + "theme": COORD, + "text": 'CoordinatorManager.for_queue("java") → JavaCoordinator\nopen two loopback-TCP servers', + }, + { + "kind": "msg", + "from": "sup", + "to": "jvm", + "color": COORD_C, + "label": "subprocess.Popen(java -jar bundle)\n(spawn the JVM)", + }, + { + "kind": "msg", + "from": "jvm", + "to": "sup", + "color": JVM_C, + "label": "TCP connect back\n(comm + logs channels)", + }, + {"kind": "msg", "from": "sup", "to": "api", "color": SUP_C, "label": "PATCH .../run\n(TI started)"}, + { + "kind": "msg", + "from": "api", + "to": "sup", + "color": API_C, + "style": "dashed", + "label": "TIRunContext\n(+ heartbeat every ~N s)", + }, + { + "kind": "msg", + "from": "sup", + "to": "jvm", + "color": SUP_C, + "label": "StartupDetails (msgpack over TCP)\n→ build Context", + }, + {"kind": "self", "actor": "jvm", "theme": JVM, "text": "Task.execute(Context, Client) [USER CODE]"}, + { + "kind": "msg", + "from": "jvm", + "to": "sup", + "color": JVM_C, + "label": "getConnection / getVariable / getXCom / setXCom\n(_RequestFrame, msgpack over TCP)", + }, + { + "kind": "msg", + "from": "sup", + "to": "api", + "color": SUP_C, + "label": "GET connection / variable / xcom\n(HTTPS + task JWT)", + }, + {"kind": "msg", "from": "api", "to": "sup", "color": API_C, "style": "dashed", "label": "result"}, + { + "kind": "msg", + "from": "sup", + "to": "jvm", + "color": SUP_C, + "label": "*Result (msgpack over TCP)\nabove 4 messages repeat per lookup", + }, + { + "kind": "msg", + "from": "jvm", + "to": "sup", + "color": JVM_C, + "label": "final state (success / failed / up-for-retry)\nmsgpack over TCP", + }, + {"kind": "msg", "from": "sup", "to": "api", "color": SUP_C, "label": "PATCH .../state\n(+ upload logs)"}, + { + "kind": "msg", + "from": "sup", + "to": "sched", + "color": SCHED_C, + "style": "dashed", + "label": "JVM process exits →\nexecute_task() returns the exit code", + }, +] + + +def _label(title: str, sub: str | None = None) -> str: + html = f"<{title}" + if sub: + safe = sub.replace("\n", "
") + html += f'
{safe}' + return html + ">" + + +def _header(g, pid: str, title: str, sub: str, theme: tuple[str, str]) -> None: + fill, border = theme + g.node( + f"{pid}__h", + label=_label(title, sub), + shape="box", + style="rounded,filled", + fillcolor=fill, + color=border, + penwidth="2", + margin="0.24,0.14", + ) + + +def _activation(g, node_id: str, num: int, text: str, theme: tuple[str, str]) -> None: + fill, border = theme + g.node( + node_id, + label=_label(f"{num}", text), + shape="box", + style="rounded,filled", + fillcolor=fill, + color=border, + penwidth="2", + margin="0.2,0.12", + ) + + +def _waypoint(g, node_id: str) -> None: + g.node(node_id, shape="point", width="0.02", color=LIFELINE) + + +def _edge_label(num: int, text: str) -> str: + # Pad the description on every side: blank lines above and below keep it clear + # of the arrow line, and leading/trailing spaces on each line keep it clear of + # the lifelines on the left and right. + lines = f"{num} · {text}".split("\n") + padded = "\n".join(f" {line} " for line in lines) + return f" \n{padded}\n " + + +def _build_sequence(g) -> None: + ids = [p[0] for p in PARTICIPANTS] + + # --- participant headers, ordered left-to-right on the top rank ---------- # + with g.subgraph() as top: + top.attr(rank="same") + for pid, title, sub, theme in PARTICIPANTS: + _header(top, pid, title, sub, theme) + for a, b in zip(ids, ids[1:]): + g.edge(f"{a}__h", f"{b}__h", style="invis") + + # --- one rank per step; way-points on every lifeline, box on the actor --- # + for i, step in enumerate(STEPS): + with g.subgraph() as row: + row.attr(rank="same") + for pid, _, _, theme in PARTICIPANTS: + nid = f"{pid}__{i}" + if step["kind"] == "self" and step["actor"] == pid: + _activation(row, nid, i + 1, step["text"], step.get("theme", theme)) + else: + _waypoint(row, nid) + + # --- dashed vertical lifelines through the way-points -------------------- # + for pid, *_ in PARTICIPANTS: + chain = [f"{pid}__h"] + [f"{pid}__{i}" for i in range(len(STEPS))] + for a, b in zip(chain, chain[1:]): + g.edge(a, b, style="dashed", arrowhead="none", color=LIFELINE, penwidth="1.3") + + # --- message arrows (horizontal, do not constrain ranking) --------------- # + for i, step in enumerate(STEPS): + if step["kind"] != "msg": + continue + g.edge( + f"{step['from']}__{i}", + f"{step['to']}__{i}", + xlabel=_edge_label(i + 1, step["label"]), + color=step["color"], + fontcolor=step["color"], + style=step.get("style", "solid"), + arrowhead="vee", + penwidth="2", + constraint="false", + ) + + _legend(g) + + +def _legend(g) -> None: + entries = [ + ("lg_sched", "Scheduler → Supervisor", SCHED), + ("lg_coord", "Coordinator layer (CoordinatorManager / JavaCoordinator)", COORD), + ("lg_sup", "Supervisor → JVM / Execution API", SUP), + ("lg_jvm", "JVM (user code) → Supervisor (loopback TCP)", JVM), + ("lg_api", "Execution API → Supervisor (response)", API), + ] + with g.subgraph(name="cluster_legend") as legend: + legend.attr( + label="Arrow color = sender · steps 9–12 repeat per Connection / Variable / XCom lookup", + labelloc="t", + style="rounded,filled", + fillcolor="#fafafa", + color="#b0bec5", + fontsize="12", + fontname="Helvetica-Bold", + margin="12", + ) + for nid, text, theme in entries: + fill, border = theme + legend.node( + nid, + label=_label(text), + shape="box", + style="rounded,filled", + fillcolor=fill, + color=border, + penwidth="2", + margin="0.2,0.1", + ) + for a, b in zip([e[0] for e in entries], [e[0] for e in entries][1:]): + legend.edge(a, b, style="invis") + # Anchor the legend below the diagram, on the left. + g.edge(f"sched__{len(STEPS) - 1}", "lg_sched", style="invis") + + +def generate_java_sdk_execution_sequence_diagram(): + image_file = MY_DIR / f"{MY_FILENAME}.png" + console.print(f"[bright_blue]Generating sequence image {image_file}") + + g = graphviz.Digraph("java_sdk_execution_sequence") + g.attr( + rankdir="TB", + splines="line", + forcelabels="true", + nodesep="1.3", + ranksep="1.2", + pad="0.5", + bgcolor="white", + fontname="Helvetica", + ) + g.attr("node", fontname="Helvetica", fontsize="13", fontcolor="#102027") + g.attr("edge", fontname="Helvetica", fontsize="10", penwidth="2", color="#546e7a") + + _build_sequence(g) + + g.render(outfile=str(image_file), format="png", cleanup=True) + console.print(f"[green]Generated sequence image {image_file}") + + +if __name__ == "__main__": + generate_java_sdk_execution_sequence_diagram() diff --git a/airflow-core/docs/img/diagram_native_language_sdk_architecture.md5sum b/airflow-core/docs/img/diagram_native_language_sdk_architecture.md5sum new file mode 100644 index 0000000000000..8149286f36321 --- /dev/null +++ b/airflow-core/docs/img/diagram_native_language_sdk_architecture.md5sum @@ -0,0 +1 @@ +d7f0c11f0b175a82b2bded558ebfa0f1 diff --git a/airflow-core/docs/img/diagram_native_language_sdk_architecture.png b/airflow-core/docs/img/diagram_native_language_sdk_architecture.png new file mode 100644 index 0000000000000..b7e0a71e06db5 Binary files /dev/null and b/airflow-core/docs/img/diagram_native_language_sdk_architecture.png differ diff --git a/airflow-core/docs/img/diagram_native_language_sdk_architecture.py b/airflow-core/docs/img/diagram_native_language_sdk_architecture.py new file mode 100644 index 0000000000000..9f802620ed324 --- /dev/null +++ b/airflow-core/docs/img/diagram_native_language_sdk_architecture.py @@ -0,0 +1,258 @@ +# 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 a native-language (compiled) Task SDK. + +The Go SDK (``go-sdk/``) is the first implementation; the same shape applies to +any future compiled-language SDK (e.g. Java). Unlike the Python Task SDK there is +**no Python Supervisor and no msgpack stdin 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. + +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_native_language_sdk_architecture_diagram(): + image_file = MY_DIR / f"{MY_FILENAME}.png" + console.print(f"[bright_blue]Generating architecture image {image_file}") + + g = graphviz.Digraph("native_language_sdk_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_native_language_sdk_architecture_diagram() diff --git a/airflow-core/docs/img/diagram_task_sdk_execution_architecture.md5sum b/airflow-core/docs/img/diagram_task_sdk_execution_architecture.md5sum new file mode 100644 index 0000000000000..b647a5229b359 --- /dev/null +++ b/airflow-core/docs/img/diagram_task_sdk_execution_architecture.md5sum @@ -0,0 +1 @@ +a197a4d28941db9bff348c56c54acaa5 diff --git a/airflow-core/docs/img/diagram_task_sdk_execution_architecture.png b/airflow-core/docs/img/diagram_task_sdk_execution_architecture.png new file mode 100644 index 0000000000000..7550e570489c7 Binary files /dev/null and b/airflow-core/docs/img/diagram_task_sdk_execution_architecture.png differ diff --git a/airflow-core/docs/img/diagram_task_sdk_execution_architecture.py b/airflow-core/docs/img/diagram_task_sdk_execution_architecture.py new file mode 100644 index 0000000000000..bdaa62d71b1d4 --- /dev/null +++ b/airflow-core/docs/img/diagram_task_sdk_execution_architecture.py @@ -0,0 +1,291 @@ +# 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 Task SDK execution. + +Contrasts the two ways a task is run: + +* the *supervised* path used in production, where the Supervisor ("Coordinator") + and the Task each run in their own **native OS process** (Python interpreter) + and talk over a socket, with the Supervisor proxying the remote Execution API; and +* the *native / in-process* path (``dag.test()`` and local runs), where the whole + thing runs **inside a single Python process** with in-memory queues instead of + sockets and an in-memory Execution API instead of HTTP. + +Rendered with graphviz directly (rather than the ``diagrams`` library) so every +label sits *inside* a sized shape — nothing overlaps, and each kind of thing gets +its own shape: 3-D box = native OS process, rounded box = an object inside a +process, component = an ASGI/FastAPI app, cylinder = the database. +""" + +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, modern palette: (fill, border) per role. +TASK = ("#e8f5e9", "#2e7d32") # Task SDK / user code +SUP = ("#e3f2fd", "#1565c0") # Supervisor ("Coordinator") +API = ("#fdecea", "#c62828") # Execution API +NATIVE = ("#f3e5f5", "#7b1fa2") # in-process "native" path +NEUTRAL = ("#eceff1", "#455a64") # database + + +def _label(title: str, sub: str | None = None) -> str: + """Build an HTML-like label: bold title over a smaller, dimmer subtitle.""" + 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") 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_task_sdk_execution_architecture_diagram(): + image_file = MY_DIR / f"{MY_FILENAME}.png" + console.print(f"[bright_blue]Generating architecture image {image_file}") + + g = graphviz.Digraph("task_sdk_execution_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") + + # ------------------------------------------------------------------ # + # Production / supervised path — several native OS processes. + # ------------------------------------------------------------------ # + with g.subgraph(name="cluster_prod") as prod: + prod.attr( + label="Production (supervised) path — one worker slot = several native OS processes", + labelloc="t", + style="rounded,filled", + fillcolor="#f7fbff", + color="#90a4ae", + penwidth="1.5", + fontsize="16", + fontname="Helvetica-Bold", + margin="18", + ) + + with prod.subgraph(name="cluster_task") as task: + task.attr( + label="Task process · forked native OS process (Python) · runs USER CODE", + labelloc="t", + style="rounded,filled", + fillcolor="#edf7ee", + color=TASK[1], + penwidth="1.5", + fontsize="13", + fontname="Helvetica-Bold", + margin="14", + ) + _node( + task, + "task_runner", + "task_runner.run()", + "RuntimeTaskInstance
→ Operator.execute() [user code]
never sees the JWT · no DB access", + shape="box3d", + theme=TASK, + ) + _node( + task, + "comms", + "SUPERVISOR_COMMS", + "CommsDecoder", + shape="box", + theme=TASK, + ) + task.edge("task_runner", "comms", style="dotted", color=TASK[1], arrowhead="none") + + with prod.subgraph(name="cluster_sup") as sup: + sup.attr( + label="Supervisor process · native OS process (Python)", + labelloc="t", + style="rounded,filled", + fillcolor="#eaf3fc", + color=SUP[1], + penwidth="1.5", + fontsize="13", + fontname="Helvetica-Bold", + margin="14", + ) + _node( + sup, + "supervisor", + "ActivitySubprocess", + "Supervisor (WatchedSubprocess base)
forks + watches the task, heartbeats," + "
forwards logs, proxies every API call", + shape="box3d", + theme=SUP, + ) + _node( + sup, + "client", + "Client", + "Execution API HTTP client
holds the short-lived task JWT", + shape="box", + theme=SUP, + ) + sup.edge("supervisor", "client", style="dotted", color=SUP[1], arrowhead="none") + + # Socket comms between the two native OS processes. + g.edge( + "comms", + "supervisor", + label="length-prefixed msgpack frames\nover the stdin socket\nToSupervisor ▸ ◂ ToTask", + color=TASK[1], + fontcolor=TASK[1], + dir="both", + penwidth="2.2", + ) + g.edge( + "comms", + "supervisor", + label="log records\n(line-based JSON, logs socket)", + color=TASK[1], + fontcolor="#5d8a60", + style="dashed", + ) + + # ------------------------------------------------------------------ # + # API server — separate process / host. + # ------------------------------------------------------------------ # + with g.subgraph(name="cluster_api") as api: + api.attr( + label="API server · separate process / host", + labelloc="t", + style="rounded,filled", + fillcolor="#fdeeec", + color=API[1], + penwidth="1.5", + fontsize="13", + fontname="Helvetica-Bold", + margin="14", + ) + _node(api, "execution_api", "Execution API", "(FastAPI)", shape="component", theme=API) + _node(api, "metadata_db", "Metadata DB", "", shape="cylinder", theme=NEUTRAL) + api.edge("execution_api", "metadata_db", style="dotted", color=API[1], dir="both") + + g.edge( + "client", + "execution_api", + label="HTTPS + task JWT\n(over the network)", + color=API[1], + fontcolor=API[1], + penwidth="2.2", + ) + + # ------------------------------------------------------------------ # + # Native / in-process path — one Python process, no fork, no sockets. + # ------------------------------------------------------------------ # + with g.subgraph(name="cluster_native") as nat: + nat.attr( + label="Native (in-process) path — dag.test() / local run" + " · ONE Python process · no fork · no sockets · no HTTP", + labelloc="t", + style="rounded,filled", + fillcolor="#faf3fc", + color=NATIVE[1], + penwidth="1.5", + fontsize="16", + fontname="Helvetica-Bold", + margin="18", + ) + _node( + nat, + "in_runner", + "task_runner.run()", + "→ Operator.execute() [user code]
same Task SDK runtime", + shape="box3d", + theme=NATIVE, + ) + _node( + nat, + "in_comms", + "InProcessSupervisorComms", + "in-memory deques, not sockets", + shape="box", + theme=NATIVE, + ) + _node(nat, "in_sup", "InProcessTestSupervisor", "(ActivitySubprocess)", shape="box3d", theme=NATIVE) + _node( + nat, + "in_api", + "InProcessExecutionAPI", + "in-memory ASGI app · no network", + shape="component", + theme=NATIVE, + ) + nat.edge( + "in_runner", + "in_comms", + label="ToTask / ToSupervisor messages\nvia deque.append / popleft", + color=NATIVE[1], + fontcolor=NATIVE[1], + dir="both", + ) + nat.edge("in_comms", "in_sup", label="direct in-process call", color=NATIVE[1], fontcolor=NATIVE[1]) + nat.edge( + "in_sup", + "in_api", + label="in-process ASGI request\n(no socket, no JWT)", + color=NATIVE[1], + fontcolor=NATIVE[1], + ) + + g.render(outfile=str(image_file), format="png", cleanup=True) + console.print(f"[green]Generated architecture image {image_file}") + + +if __name__ == "__main__": + generate_task_sdk_execution_architecture_diagram() diff --git a/airflow-core/docs/img/diagram_task_sdk_execution_sequence.md5sum b/airflow-core/docs/img/diagram_task_sdk_execution_sequence.md5sum new file mode 100644 index 0000000000000..db39bb21d054a --- /dev/null +++ b/airflow-core/docs/img/diagram_task_sdk_execution_sequence.md5sum @@ -0,0 +1 @@ +c6071c060cf352f65dfaf1240aaa0755 diff --git a/airflow-core/docs/img/diagram_task_sdk_execution_sequence.png b/airflow-core/docs/img/diagram_task_sdk_execution_sequence.png new file mode 100644 index 0000000000000..8f0f11ec7081d Binary files /dev/null and b/airflow-core/docs/img/diagram_task_sdk_execution_sequence.png differ diff --git a/airflow-core/docs/img/diagram_task_sdk_execution_sequence.py b/airflow-core/docs/img/diagram_task_sdk_execution_sequence.py new file mode 100644 index 0000000000000..a3470121d081c --- /dev/null +++ b/airflow-core/docs/img/diagram_task_sdk_execution_sequence.py @@ -0,0 +1,313 @@ +# 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", +# ] +# /// +""" +UML-style sequence diagram for a supervised Task SDK task run. + +Each participant gets its own vertical lifeline; messages are horizontal arrows +between lifelines, read top to bottom. The **Supervisor** sits in the middle so +the Task <-> Supervisor round-trip (the task asks for a Connection/Variable/XCom +and gets the answer back) and the Supervisor <-> Execution API round-trip are +both drawn as adjacent request/response pairs. + +Arrows are colored by sender: + +* blue — Supervisor ("Coordinator") → Task / Execution API +* green — Task (task_runner, user code) → Supervisor +* red — Execution API (FastAPI on the API server) → Supervisor (responses) +* amber — Executor (start / end of the run) + +Graphviz has no native sequence-diagram shape, so lifelines are drawn as dashed +vertical edges through invisible way-points, one row per message, with each +message as a ``constraint=false`` horizontal edge on that row. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +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") + +# (fill, border) per participant — consistent with the other Task SDK diagrams. +EXEC = ("#fff3e0", "#ef6c00") # executor / terminal (amber) +TASK = ("#e8f5e9", "#2e7d32") # task_runner + user code (green) +SUP = ("#e3f2fd", "#1565c0") # supervisor (blue) +API = ("#fdecea", "#c62828") # execution API (red) + +EXEC_C, TASK_C, SUP_C, API_C = EXEC[1], TASK[1], SUP[1], API[1] + +LIFELINE = "#b0bec5" + +# Participants, left to right. Supervisor is central so both round-trips are +# drawn between neighboring lifelines. +PARTICIPANTS = [ + ("exec", "Executor", "supervise_task()", EXEC), + ("task", "Task", "task_runner · user code", TASK), + ("sup", "Supervisor", "ActivitySubprocess", SUP), + ("api", "Execution API", "FastAPI · TEI / AIP-72", API), +] + +# Each step is one row. A "msg" is an arrow between two lifelines; a "self" is an +# activation box on a single lifeline (local processing, no message). +STEPS: list[dict[str, Any]] = [ + { + "kind": "msg", + "from": "exec", + "to": "sup", + "color": EXEC_C, + "label": "supervise_task()\nvia BaseExecutor.run_workload", + }, + { + "kind": "self", + "actor": "sup", + "theme": SUP, + "text": "ActivitySubprocess.start()\nfork the task process, open the sockets", + }, + {"kind": "msg", "from": "sup", "to": "api", "color": SUP_C, "label": "PATCH .../run\n(TI started)"}, + { + "kind": "msg", + "from": "api", + "to": "sup", + "color": API_C, + "style": "dashed", + "label": "TIRunContext\n(+ heartbeat every ~N s)", + }, + { + "kind": "msg", + "from": "sup", + "to": "task", + "color": SUP_C, + "label": "StartupDetails\n(_ResponseFrame, msgpack —\nthe one un-prompted message)", + }, + { + "kind": "self", + "actor": "task", + "theme": TASK, + "text": "parse() → RuntimeTaskInstance\nrun() → Operator.execute() [USER CODE]", + }, + { + "kind": "msg", + "from": "task", + "to": "sup", + "color": TASK_C, + "label": "GetConnection / GetVariable / GetXCom\n(_RequestFrame, msgpack over socket)", + }, + { + "kind": "msg", + "from": "sup", + "to": "api", + "color": SUP_C, + "label": "GET connection / variable / xcom\n(HTTPS + task JWT)", + }, + {"kind": "msg", "from": "api", "to": "sup", "color": API_C, "style": "dashed", "label": "result"}, + { + "kind": "msg", + "from": "sup", + "to": "task", + "color": SUP_C, + "label": "*Result (_ResponseFrame, msgpack)\nabove 4 messages repeat per lookup", + }, + { + "kind": "msg", + "from": "task", + "to": "sup", + "color": TASK_C, + "label": "SucceedTask / DeferTask / RetryTask\n(final state, msgpack over socket)", + }, + {"kind": "msg", "from": "sup", "to": "api", "color": SUP_C, "label": "PATCH .../state\n(+ upload logs)"}, + { + "kind": "msg", + "from": "sup", + "to": "exec", + "color": EXEC_C, + "style": "dashed", + "label": "task process exits →\nsupervise_task() returns the exit code", + }, +] + + +def _label(title: str, sub: str | None = None) -> str: + html = f"<{title}" + if sub: + safe = sub.replace("\n", "
") + html += f'
{safe}' + return html + ">" + + +def _header(g, pid: str, title: str, sub: str, theme: tuple[str, str]) -> None: + fill, border = theme + g.node( + f"{pid}__h", + label=_label(title, sub), + shape="box", + style="rounded,filled", + fillcolor=fill, + color=border, + penwidth="2", + margin="0.24,0.14", + ) + + +def _activation(g, node_id: str, num: int, text: str, theme: tuple[str, str]) -> None: + fill, border = theme + g.node( + node_id, + label=_label(f"{num}", text), + shape="box", + style="rounded,filled", + fillcolor=fill, + color=border, + penwidth="2", + margin="0.2,0.12", + ) + + +def _waypoint(g, node_id: str) -> None: + g.node(node_id, shape="point", width="0.02", color=LIFELINE) + + +def _edge_label(num: int, text: str) -> str: + # Pad the description on every side: blank lines above and below keep it clear + # of the arrow line, and leading/trailing spaces on each line keep it clear of + # the lifelines on the left and right. + lines = f"{num} · {text}".split("\n") + padded = "\n".join(f" {line} " for line in lines) + return f" \n{padded}\n " + + +def _build_sequence(g) -> None: + ids = [p[0] for p in PARTICIPANTS] + + # --- participant headers, ordered left-to-right on the top rank ---------- # + with g.subgraph() as top: + top.attr(rank="same") + for pid, title, sub, theme in PARTICIPANTS: + _header(top, pid, title, sub, theme) + for a, b in zip(ids, ids[1:]): + g.edge(f"{a}__h", f"{b}__h", style="invis") + + # --- one rank per step; way-points on every lifeline, box on the actor --- # + for i, step in enumerate(STEPS): + with g.subgraph() as row: + row.attr(rank="same") + for pid, _, _, theme in PARTICIPANTS: + nid = f"{pid}__{i}" + if step["kind"] == "self" and step["actor"] == pid: + _activation(row, nid, i + 1, step["text"], step.get("theme", theme)) + else: + _waypoint(row, nid) + + # --- dashed vertical lifelines through the way-points -------------------- # + for pid, *_ in PARTICIPANTS: + chain = [f"{pid}__h"] + [f"{pid}__{i}" for i in range(len(STEPS))] + for a, b in zip(chain, chain[1:]): + g.edge(a, b, style="dashed", arrowhead="none", color=LIFELINE, penwidth="1.3") + + # --- message arrows (horizontal, do not constrain ranking) --------------- # + for i, step in enumerate(STEPS): + if step["kind"] != "msg": + continue + g.edge( + f"{step['from']}__{i}", + f"{step['to']}__{i}", + xlabel=_edge_label(i + 1, step["label"]), + color=step["color"], + fontcolor=step["color"], + style=step.get("style", "solid"), + arrowhead="vee", + penwidth="2", + constraint="false", + ) + + _legend(g) + + +def _legend(g) -> None: + entries = [ + ("lg_sup", "Supervisor → Task / Execution API", SUP), + ("lg_task", "Task (user code) → Supervisor", TASK), + ("lg_api", "Execution API → Supervisor (response)", API), + ("lg_exec", "Executor (start / end of the run)", EXEC), + ] + with g.subgraph(name="cluster_legend") as legend: + legend.attr( + label="Arrow color = sender · steps 7–10 repeat per Connection / Variable / XCom lookup", + labelloc="t", + style="rounded,filled", + fillcolor="#fafafa", + color="#b0bec5", + fontsize="12", + fontname="Helvetica-Bold", + margin="12", + ) + for nid, text, theme in entries: + fill, border = theme + legend.node( + nid, + label=_label(text), + shape="box", + style="rounded,filled", + fillcolor=fill, + color=border, + penwidth="2", + margin="0.2,0.1", + ) + for a, b in zip([e[0] for e in entries], [e[0] for e in entries][1:]): + legend.edge(a, b, style="invis") + # Anchor the legend below the diagram, on the left. + g.edge(f"exec__{len(STEPS) - 1}", "lg_sup", style="invis") + + +def generate_task_sdk_execution_sequence_diagram(): + image_file = MY_DIR / f"{MY_FILENAME}.png" + console.print(f"[bright_blue]Generating sequence image {image_file}") + + g = graphviz.Digraph("task_sdk_execution_sequence") + g.attr( + rankdir="TB", + splines="line", + forcelabels="true", + nodesep="1.3", + ranksep="1.2", + pad="0.5", + bgcolor="white", + fontname="Helvetica", + ) + g.attr("node", fontname="Helvetica", fontsize="13", fontcolor="#102027") + g.attr("edge", fontname="Helvetica", fontsize="10", penwidth="2", color="#546e7a") + + _build_sequence(g) + + g.render(outfile=str(image_file), format="png", cleanup=True) + console.print(f"[green]Generated sequence image {image_file}") + + +if __name__ == "__main__": + generate_task_sdk_execution_sequence_diagram() diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 40eff22af35a6..c41eb5427fdf6 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -1299,6 +1299,8 @@ proto Protobuf protobuf provisioner +proxied +proxies psql psrp psycopg