Copilot commented on code in PR #68221:
URL: https://github.com/apache/airflow/pull/68221#discussion_r3372884587
##########
go-sdk/sdk/errors.go:
##########
@@ -33,4 +33,8 @@ var VariableNotFound = errors.New("variable not found")
// See the “GetConnection“ method of [ConnectionClient] for an example
var ConnectionNotFound = errors.New("connection not found")
+// XComNotFound is an error value used to signal that an XCom value could not
be found (and that there were
+// no communication issues to the API server).
+//
+// See the “GetXCom“ method of [XComClient] for an example
var XComNotFound = errors.New("xcom not found")
Review Comment:
The docstrings for the NotFound sentinel errors use the phrase
"communication issues to the API server", which is ungrammatical; it should be
"with the API server". Since this PR adds the XComNotFound docstring, it’s a
good opportunity to correct the wording consistently across all three sentinels
for clearer GoDoc output.
##########
go-sdk/README.md:
##########
@@ -74,63 +251,97 @@ Since Go is a compiled language (putting aside projects
such as [YAEGI](https://
secret_key: "u0ZDb2ccINAbhzNmvYzclw=="
```
- You can also set these options via environment variables of
`AIRFLOW__${section}_${key}`, for example `AIRFLOW__API_AUTH__SECRET_KEY`.
+ You can also set these options via environment variables of
`AIRFLOW__${SECTION}__${KEY}`, for example
+ `AIRFLOW__API_AUTH__SECRET_KEY`.
-- Install the worker
+- Install the worker:
```bash
go install github.com/apache/airflow/go-sdk/cmd/airflow-go-edge-worker@latest
```
-- Run it!
+- Run it:
```bash
airflow-go-edge-worker run --queues golang
```
-### Example Dag:
+- Deploy the matching Python stub Dag (above) into Airflow.
-You will need to create a python Dag and deploy it in to the Airflow
+## Known limitations
-```python
-from airflow.sdk import dag, task
+A non-exhaustive list of features the **Edge Worker (go-plugin)** path has yet
to implement. These are the
+main reason the coordinator-based path is recommended: in that 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 (i.e. S3/GCS etc.).
+- XCom reading/writing through non-default XCom backends.
[email protected](queue="golang")
-def extract(): ...
+## How it works
+The same bundle binary speaks two different protocols; which one it uses is
decided at launch by the CLI
+flags it was invoked with. User code (`func main`) is identical either way.
[email protected](queue="golang")
-def transform(): ...
+### Coordinator protocol
+```
+Python supervisor / task runner
+ │ finds + validates the bundle, then forks it:
+ ▼
+ <bundle binary> --comm=127.0.0.1:P1 --logs=127.0.0.1:P2
+ │ binary dials BACK over TCP loopback (msgpack-over-IPC)
+ ▼
+ GetConnection / GetVariable / GetXCom / SetXCom ... → SucceedTask / TaskState
+```
-@dag()
-def simple_dag():
+- The Python `ExecutableCoordinator` forks the bundle binary with
`--comm`/`--logs` addresses it is already
+ listening on. The binary dials back (it never listens) and speaks a
length-prefixed msgpack-over-IPC wire
+ protocol on the comm socket, with structured JSON-line logs on the logs
socket.
+- The Python runtime is the worker. It proxies every `GetConnection` /
`GetVariable` / `GetXCom` /
+ `SetXCom` call through to the Execution API. The Go binary just runs the
task function.
- extract() >> transform()
+The Go side of the protocol is implemented in `pkg/execution/`. On the Python
side it is the
+`ExecutableCoordinator` in
`task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py`.
+### Edge Worker protocol
-simple_dag()
+```
+Airflow scheduler ──Edge Executor API──► airflow-go-edge-worker
──go-plugin/gRPC──► bundle binary
+ (ExecuteTaskWorkload) (long-running Go process)
(child process)
```
-Here we see the `@task.stub` which tells the Dag parser about the "shape" of
the go tasks, and lets us define
-the relationships between them
-
-> [!NOTE]
-> Yes, you still have to have a python Dag file for now. This is a known
limitation at the moment.
+- `airflow-go-edge-worker` is a long-running Go process. It registers with the
scheduler, polls the Edge
+ Executor API for `ExecuteTaskWorkload`s, and heartbeats.
+- For each workload it execs the bundle binary as a child and connects over
HashiCorp
+ [`go-plugin`](https://github.com/hashicorp/go-plugin) (gRPC over a
handshake-gated socket).
+- The Task API itself has no way to deliver an `ExecuteTaskWorkload` to a Go
worker, so the Edge Executor
+ API fills that gap. Longer term that API will likely need stabilising and
versioning.
-## Known missing features
+## Architectural decisions
-A non-exhaustive list of features we have yet to implement
+The [`adr/`](./adr) directory records the design decisions behind the SDK:
-- Support for putting tasks into state other than success or
failed/up-for-retry (deferred, failed-without-retries etc.)
-- Remote task logs (i.e. S3/GCS etc)
-- XCom reading/writing from other XCom backends
+- [ADR 0001](./adr/0001-bundle-packing-options.md): bundle-packing options.
+- [ADR 0002](./adr/0002-use-go-tool-directive-for-bundle-packer.md): deliver
the bundle packer via the
+ Go 1.24 `tool` directive.
+- [ADR 0003](./adr/0003-coordinator-protocol-msgpack-ipc.md): dual-mode
coordinator protocol, where one
+ binary speaks both go-plugin gRPC (Edge Worker) and msgpack-over-IPC (Python
coordinator).
+- [ADR 0004](./adr/0004-self-contained-executable-bundle.md): the
self-contained executable bundle, where
+ the executable *is* the bundle.
+The normative, language-agnostic on-disk bundle format (the footer layout,
manifest fields, and what the
+`ExecutableCoordinator` reads) is specified in
+[`executable-bundle-spec.rst`](../task-sdk/docs/executable-bundle-spec.rst).
`airflow-go-pack` produces
+bundles conforming to that spec.
Review Comment:
The README mentions it will be published on pkg.go.dev. Relative links that
traverse outside the module root (like ../task-sdk/...) won’t resolve there
because pkg.go.dev only has the go-sdk module’s file tree. Using an absolute
GitHub (or docs-site) URL here avoids a broken link for readers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
