shivaam commented on code in PR #72046:
URL: https://github.com/apache/airflow/pull/72046#discussion_r3996857664
##########
task-sdk/src/airflow/sdk/coordinators/node/coordinator.py:
##########
@@ -45,80 +40,46 @@
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.
- """
+def _select_bundle(bundles_root: Sequence[pathlib.Path], dag_id: str) ->
ResolvedBundle:
+ """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
- 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),
- )
- except (TypeError, ValueError) as exc:
+ if not candidate.is_file():
+ continue
+ metadata = read_bundle(candidate)
+ if dag_id not in metadata.dag_ids:
+ log.debug(
+ "TypeScript bundle does not contain requested Dag;
skipping",
+ path=candidate,
+ root=root,
+ dag_id=dag_id,
+ )
+ continue
+ bundle = ResolvedBundle(path=candidate,
schema_version=metadata.supervisor_schema_version)
+ except (OSError, TypeError, ValueError) as exc:
log.debug(
- "TypeScript bundle metadata rejected; skipping",
+ "TypeScript bundle rejected; skipping",
path=candidate,
root=root,
- exc_info=True,
+ reason=str(exc),
Review Comment:
Kept exc_info
--
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]
