You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "shivaam (via GitHub)" <gi...@apache.org> on 2026/06/15 00:41:46 UTC
[PR] Add TypeScript task coordinator [airflow]
shivaam opened a new pull request, #68548:
URL: https://github.com/apache/airflow/pull/68548
This adds the Python-side coordinator needed to run TypeScript tasks through Airflow coordinator mode.
It pairs with the public TypeScript Task SDK interface in #67908. A Python Dag still defines the scheduling graph with stub tasks, while the task implementation can live in a TypeScript bundle. At execution time, Airflow routes the stub task to this coordinator, which starts the Node.js bundle through the existing supervisor subprocess path.
The expected bundle layout is:
```text
/opt/airflow/ts-bundles/
bundle.mjs
airflow-metadata.yaml
```
`bundle.mjs` is the Node.js entrypoint for the TypeScript task runtime. `airflow-metadata.yaml` describes the bundle and provides the supervisor schema version the coordinator should use when talking to the runtime.
Example metadata:
```yaml
airflow_bundle_metadata_version: "1.0"
sdk:
language: typescript
version: "0.1.0-alpha.0"
supervisor_schema_version: "2026-06-16"
source: src/airflow.ts
dags:
sales_pipeline:
tasks:
- extract
- transform
```
For this first PR, the coordinator only needs `sdk.supervisor_schema_version`, but the metadata file follows the same general shape as Airflow executable bundle metadata so TypeScript bundle tooling can grow into it later.
Example Airflow config:
```ini
[sdk]
coordinators = {
"ts": {
"classpath": "airflow.sdk.coordinators.typescript.TypescriptCoordinator",
"kwargs": {"bundles_root": ["/opt/airflow/ts-bundles"]}
}
}
queue_to_coordinator = {"typescript": "ts"}
```
This first version keeps bundle selection deliberately simple. `bundles_root` is an ordered fallback list of directories, and the first valid `bundle.mjs` with valid metadata is used. Dag/task-aware selection across multiple TypeScript bundles is left for follow-up work once the TypeScript bundle tooling exists.
### Testing
I ran a full local daemon E2E with `api-server`, `dag-processor`, and `scheduler`.
The E2E run verified successful TypeScript task execution for variable access, XComs, connections, and SIGTERM cancellation handling.
### Follow-up work
- Add the TypeScript SDK runtime PR with `startCoordinator()`.
- Add an official TypeScript pack/build tool that generates `bundle.mjs` and `airflow-metadata.yaml`.
- Add Dag/task-aware multi-bundle selection after bundle tooling exists.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes - OpenAI Codex
Generated-by: OpenAI Codex following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488351506
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
Review Comment:
I agree the single artifact model would avoid drift between bundle.mjs and its metadata compared to two file model. For this PR I’d keep the sidecar airflow-metadata.yaml and raise a follow up PR later once the full sdk and coordinator is shipped.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3487559898
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
Review Comment:
I wonder do we even need separated artifacts here?
For Java, the Jar is basically a zip file. For Go, the binary is self contained the metadata at the end of the binary.
How about having the metadata as the comment (perhaps with special format) within the .mjs as single artifact?
##########
task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py:
##########
@@ -0,0 +1,28 @@
+#
+# 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.
+"""
+TypeScript runtime coordinator for the Apache Airflow Task SDK.
+
+See :class:`TypescriptCoordinator` for details.
+"""
+
+from __future__ import annotations
+
+from airflow.sdk.coordinators.typescript.coordinator import TypescriptCoordinator
+
+__all__ = ["TypescriptCoordinator"]
Review Comment:
Agree with TP on the naming, NodeCoordinator make sense to me.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488026464
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
Review Comment:
A block comment at the and of the mjs would work. I believe this can be done automatically with a build-time plugin (more similar to Java’s Gradle approach).
This can be addressed after this PR though.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript task coordinator [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488536778
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -0,0 +1,171 @@
+#
+# 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.
+"""Node.js runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.node")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
+
+
+def _validate_schema_version(instance, _, value) -> str:
+ return get_schema_version_migrator().resolve_version(str(value))
+
+
+@attrs.define
+class _NodeBundle:
+ path: pathlib.Path
+ schema_version: str = attrs.field(validator=_validate_schema_version)
+
+
+def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]:
+ if not metadata_path.is_file():
+ raise ValueError(f"missing {METADATA_FILENAME}")
+
+ try:
+ with metadata_path.open(encoding="utf-8") as metadata_file:
+ data = yaml.safe_load(metadata_file)
+ except OSError as exc:
+ raise ValueError(f"cannot read {METADATA_FILENAME}: {exc}") from exc
+ except yaml.YAMLError as exc:
+ raise ValueError(f"cannot parse {METADATA_FILENAME}: {exc}") from exc
+
+ if not isinstance(data, dict):
+ raise ValueError(f"{METADATA_FILENAME} must contain a mapping")
+ return data
+
+
+def _supervisor_schema_version(metadata: dict[str, Any]) -> str:
+ sdk = metadata.get("sdk")
+ if not isinstance(sdk, dict):
+ raise ValueError("missing sdk metadata mapping")
+
+ value = sdk.get("supervisor_schema_version")
+ if not isinstance(value, str) or not value:
+ raise ValueError("missing or invalid sdk.supervisor_schema_version")
+ return value
+
+
+def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _NodeBundle:
+ """
+ Locate the ``.mjs`` entry point in *bundles_root*.
+
+ Scans each configured directory for ``bundle.mjs`` and reads the sibling
+ ``airflow-metadata.yaml`` for the bundle's supervisor schema version.
+
+ This is an ordered fallback search, not Dag/task-aware multi-bundle
+ routing. The first bundle found wins. A future version can use the
+ metadata's ``dags`` section together with ``TaskInstance.dag_id`` and
+ ``TaskInstance.task_id`` to select the bundle that owns a specific task.
+ """
+ rejected: list[tuple[pathlib.Path, str]] = []
+ for root in bundles_root:
+ candidate = root / BUNDLE_FILENAME
+ if not candidate.is_file():
+ continue
+ try:
+ metadata = _read_bundle_metadata(root / METADATA_FILENAME)
+ log.debug("Selected TypeScript bundle", path=candidate, root=root)
+ return _NodeBundle(
+ path=candidate,
+ schema_version=_supervisor_schema_version(metadata),
+ )
+ except (TypeError, ValueError) as exc:
+ log.debug(
+ "TypeScript bundle metadata rejected; skipping", path=candidate, root=root, error=str(exc)
+ )
Review Comment:
```suggestion
log.debug(
"TypeScript bundle metadata rejected; skipping",
path=candidate,
root=root,
exc_info=True,
)
```
This would keep the traceback for debuggability. (`exc_info` is a special key)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488570315
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -0,0 +1,171 @@
+#
+# 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.
+"""Node.js runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.node")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
+
+
+def _validate_schema_version(instance, _, value) -> str:
+ return get_schema_version_migrator().resolve_version(str(value))
+
+
+@attrs.define
+class _NodeBundle:
+ path: pathlib.Path
+ schema_version: str = attrs.field(validator=_validate_schema_version)
+
+
+def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]:
+ if not metadata_path.is_file():
+ raise ValueError(f"missing {METADATA_FILENAME}")
+
+ try:
+ with metadata_path.open(encoding="utf-8") as metadata_file:
+ data = yaml.safe_load(metadata_file)
+ except OSError as exc:
+ raise ValueError(f"cannot read {METADATA_FILENAME}: {exc}") from exc
+ except yaml.YAMLError as exc:
+ raise ValueError(f"cannot parse {METADATA_FILENAME}: {exc}") from exc
+
+ if not isinstance(data, dict):
+ raise ValueError(f"{METADATA_FILENAME} must contain a mapping")
+ return data
+
+
+def _supervisor_schema_version(metadata: dict[str, Any]) -> str:
+ sdk = metadata.get("sdk")
+ if not isinstance(sdk, dict):
+ raise ValueError("missing sdk metadata mapping")
+
+ value = sdk.get("supervisor_schema_version")
+ if not isinstance(value, str) or not value:
+ raise ValueError("missing or invalid sdk.supervisor_schema_version")
+ return value
+
+
+def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _NodeBundle:
+ """
+ Locate the ``.mjs`` entry point in *bundles_root*.
+
+ Scans each configured directory for ``bundle.mjs`` and reads the sibling
+ ``airflow-metadata.yaml`` for the bundle's supervisor schema version.
+
+ This is an ordered fallback search, not Dag/task-aware multi-bundle
+ routing. The first bundle found wins. A future version can use the
+ metadata's ``dags`` section together with ``TaskInstance.dag_id`` and
+ ``TaskInstance.task_id`` to select the bundle that owns a specific task.
+ """
+ rejected: list[tuple[pathlib.Path, str]] = []
+ for root in bundles_root:
+ candidate = root / BUNDLE_FILENAME
+ if not candidate.is_file():
+ continue
+ try:
+ metadata = _read_bundle_metadata(root / METADATA_FILENAME)
+ log.debug("Selected TypeScript bundle", path=candidate, root=root)
+ return _NodeBundle(
+ path=candidate,
+ schema_version=_supervisor_schema_version(metadata),
+ )
+ except (TypeError, ValueError) as exc:
+ log.debug(
+ "TypeScript bundle metadata rejected; skipping", path=candidate, root=root, error=str(exc)
+ )
Review Comment:
Done. Will remember this for future. :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3487265289
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
+
+
+def _validate_schema_version(instance, _, value) -> str:
+ return get_schema_version_migrator().resolve_version(str(value))
+
+
+@attrs.define
+class _TypescriptBundle:
+ path: pathlib.Path
+ schema_version: str = attrs.field(validator=_validate_schema_version)
+
+
+def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]:
+ if not metadata_path.is_file():
+ raise ValueError(f"missing {METADATA_FILENAME}")
+
+ try:
+ data = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
+ except OSError as exc:
+ raise ValueError(f"cannot read {METADATA_FILENAME}: {exc}") from exc
+ except yaml.YAMLError as exc:
+ raise ValueError(f"cannot parse {METADATA_FILENAME}: {exc}") from exc
+
+ if not isinstance(data, dict):
+ raise ValueError(f"{METADATA_FILENAME} must contain a mapping")
+ return data
+
+
+def _supervisor_schema_version(metadata: dict[str, Any]) -> str:
+ sdk = metadata.get("sdk")
+ if not isinstance(sdk, dict):
+ raise ValueError("missing sdk metadata mapping")
+
+ value = sdk.get("supervisor_schema_version")
+ if not isinstance(value, str) or not value:
+ raise ValueError("missing or invalid sdk.supervisor_schema_version")
+ return value
+
+
+def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _TypescriptBundle:
+ """
+ Locate the ``.mjs`` entry point in *bundles_root*.
+
+ Scans each configured directory for ``bundle.mjs`` and reads the sibling
+ ``airflow-metadata.yaml`` for the bundle's supervisor schema version.
+
+ This is an ordered fallback search, not Dag/task-aware multi-bundle
+ routing. The first bundle found wins. A future version can use the
+ metadata's ``dags`` section together with ``TaskInstance.dag_id`` and
+ ``TaskInstance.task_id`` to select the bundle that owns a specific task.
+ """
+ for root in bundles_root:
+ candidate = root / BUNDLE_FILENAME
+ if not candidate.is_file():
+ continue
+ metadata = _read_bundle_metadata(root / METADATA_FILENAME)
+ log.debug("Selected TypeScript bundle", path=candidate, root=root)
+ return _TypescriptBundle(
+ path=candidate,
+ schema_version=_supervisor_schema_version(metadata),
Review Comment:
If metadata in a bundle is malformed, this would cause an unhandled ValueError bubble out of the coordinator. This is inconsistent with other coordinators; both Java and Executable would discard the artifact with a warning and keep scanning instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #68548:
URL: https://github.com/apache/airflow/pull/68548#issuecomment-4823331400
I think the class name should be `TypeScriptCoordinator` since that’s the canonical spelling. The module name can probably just be `ts` but that’s not a big deal, either way works.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #68548:
URL: https://github.com/apache/airflow/pull/68548#issuecomment-4824420408
I’m thinking, maybe the coordinator should be NodeCoordinator instead. Go does not have a GoCoordinator but a more general ExecutableCoordinator. Java has JavaCoordinator, but that’s not (mainly) because the SDK targets Java (it doesn’t _just_ target Java but all JVM languages), but more because the command that runs the SDK bundle is called `java` (even though the command runs a JVM—it’s what it is). So NodeCoordinator arguably is more suitable than TypeScriptCoordinator since the former better describes what the coordinator actually does, on the same level as the other existing coordinators.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488426745
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
+
+
+def _validate_schema_version(instance, _, value) -> str:
+ return get_schema_version_migrator().resolve_version(str(value))
+
+
+@attrs.define
+class _TypescriptBundle:
+ path: pathlib.Path
+ schema_version: str = attrs.field(validator=_validate_schema_version)
+
+
+def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]:
+ if not metadata_path.is_file():
+ raise ValueError(f"missing {METADATA_FILENAME}")
+
+ try:
+ data = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
Review Comment:
Done. I agree that we can refactor it with executable coordinator as a separate PR.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488425801
##########
task-sdk/src/airflow/sdk/coordinators/typescript/__init__.py:
##########
@@ -0,0 +1,28 @@
+#
+# 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.
+"""
+TypeScript runtime coordinator for the Apache Airflow Task SDK.
+
+See :class:`TypescriptCoordinator` for details.
+"""
+
+from __future__ import annotations
+
+from airflow.sdk.coordinators.typescript.coordinator import TypescriptCoordinator
+
+__all__ = ["TypescriptCoordinator"]
Review Comment:
Udpated
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488351506
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
Review Comment:
I agree the single-artifact model would avoid drift between bundle.mjs and its metadata, and that is probably where the packer should go. For this PR I’d keep the sidecar airflow-metadata.yaml because it is easier to inspect/debug and keeps the coordinator minimal. I will raise a follow up PR for this.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3488362797
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
+
+
+def _validate_schema_version(instance, _, value) -> str:
+ return get_schema_version_migrator().resolve_version(str(value))
+
+
+@attrs.define
+class _TypescriptBundle:
+ path: pathlib.Path
+ schema_version: str = attrs.field(validator=_validate_schema_version)
+
+
+def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]:
+ if not metadata_path.is_file():
+ raise ValueError(f"missing {METADATA_FILENAME}")
+
+ try:
+ data = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
+ except OSError as exc:
+ raise ValueError(f"cannot read {METADATA_FILENAME}: {exc}") from exc
+ except yaml.YAMLError as exc:
+ raise ValueError(f"cannot parse {METADATA_FILENAME}: {exc}") from exc
+
+ if not isinstance(data, dict):
+ raise ValueError(f"{METADATA_FILENAME} must contain a mapping")
+ return data
+
+
+def _supervisor_schema_version(metadata: dict[str, Any]) -> str:
+ sdk = metadata.get("sdk")
+ if not isinstance(sdk, dict):
+ raise ValueError("missing sdk metadata mapping")
+
+ value = sdk.get("supervisor_schema_version")
+ if not isinstance(value, str) or not value:
+ raise ValueError("missing or invalid sdk.supervisor_schema_version")
+ return value
+
+
+def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> _TypescriptBundle:
+ """
+ Locate the ``.mjs`` entry point in *bundles_root*.
+
+ Scans each configured directory for ``bundle.mjs`` and reads the sibling
+ ``airflow-metadata.yaml`` for the bundle's supervisor schema version.
+
+ This is an ordered fallback search, not Dag/task-aware multi-bundle
+ routing. The first bundle found wins. A future version can use the
+ metadata's ``dags`` section together with ``TaskInstance.dag_id`` and
+ ``TaskInstance.task_id`` to select the bundle that owns a specific task.
+ """
+ for root in bundles_root:
+ candidate = root / BUNDLE_FILENAME
+ if not candidate.is_file():
+ continue
+ metadata = _read_bundle_metadata(root / METADATA_FILENAME)
+ log.debug("Selected TypeScript bundle", path=candidate, root=root)
+ return _TypescriptBundle(
+ path=candidate,
+ schema_version=_supervisor_schema_version(metadata),
Review Comment:
Switched to keep scanning for the bundle so it aligns with java and go now.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #68548:
URL: https://github.com/apache/airflow/pull/68548#issuecomment-4827016204
> I’m thinking, maybe the coordinator should be NodeCoordinator instead. Go does not have a GoCoordinator but a more general ExecutableCoordinator. Java has JavaCoordinator, but that’s not (mainly) because the SDK targets Java (it doesn’t _just_ target Java but all JVM languages), but more because the command that runs the SDK bundle is called `java` (even though the command runs a JVM—it’s what it is). So NodeCoordinator arguably is more suitable than TypeScriptCoordinator since the former better describes what the coordinator actually does, on the same level as the other existing coordinators.
Yes, that makes sense. I will change it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add NodeCoordinator [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #68548:
URL: https://github.com/apache/airflow/pull/68548
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Do not merge] Add TypeScript task coordinator [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68548:
URL: https://github.com/apache/airflow/pull/68548#discussion_r3487065185
##########
task-sdk/src/airflow/sdk/coordinators/typescript/coordinator.py:
##########
@@ -0,0 +1,158 @@
+#
+# 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.
+"""TypeScript runtime coordinator that launches a Node.js subprocess for task execution."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+from typing import TYPE_CHECKING, Any
+
+import attrs
+import structlog
+import yaml
+
+from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.execution_time.schema import get_schema_version_migrator
+
+if TYPE_CHECKING:
+ from collections.abc import Sequence
+
+ from structlog.typing import FilteringBoundLogger
+
+ from airflow.sdk.api.datamodels._generated import TaskInstance
+
+log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.typescript")
+
+BUNDLE_FILENAME = "bundle.mjs"
+METADATA_FILENAME = "airflow-metadata.yaml"
+
+
+def _validate_schema_version(instance, _, value) -> str:
+ return get_schema_version_migrator().resolve_version(str(value))
+
+
+@attrs.define
+class _TypescriptBundle:
+ path: pathlib.Path
+ schema_version: str = attrs.field(validator=_validate_schema_version)
+
+
+def _read_bundle_metadata(metadata_path: pathlib.Path) -> dict[str, Any]:
+ if not metadata_path.is_file():
+ raise ValueError(f"missing {METADATA_FILENAME}")
+
+ try:
+ data = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
Review Comment:
This can probably use yaml.safe_load against `metadata_path.open()` instead (a file handle) to save some memory. Not a big deal.
A side note: we should probably refactor this a bit with ExecutableCoordinator to avoid duplicated logic on metadata and schema version parsing. This is for a follow-up PR instead of this one.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org