[DON'T MERGE] Go-SDK with Coordinator interface

[jason810496]

Disclaimer

The coordinator interface of this branch is not up to date with the latest discuss on Dev List.
I will wait until the final decision settle down then make the interface up to date and break down this PR into smaller pieces.

The implementation on this branch already worked for both "multi-lang tasks" and "defining the whole Dag in Go" features anyway.

Try it out

1. Checkout to this feature branch
2. in files/airflow-breeze-config/environment_variables.env

AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST='[{"name": "go-bundles", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {"path": "/files/go-bundles", "refresh_interval": 20}}, {"name": "multi-lang", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {"path": "/files/multi-lang", "refresh_interval": 20}}]'
AIRFLOW__SDK__QUEUE_TO_SDK='{"go-task": "executable"}'
AIRFLOW__EXECUTABLE__BUNDLES_FOLDER='/files/multi-lang-go-bundles'
AIRFLOW_CONN_TEST_HTTP='{"conn_type": "http", "login": "user", "password": "pass", "host": "example.com", "port": 1234, "extra": {"param1": "val1", "param2": "val2"}}'
AIRFLOW_VAR_MY_VARIABLE=123

3. breeze start-airflow --backend postgres
4. in Breeze container shell tab

mkdir -p /opt/airflow/go-sdk/bin /files/go-bundles /files/multi-lang-go-bundles /files/multi-lang

# Build #1: simple_dag (pure Go)
cd /opt/airflow/go-sdk/example/bundle && go tool airflow-go-pack \
    --output /opt/airflow/go-sdk/bin/example_dags_simple \
    -- -ldflags "-X main.dagId=simple_dag"
cp /opt/airflow/go-sdk/bin/example_dags_simple /files/go-bundles/example_dags
sudo chown airflow:airflow /files/go-bundles/example_dags

# Build #2: go_multi_lang (Python stub backend)
cd /opt/airflow/go-sdk/example/bundle && go tool airflow-go-pack \
    --output /opt/airflow/go-sdk/bin/example_dags_stub \
    -- -ldflags "-X main.dagId=go_multi_lang"
cp /opt/airflow/go-sdk/bin/example_dags_stub /files/multi-lang-go-bundles/example_dags
sudo chown airflow:airflow /files/multi-lang-go-bundles/example_dags

# Old coordinator interface (not up to date with the latest discussion, but it works)
cd /opt/airflow && pip install providers/sdk/executable

# create Stub Dag if not already created. dag_id must match the dagId baked
# into the /files/multi-lang-go-bundles/ binary above ("go_multi_lang"), and the stub
# must expose one @task.stub per Go task the scheduler should trigger.
cat <<EOF > /files/multi-lang/stub_dag.py
from airflow.sdk import dag, task

@task.stub(queue="go-task")
def extract(): ...

@task.stub(queue="go-task")
def transform(): ...

@task
def load():
    print("This is a Python task that depends on two Go tasks!")

@dag()
def go_multi_lang():
    extract() >> transform() >> load()

go_multi_lang()
EOF

5. Then restart dag-processor and scheduler (press r at scheduler and dag-processor tab)

Note

The same Go source (go-sdk/example/bundle/main.go) is built twice with a different main.dagId override per build:

• dagId=simple_dag -> pure-Go bundle, dropped into /files/go-bundles/
• dagId=go_multi_lang -> stub-mode binary, dropped into /files/multi-lang-go-bundles/ next to stub_dag.py (which declares the matching dag_id on the Python side).

airflow-go-pack forwards everything after -- straight to go build, so we pass -ldflags "-X main.dagId=..." to overwrite the package-level dagId string in the example bundle.