jason810496 commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r4003576658
##########
ts-sdk/tests/cli/fixtures/bundle-v1.mjs:
##########
@@ -0,0 +1,24 @@
+//#
airflowBundle={"code":{"start":"0000000000000203","end":"0000000000000592","sha256":"f814358e0d4aa5d38c10c365515179049171bb9d4bba5b80c06ea549d6f16337"},"metadata":{"start":"000000000000013e","end":"0000000000000202","sha256":"a51dfd6f0c9e8ea867900e55c0387b556d3cb0e98321b62d4625f522ed465041"}}
Review Comment:
I just realized the `.mjs` here isn't the min js.
From my perspative, we should compile into `.min.js` so that user ourself
only need to deploy one artifact instead of multiple artifacts in the DagBundle.
The shape I imagined of `.mjs` is
- the current metadata layout as comments
- have a multiple line comments that comment out the visible entry point
file source
- then the actual compiled min JS content (should be human unreadable)
Otherwise the `code` in the `airflowBundle` means nothing.
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -19,106 +19,75 @@
from __future__ import annotations
-import base64
import os
import pathlib
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
import attrs
import structlog
-from airflow.sdk.coordinators._bundle_metadata import (
- ResolvedBundle,
- convert_roots,
- extract_supervisor_schema_version,
- parse_metadata_mapping,
-)
+from airflow.sdk.coordinators._bundle_metadata import ResolvedBundle,
convert_roots
from airflow.sdk.coordinators._subprocess import SubprocessCoordinator
+from airflow.sdk.coordinators.node._bundle_reader import read_bundle
if TYPE_CHECKING:
from collections.abc import Sequence
from structlog.typing import FilteringBoundLogger
+ from typing_extensions import Self
from airflow.sdk.api.datamodels._generated import TaskInstance
log: FilteringBoundLogger =
structlog.get_logger(logger_name="coordinators.node")
BUNDLE_FILENAME = "bundle.mjs"
-EMBEDDED_METADATA_MARKER = b"//# airflowMetadata="
-EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024
-def _read_embedded_metadata(bundle_path: pathlib.Path) -> dict[str, Any]:
- """
- Read the manifest ``airflow-ts-pack`` embeds in the bundle itself.
-
- The packer prepends the metadata as a leading
- ``//# airflowMetadata=<base64>`` line comment, keeping bundle and metadata
- a single artifact. Raises ``ValueError`` when the bundle has no such
marker.
- """
- try:
- with bundle_path.open("rb") as bundle_file:
- line = bundle_file.readline(EMBEDDED_METADATA_MAX_BYTES + 1)
- except OSError as exc:
- raise ValueError(f"cannot read {bundle_path.name}: {exc}") from exc
-
- if not line.startswith(EMBEDDED_METADATA_MARKER):
- raise ValueError(f"{bundle_path.name} has no embedded airflow
metadata; rebuild with airflow-ts-pack")
- if len(line) > EMBEDDED_METADATA_MAX_BYTES:
- raise ValueError(
- f"embedded airflow metadata exceeds {EMBEDDED_METADATA_MAX_BYTES}
bytes; "
- f"rebuild {bundle_path.name} with airflow-ts-pack"
- )
-
- payload = line[len(EMBEDDED_METADATA_MARKER) :].strip()
- try:
- decoded = base64.b64decode(payload, validate=True)
- except ValueError as exc:
- raise ValueError(f"cannot parse embedded airflow metadata: {exc}")
from exc
- return parse_metadata_mapping(decoded, source="embedded airflow metadata")
-
-
-def _find_bundle(bundles_root: Sequence[pathlib.Path]) -> ResolvedBundle:
- """
- Locate the ``.mjs`` entry point in *bundles_root*.
-
- Scans each configured directory for ``bundle.mjs`` and reads the bundle's
- supervisor schema version from the metadata embedded in the bundle.
-
- 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_embedded_metadata(candidate)
- log.debug("Selected TypeScript bundle", path=candidate, root=root)
- return ResolvedBundle(
- path=candidate,
- schema_version=extract_supervisor_schema_version(metadata),
[email protected]
+class _Bundle(ResolvedBundle):
+ @classmethod
+ def find(cls, bundles_root: Sequence[pathlib.Path], dag_id: str) -> Self:
+ """Return the first verified configured bundle that declares
*dag_id*."""
+ rejected: list[tuple[pathlib.Path, str]] = []
+ for root in bundles_root:
+ candidate = root / BUNDLE_FILENAME
Review Comment:
Just found out that we hard coded the artifact file name.
--
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]
