This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 65dad4f851c [v3-4-test] Run the lang-SDK native Dag tests on
Kubernetes (#75930) (#76008)
65dad4f851c is described below
commit 65dad4f851c8b3ba24501730b3cb8262d3a2ba6f
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 30 17:40:00 2026 +0800
[v3-4-test] Run the lang-SDK native Dag tests on Kubernetes (#75930)
(#76008)
* Name the native Dag k8s tests so the lang-SDK job selects them
The lang-SDK Kubernetes job runs `-k TestLangSdk`, which matched only
TestLangSdkMixedLanguageDag. The native Java and TypeScript classes were
deselected, so Kubernetes verified stub tasks only.
* Run the native Go Dag on Kubernetes
Go Dag parsing landed with ExecutableDagImporter, and the compose e2e covers
it, but the Kubernetes harness provisioned no Go native Dag bundle, so the
Go class was a skipped placeholder.
`breeze k8s setup-lang-sdk-test` now packs `go-sdk/example/native` into a
`lang-sdk-native-go` bucket, served as a Dag bundle. The single
ExecutableCoordinator already parses any executable bundle in every Dag
bundle, and the `golang` queue is already routed, so no coordinator config
changes.
* Serve the typescript_example handlers from ts-task-handlers
The native TypeScript Dag's trigger_downstream starts a typescript_example
run and waits for it, but that Dag's stub tasks could not start: the only
bundle in ts-task-handlers declares lang_sdk_mixed_language, so every task
failed to launch and the awaited run failed. The gap was invisible while
the native Dag tests were deselected.
---------
(cherry picked from commit 8401a190fa8174e8089914408fe82797a202f71b)
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
.../airflow_breeze/commands/kubernetes_commands.py | 106 +++++++++++++++-----
.../tests/test_kubernetes_lang_sdk_commands.py | 44 +++++++--
kubernetes-tests/lang_sdk/README.md | 18 ++--
kubernetes-tests/lang_sdk/config/values.yaml | 9 +-
.../kubernetes_tests/test_lang_sdk_native_dag.py | 110 ++++++++++++++++++---
5 files changed, 232 insertions(+), 55 deletions(-)
diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
index c980fcd1d13..dfc24ed0e0c 100644
--- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
+++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
@@ -2548,6 +2548,9 @@ LANG_SDK_PATH = AIRFLOW_ROOT_PATH / "kubernetes-tests" /
"lang_sdk"
# ts_example is a standalone pnpm project that resolves the SDK through a
file: dependency.
LANG_SDK_GO_EXAMPLE_PATH = LANG_SDK_PATH / "go_example"
LANG_SDK_GO_BUNDLE_NAME = "lang_sdk_mixed_language"
+# The native Go Dag fixture the compose e2e also packs; a package of the
go-sdk module itself.
+LANG_SDK_GO_NATIVE_BUNDLE_PKG = "./example/native"
+LANG_SDK_GO_NATIVE_BUNDLE_NAME = "go_native_dags"
LANG_SDK_JAVA_EXAMPLE_PATH = LANG_SDK_PATH / "java_example"
# The reusable native Java Dag fixture the compose e2e also builds.
LANG_SDK_JAVA_NATIVE_BUNDLE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" /
"java-native-bundle"
@@ -2574,6 +2577,9 @@ LANG_SDK_AIRFLOWIGNORE_ALL = LANG_SDK_PATH /
"airflowignore_all"
LANG_SDK_TS_EXAMPLE_PATH = AIRFLOW_ROOT_PATH / "ts-sdk" / "example"
LANG_SDK_TS_BUILDER_IMAGE = os.environ.get("NODE_BUILDER_IMAGE",
"node:22-alpine")
LANG_SDK_TS_BUNDLE_NAME = "bundle.min.mjs"
+# ts-sdk/example also holds the handlers of the Python-declared
typescript_example Dag, so it
+# goes into ts-task-handlers as well, under a name that does not clash with
the bundle there.
+LANG_SDK_TS_NATIVE_HANDLER_NAME = "native_bundle.min.mjs"
LANG_SDK_AWS_CONN_URI = (
"aws://test:test@/?region_name=us-east-1&"
"endpoint_url=http%3A%2F%2Flocalstack.airflow.svc.cluster.local%3A4566"
@@ -2679,7 +2685,12 @@ def _lang_sdk_fetch_upstream_sdk_sources(staging: Path,
output: Output | None) -
def _lang_sdk_build_go_bundle(
staging: Path, go_sdk_source: Path, output: Output | None, *, native: bool
= False
) -> None:
- """Build the Go bundle into ``staging/go-task-handlers`` and copy the
result into the staging dir.
+ """Pack the mixed-language Go bundle and the native Go Dag bundle into
their staging dirs.
+
+ ``go_example`` holds the handlers of the mixed-language Dag's stub tasks
+ (``staging/go-task-handlers``). ``go-sdk/example/native`` declares the
``go_native_*`` Dags with
+ no Python file, and is packed into ``staging/lang-sdk-native-go``, where
the Dag processor
+ parses it.
By default the build runs in an ephemeral Go toolchain container so the
host needs no Go install,
writing its caches into a gitignored dir under the repo. In ``native``
mode (used in CI, where the
@@ -2694,17 +2705,22 @@ def _lang_sdk_build_go_bundle(
"""
go_dir = staging / "go-task-handlers"
go_dir.mkdir(parents=True, exist_ok=True)
+ native_go_dir = staging / "lang-sdk-native-go"
+ native_go_dir.mkdir(parents=True, exist_ok=True)
example_rel = LANG_SDK_GO_EXAMPLE_PATH.relative_to(AIRFLOW_ROOT_PATH)
workspace = staging / "go_workspace"
example_path = workspace / example_rel
+ go_sdk_path = workspace / "go-sdk"
example_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(LANG_SDK_GO_EXAMPLE_PATH, example_path,
ignore=shutil.ignore_patterns(".home"))
# In dry-run the fetch/extract commands are skipped, so the extracted copy
and build outputs
# never materialize -- skip the filesystem work that depends on them.
if not get_dry_run():
- shutil.copytree(go_sdk_source, workspace / "go-sdk")
+ shutil.copytree(go_sdk_source, go_sdk_path)
output_bin = example_path / "bin" / LANG_SDK_GO_BUNDLE_NAME
output_bin.parent.mkdir(parents=True, exist_ok=True)
+ native_output_bin = go_sdk_path / "bin" / LANG_SDK_GO_NATIVE_BUNDLE_NAME
+ native_output_bin.parent.mkdir(parents=True, exist_ok=True)
# CGO_ENABLED=0 yields a fully static binary that runs on the stock
worker. The package built is
# the current dir (".") because go_example is its own module.
@@ -2725,6 +2741,20 @@ def _lang_sdk_build_go_bundle(
output=output,
check=True,
)
+ run_command(
+ [
+ "go",
+ "tool",
+ "airflow-go-pack",
+ "--output",
+ str(native_output_bin),
+ LANG_SDK_GO_NATIVE_BUNDLE_PKG,
+ ],
+ cwd=go_sdk_path,
+ env=go_env,
+ output=output,
+ check=True,
+ )
else:
uid_gid = f"{os.getuid()}:{os.getgid()}"
go_example_ctr = f"/repo/{example_rel.as_posix()}"
@@ -2733,30 +2763,33 @@ def _lang_sdk_build_go_bundle(
# the real go_example's gitignored cache dir so the caches persist
across scratch workspaces.
(LANG_SDK_GO_EXAMPLE_PATH / ".home").mkdir(parents=True, exist_ok=True)
get_console(output=output).print(f"[info]Building Go bundle in
{LANG_SDK_GO_BUILDER_IMAGE}")
- docker_base = [
- "docker",
- "run",
- "--rm",
- "--user",
- uid_gid,
- "-e",
- f"HOME={go_example_ctr}/.home",
- "-e",
- "USER=airflow",
- "-e",
- "CGO_ENABLED=0",
- "-v",
- f"{workspace}:/repo",
- "-v",
- f"{LANG_SDK_GO_EXAMPLE_PATH / '.home'}:{go_example_ctr}/.home",
- "-w",
- go_example_ctr,
- LANG_SDK_GO_BUILDER_IMAGE,
- ]
- run_command([*docker_base, "go", "mod", "tidy"], output=output,
check=True)
+
+ def docker_base(workdir: str) -> list[str]:
+ return [
+ "docker",
+ "run",
+ "--rm",
+ "--user",
+ uid_gid,
+ "-e",
+ f"HOME={go_example_ctr}/.home",
+ "-e",
+ "USER=airflow",
+ "-e",
+ "CGO_ENABLED=0",
+ "-v",
+ f"{workspace}:/repo",
+ "-v",
+ f"{LANG_SDK_GO_EXAMPLE_PATH / '.home'}:{go_example_ctr}/.home",
+ "-w",
+ workdir,
+ LANG_SDK_GO_BUILDER_IMAGE,
+ ]
+
+ run_command([*docker_base(go_example_ctr), "go", "mod", "tidy"],
output=output, check=True)
run_command(
[
- *docker_base,
+ *docker_base(go_example_ctr),
"go",
"tool",
"airflow-go-pack",
@@ -2767,8 +2800,22 @@ def _lang_sdk_build_go_bundle(
output=output,
check=True,
)
+ run_command(
+ [
+ *docker_base("/repo/go-sdk"),
+ "go",
+ "tool",
+ "airflow-go-pack",
+ "--output",
+ f"/repo/go-sdk/bin/{LANG_SDK_GO_NATIVE_BUNDLE_NAME}",
+ LANG_SDK_GO_NATIVE_BUNDLE_PKG,
+ ],
+ output=output,
+ check=True,
+ )
if not get_dry_run():
shutil.copy(output_bin, go_dir / LANG_SDK_GO_BUNDLE_NAME)
+ shutil.copy(native_output_bin, native_go_dir /
LANG_SDK_GO_NATIVE_BUNDLE_NAME)
def _lang_sdk_build_ts_bundles(staging: Path, output: Output | None, *,
native: bool = False) -> None:
@@ -3047,7 +3094,11 @@ def _lang_sdk_upload_artifacts(
``.airflowignore`` that excludes everything in it, so the Dag processor
never tries to
parse the packed Go bundle or the Java/TypeScript handler-only bundles as
Dags -- a
coordinator's own artifact lookup does not read ``.airflowignore``, so
stub tasks still
- find their handlers there. The two ``lang-sdk-native-*`` buckets carry no
such file.
+ find their handlers there. The three ``lang-sdk-native-*`` buckets carry
no such file.
+
+ ``ts-task-handlers`` holds two bundles: the mixed-language one and
``ts-sdk/example``, which
+ carries the handlers of ``typescript_example``. The coordinator picks
between them by the
+ dag_ids each declares, so the native Dag's ``trigger_downstream`` has a
run that can succeed.
"""
pod = run_command_with_k8s_env(
[
@@ -3070,6 +3121,7 @@ def _lang_sdk_upload_artifacts(
).stdout.strip()
go_bundle = staging / "go-task-handlers" / LANG_SDK_GO_BUNDLE_NAME
+ native_go_bundle = staging / "lang-sdk-native-go" /
LANG_SDK_GO_NATIVE_BUNDLE_NAME
native_ts_bundle = staging / "lang-sdk-native-ts" / LANG_SDK_TS_BUNDLE_NAME
handler_ts_bundle = staging / "ts-task-handlers" / LANG_SDK_TS_BUNDLE_NAME
if get_dry_run():
@@ -3096,12 +3148,14 @@ def _lang_sdk_upload_artifacts(
(downstream_dag, "/tmp/typescript_example.py"),
(native_java_jar, "/tmp/native_app.jar"),
(native_ts_bundle, "/tmp/native_bundle.min.mjs"),
+ (native_go_bundle, "/tmp/native_go_bundle"),
]
buckets = [
"go-task-handlers",
"java-task-handlers",
"ts-task-handlers",
"dags",
+ "lang-sdk-native-go",
"lang-sdk-native-java",
"lang-sdk-native-ts",
]
@@ -3116,6 +3170,8 @@ def _lang_sdk_upload_artifacts(
("/tmp/typescript_example.py", "s3://dags/typescript_example.py"),
("/tmp/native_app.jar",
f"s3://lang-sdk-native-java/{native_java_jar.name}"),
("/tmp/native_bundle.min.mjs",
f"s3://lang-sdk-native-ts/{LANG_SDK_TS_BUNDLE_NAME}"),
+ ("/tmp/native_bundle.min.mjs",
f"s3://ts-task-handlers/{LANG_SDK_TS_NATIVE_HANDLER_NAME}"),
+ ("/tmp/native_go_bundle",
f"s3://lang-sdk-native-go/{LANG_SDK_GO_NATIVE_BUNDLE_NAME}"),
]
for src, dest in copies:
diff --git a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
index e1d10abd019..ea74987c0b6 100644
--- a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
+++ b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
@@ -53,8 +53,9 @@ def go_example(tmp_path, monkeypatch):
def upstream_go_sdk(tmp_path):
"""A fake extracted upstream-main go-sdk copy, distinguishable from
anything under go_example."""
sdk_dir = tmp_path / "upstream_go_sdk"
- sdk_dir.mkdir()
+ (sdk_dir / "bin").mkdir(parents=True)
(sdk_dir / "marker.go").write_text("upstream")
+ (sdk_dir / "bin" /
kubernetes_commands.LANG_SDK_GO_NATIVE_BUNDLE_NAME).write_text("binary")
return sdk_dir
@@ -95,27 +96,32 @@ class TestLangSdkBuildGoBundle:
def test_native_uses_host_go_toolchain(self, mock_run, tmp_path,
go_example, upstream_go_sdk):
_lang_sdk_build_go_bundle(tmp_path, upstream_go_sdk, None, native=True)
- cmd = mock_run.call_args.args[0]
+ tidy_call, handlers_call, native_call = mock_run.call_args_list
+ cmd = handlers_call.args[0]
assert cmd[:3] == ["go", "tool", "airflow-go-pack"]
assert "docker" not in cmd
- workspace_example = mock_run.call_args.kwargs["cwd"]
+ workspace_example = handlers_call.kwargs["cwd"]
# The build runs against a scratch workspace copy, not the real
go_example dir.
assert workspace_example != go_example
assert workspace_example.name == go_example.name
- assert mock_run.call_args.kwargs["env"]["CGO_ENABLED"] == "0"
+ assert handlers_call.kwargs["env"]["CGO_ENABLED"] == "0"
# The workspace mirrors the repo layout, with go-sdk swapped for the
upstream copy.
- assert (workspace_example.parent / "go-sdk" / "marker.go").read_text()
== "upstream"
+ workspace_go_sdk = workspace_example.parent / "go-sdk"
+ assert (workspace_go_sdk / "marker.go").read_text() == "upstream"
assert (tmp_path / "go-task-handlers" /
kubernetes_commands.LANG_SDK_GO_BUNDLE_NAME).exists()
# The scratch copy is re-tidied against the upstream go-sdk before
packing, in the same dir.
- tidy_call = mock_run.call_args_list[0]
assert tidy_call.args[0] == ["go", "mod", "tidy"]
assert tidy_call.kwargs["cwd"] == workspace_example
+ assert native_call.args[0][:3] == ["go", "tool", "airflow-go-pack"]
+ assert native_call.args[0][-1] ==
kubernetes_commands.LANG_SDK_GO_NATIVE_BUNDLE_PKG
+ assert native_call.kwargs["cwd"] == workspace_go_sdk
+ assert (tmp_path / "lang-sdk-native-go" /
kubernetes_commands.LANG_SDK_GO_NATIVE_BUNDLE_NAME).exists()
@mock.patch.object(kubernetes_commands, "run_command")
def test_container_mode_runs_in_docker(self, mock_run, tmp_path,
go_example, upstream_go_sdk):
_lang_sdk_build_go_bundle(tmp_path, upstream_go_sdk, None,
native=False)
- cmd = mock_run.call_args.args[0]
+ tidy_cmd, cmd, native_cmd = (call.args[0] for call in
mock_run.call_args_list)
assert cmd[0] == "docker"
assert kubernetes_commands.LANG_SDK_GO_BUILDER_IMAGE in cmd
# The workspace, not AIRFLOW_ROOT_PATH, is mounted at /repo; the
persistent HOME cache
@@ -126,10 +132,12 @@ class TestLangSdkBuildGoBundle:
home_mount = next(m for m in mounts if m.endswith("/.home"))
assert home_mount.startswith(str(go_example / ".home"))
# The scratch copy is re-tidied in the same container image before
packing.
- tidy_cmd = mock_run.call_args_list[0].args[0]
assert tidy_cmd[0] == "docker"
assert kubernetes_commands.LANG_SDK_GO_BUILDER_IMAGE in tidy_cmd
assert tidy_cmd[-3:] == ["go", "mod", "tidy"]
+ assert native_cmd[native_cmd.index("-w") + 1] == "/repo/go-sdk"
+ assert native_cmd[-1] ==
kubernetes_commands.LANG_SDK_GO_NATIVE_BUNDLE_PKG
+ assert (tmp_path / "lang-sdk-native-go" /
kubernetes_commands.LANG_SDK_GO_NATIVE_BUNDLE_NAME).exists()
@pytest.fixture
@@ -380,6 +388,9 @@ class TestLangSdkDryRun:
_lang_sdk_build_go_bundle(tmp_path, tmp_path /
"missing_upstream_go_sdk", None, native=True)
assert not (tmp_path / "go-task-handlers" /
kubernetes_commands.LANG_SDK_GO_BUNDLE_NAME).exists()
+ assert not (
+ tmp_path / "lang-sdk-native-go" /
kubernetes_commands.LANG_SDK_GO_NATIVE_BUNDLE_NAME
+ ).exists()
def test_build_java_jars_skips_jar_copies(
self, dry_run, tmp_path, java_example, java_native_bundle,
upstream_java_sdk
@@ -403,6 +414,23 @@ class TestLangSdkDryRun:
assert str(tmp_path / "lang-sdk-native-java" / "app.jar") in cp_sources
+class TestLangSdkUploadArtifacts:
+ @mock.patch.object(kubernetes_commands, "run_command_with_k8s_env")
+ def
test_native_ts_bundle_also_serves_the_typescript_example_handlers(self,
mock_run, dry_run, tmp_path):
+ """typescript_example's stub tasks look in ts-task-handlers, so the
bundle has to be there too."""
+ mock_run.return_value = mock.Mock(stdout="")
+
+ _lang_sdk_upload_artifacts(tmp_path, "3.11", "v1.35.0", None)
+
+ uploads = [
+ tuple(argv[-2:]) for call in mock_run.call_args_list if "awslocal"
in (argv := call.args[0])
+ ]
+ assert ("/tmp/native_bundle.min.mjs",
"s3://lang-sdk-native-ts/bundle.min.mjs") in uploads
+ assert ("/tmp/native_bundle.min.mjs",
"s3://ts-task-handlers/native_bundle.min.mjs") in uploads
+ # The mixed-language handlers keep their own name, so the two do not
overwrite each other.
+ assert ("/tmp/handler_bundle.min.mjs",
"s3://ts-task-handlers/bundle.min.mjs") in uploads
+
+
class TestSetupLangSdkTestNativeSelection:
@pytest.mark.parametrize(
("env_value", "expected_native"),
diff --git a/kubernetes-tests/lang_sdk/README.md
b/kubernetes-tests/lang_sdk/README.md
index 3ee89620669..a8be41a73cc 100644
--- a/kubernetes-tests/lang_sdk/README.md
+++ b/kubernetes-tests/lang_sdk/README.md
@@ -25,8 +25,11 @@ Two end-to-end tests exercise every lang-SDK coordinator
(Go, Java, TypeScript)
- **`test_lang_sdk_mixed_language.py`** runs one Dag mixing **Python + Go +
Java + TypeScript**
tasks, using the `extra.pod_template_file` routing in the `[sdk]
coordinators` config.
- **`test_lang_sdk_native_dag.py`** runs one Dag **declared entirely in each
language SDK** (no
- Python file at all): the Dag processor asks the packed Java jar / TypeScript
bundle to parse
- itself. Go has no native Dag parsing support yet, so its class is
unconditionally skipped.
+ Python file at all): the Dag processor asks the packed Go binary / Java jar
/ TypeScript bundle
+ to parse itself.
+
+Both files name their classes `TestLangSdk*`, because the CI job that
provisions this environment
+selects its tests with `-k TestLangSdk`.
Every coordinator is configured once, with no `[sdk]
dag_bundle_to_coordinator` entry needed:
with exactly one coordinator per language, there is nothing to disambiguate.
@@ -46,6 +49,7 @@ with exactly one coordinator per language, there is nothing
to disambiguate.
then marks the matched file
executable) → forks the
Go binary / Java jar /
TypeScript bundle
+ lang-sdk-native-go ── S3DagBundle ──► dag-processor runs the binary
(go-sdk coordinator)
lang-sdk-native-java ── S3DagBundle ──► dag-processor parses the jar
(java-sdk coordinator)
lang-sdk-native-ts ── S3DagBundle ──► dag-processor parses the bundle
(ts-sdk coordinator)
```
@@ -61,7 +65,7 @@ Key points:
that excludes everything in them, so the Dag processor never tries to parse
the packed Go bundle
or the Java/TypeScript handler-only bundles as Dags. A coordinator's own
artifact lookup does not
read `.airflowignore`, so stub tasks still find their handlers there.
-- `lang-sdk-native-java` and `lang-sdk-native-ts` carry no such file: parsing
them is the point.
+- The three `lang-sdk-native-*` buckets carry no such file: parsing them is
the point.
## Components
@@ -79,15 +83,17 @@ Key points:
The Go binary, Java jar, TypeScript bundle and the Dag files share one object
store (localstack)
but live in **separate buckets** (`go-task-handlers`, `java-task-handlers`,
`ts-task-handlers`,
-`dags`, `lang-sdk-native-java`, `lang-sdk-native-ts`).
+`dags`, `lang-sdk-native-go`, `lang-sdk-native-java`, `lang-sdk-native-ts`).
The native TypeScript bundle is `ts-sdk/example`, packed by `airflow-ts-pack`.
It carries both
halves of that example: the handlers for the Python-declared
`typescript_example` Dag, and the
natively declared `typescript_native_example`. `typescript_example.py` is
uploaded to the `dags`
bucket too, because the native Dag's trigger task starts a run of it and waits
for that run,
deferring to `DagStateTrigger` in the Python triggerer. The test un-pauses
`typescript_example`
-for that reason. The native Java Dag reuses
`airflow-e2e-tests/java-native-bundle`, the same fixture
-the compose e2e builds.
+for that reason, and the same bundle is uploaded to `ts-task-handlers` under a
second name, since
+that is where `typescript_example`'s own stub tasks look for their handlers.
The coordinator picks
+between the two bundles in that bucket by the dag_ids each declares. The
native Java Dag reuses `airflow-e2e-tests/java-native-bundle` and the native
+Go Dags reuse `go-sdk/example/native`, the same fixtures the compose e2e
builds.
## Which SDK sources get built
diff --git a/kubernetes-tests/lang_sdk/config/values.yaml
b/kubernetes-tests/lang_sdk/config/values.yaml
index 0e628696fa0..262f919d0e9 100644
--- a/kubernetes-tests/lang_sdk/config/values.yaml
+++ b/kubernetes-tests/lang_sdk/config/values.yaml
@@ -44,7 +44,7 @@ scheduler:
mountPath: /opt/airflow/lang_sdk/pod_templates
# Serve the baked-in example Dags (dags-folder), the lang-SDK stub Dag, every
-# coordinator's task-handler bundle, and the two native Dag bundles. This must
be
+# coordinator's task-handler bundle, and the three native Dag bundles. This
must be
# additive: the standard KubernetesExecutor tests share this run and trigger
# example Dags, so dropping dags-folder marks them stale and their runs 404.
# Configure it here (dagProcessor.dagBundleConfigList) rather than overriding
@@ -55,7 +55,7 @@ scheduler:
# that excludes every file in them, so the Dag processor never parses the
# packed Go bundle or the Java/TypeScript handler-only bundles as Dags; a
# coordinator's own artifact lookup does not read .airflowignore, so stub tasks
-# still find their handlers there. The two lang-sdk-native-* bundles carry no
+# still find their handlers there. The three lang-sdk-native-* bundles carry no
# such file: that is exactly what the Dag processor is meant to parse.
dagProcessor:
dagBundleConfigList:
@@ -82,6 +82,11 @@ dagProcessor:
kwargs:
bucket_name: "ts-task-handlers"
aws_conn_id: "aws_localstack"
+ - name: lang-sdk-native-go
+ classpath: "airflow.providers.amazon.aws.bundles.s3.S3DagBundle"
+ kwargs:
+ bucket_name: "lang-sdk-native-go"
+ aws_conn_id: "aws_localstack"
- name: lang-sdk-native-java
classpath: "airflow.providers.amazon.aws.bundles.s3.S3DagBundle"
kwargs:
diff --git
a/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_native_dag.py
b/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_native_dag.py
index 8f39649f42a..5af20530b6c 100644
--- a/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_native_dag.py
+++ b/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_native_dag.py
@@ -19,11 +19,14 @@ End-to-end tests of Dags declared entirely in a language
SDK (no Python file at
KubernetesExecutor.
The Dag processor parses each native Dag through the coordinator its language
routes to,
-from a Dag bundle holding only that language's artifact
(``lang-sdk-native-java`` /
-``lang-sdk-native-ts``). The same coordinators also serve the mixed-language
Dag's stub
-tasks (see ``test_lang_sdk_mixed_language.py``); a native task ignores
-``task_handler_bundle_name`` and reads its own Dag bundle instead, so one
coordinator per
-language is enough -- no ``[sdk] dag_bundle_to_coordinator`` is configured.
+from a Dag bundle holding only that language's artifact
(``lang-sdk-native-go`` /
+``lang-sdk-native-java`` / ``lang-sdk-native-ts``). The same coordinators also
serve the
+mixed-language Dag's stub tasks (see ``test_lang_sdk_mixed_language.py``); a
native task
+ignores ``task_handler_bundle_name`` and reads its own Dag bundle instead, so
one coordinator
+per language is enough -- no ``[sdk] dag_bundle_to_coordinator`` is configured.
+
+The class names all start with ``TestLangSdk`` because the CI job that
provisions this
+environment selects its tests with ``-k TestLangSdk``.
Prerequisites are provisioned by ``breeze k8s setup-lang-sdk-test``.
"""
@@ -45,18 +48,97 @@ _skip_unless_lang_sdk = pytest.mark.skipif(
)
[email protected](
- reason="Native Go Dag parsing is not supported yet: there is no Go Dag
importer, and "
- "ExecutableCoordinator has no parse-Dag command. Unskip once a Go bundle
can answer a "
- "Dag-parse request."
-)
-class TestNativeGoDagOnKubernetes(BaseK8STest):
+@_skip_unless_lang_sdk
+class TestLangSdkNativeGoDag(BaseK8STest):
+ """Declared entirely in Go; the same ``go-sdk/example/native`` the compose
e2e packs.
+
+ Covers the path a Dag with no Python file takes: the Dag processor runs
the packed
+ ``go_native_dags`` binary through the Executable coordinator to parse it,
the scheduler reads
+ the serialized Dag, and each task runs in its own pod on the ``golang``
queue the Dag sets.
+ The graph has nested task groups, a fan-in, an ``If``, a ``Switch`` and a
trigger-rule join, so
+ pod-per-task scheduling is exercised against branches that skip rather
than a linear chain.
+ """
+
+ DAG_ID = "go_native_pipeline"
+ TASK_IDS = [
+ "seed",
+ "extract.regions.north",
+ "extract.regions.south",
+ "extract.summarize",
+ "anyRows",
+ "loadRows",
+ "pick",
+ "publishWeekly",
+ "cleanup",
+ ]
+ # The Else side of the If, and the Case the Switch does not choose.
+ SKIPPED_TASK_IDS = ["reportEmpty", "publishDaily"]
+
+ def _ensure_variable(self, key: str, value: str) -> None:
+ resp = self.session.post(f"http://{self.host}/variables";, json={"key":
key, "value": value})
+ assert resp.status_code in (200, 201, 409), f"Could not create
variable {key}: {resp.text}"
+
+ def _ensure_connection(self) -> None:
+ """Create the ``test_http`` Connection that ``seed`` reads, as the
compose e2e defines it."""
+ resp = self.session.post(
+ f"http://{self.host}/connections";,
+ json={
+ "connection_id": "test_http",
+ "conn_type": "http",
+ "login": "user",
+ "password": "pass",
+ "host": "example.com",
+ "port": 1234,
+ "extra": '{"param1": "val1", "param2": "val2"}',
+ },
+ )
+ assert resp.status_code in (200, 201, 409), f"Could not create
connection test_http: {resp.text}"
+
+ @pytest.mark.execution_timeout(900)
def test_native_go_dag_succeeds(self):
- pass
+ # The cadence names the case the Switch chooses, so `publishDaily` is
the side that skips.
+ # Both regions are non-empty literals, so the If takes its Then branch
and `reportEmpty`
+ # is the side that skips.
+ self._ensure_variable("go_native_greeting", "hello from the k8s test")
+ self._ensure_variable("go_native_cadence", "weekly")
+ self._ensure_connection()
+
+ dag_run_id, logical_date = self.start_job_in_kubernetes(self.DAG_ID,
self.host)
+ print(f"Triggered {self.DAG_ID} run {dag_run_id}
(logical_date={logical_date})")
+
+ for task_id in self.TASK_IDS:
+ self.monitor_task(
+ host=self.host,
+ dag_run_id=dag_run_id,
+ dag_id=self.DAG_ID,
+ task_id=task_id,
+ expected_final_state="success",
+ timeout=_TIMEOUT,
+ )
+
+ # The branch that was not taken is skipped, not failed, which is what
proves the skip
+ # reached the supervisor rather than the task erroring.
+ for skipped in self.SKIPPED_TASK_IDS:
+ self.monitor_task(
+ host=self.host,
+ dag_run_id=dag_run_id,
+ dag_id=self.DAG_ID,
+ task_id=skipped,
+ expected_final_state="skipped",
+ timeout=_TIMEOUT,
+ )
+
+ self.ensure_dag_expected_state(
+ host=self.host,
+ logical_date=logical_date,
+ dag_id=self.DAG_ID,
+ expected_final_state="success",
+ timeout=_TIMEOUT,
+ )
@_skip_unless_lang_sdk
-class TestNativeJavaDagOnKubernetes(BaseK8STest):
+class TestLangSdkNativeJavaDag(BaseK8STest):
"""Declared entirely in Java; the interface API half of
``airflow-e2e-tests/java-native-bundle``.
Covers the path a Dag with no Python file takes: the Dag processor parses
the
@@ -127,7 +209,7 @@ class TestNativeJavaDagOnKubernetes(BaseK8STest):
@_skip_unless_lang_sdk
-class TestNativeTypeScriptDagOnKubernetes(BaseK8STest):
+class TestLangSdkNativeTypeScriptDag(BaseK8STest):
"""The same Dag the SDK and Airflow e2e suites run, under the Node
coordinator.
Covers the path a Dag with no Python file takes: the Dag processor parses
the
