shahar1 commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4052465517
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -136,6 +179,54 @@ def refresh(self) -> None:
delete_stale=True,
)
+ def _refresh_from_archive(self) -> None:
+ """Stage the Dag bundle by downloading and unpacking the single
archive object."""
+ client = self.s3_hook.get_conn()
+ head = client.head_object(Bucket=self.bucket_name,
Key=self.archive_key)
+ etag: str = head.get("ETag", "")
+
+ marker = self.s3_dags_dir / self.archive_etag_marker
+ if etag and marker.is_file() and marker.read_text() == etag:
+ self._log.debug(
+ "Dag bundle archive 's3://%s/%s' is unchanged (ETag %s),
skipping staging",
+ self.bucket_name,
+ self.archive_key,
+ etag,
+ )
+ return
+
+ staging_dir = Path(tempfile.mkdtemp(dir=self.s3_dags_dir.parent,
prefix=".s3-archive-staging-"))
+ try:
+ archive_path = staging_dir / "_bundle_archive"
+ client.download_file(self.bucket_name, self.archive_key,
os.fspath(archive_path))
+
+ unpack_dir = staging_dir / "unpacked"
+ unpack_dir.mkdir()
+ with tarfile.open(archive_path, "r:*") as tar:
+ tar.extractall(unpack_dir, filter="data")
+ archive_path.unlink()
+
+ if etag:
+ (unpack_dir / self.archive_etag_marker).write_text(etag)
+
+ # Swap the freshly unpacked tree into place so a partially staged
+ # bundle is never observable at self.s3_dags_dir.
+ old_dir = self.s3_dags_dir.parent /
f".s3-archive-old-{uuid.uuid4().hex}"
Review Comment:
A failed replacement can leave the bundle missing and its backup abandoned.
To install the new bundle, the code:
- Moves the current bundle to a temporary backup location.
- Moves the newly unpacked bundle into the normal location.
- Deletes the backup.
If the second step fails, the current bundle has already been moved away,
and nothing restores it. The normal bundle location can therefore remain
missing if the fallback also fails.
The backup is also placed outside the temporary staging folder, so the
guaranteed cleanup does not remove it. Failed updates can leave abandoned
copies consuming disk space.
The suggestion puts the backup inside the staging folder and restores it if
replacement fails. Both changes matter: cleanup alone would discard the backup
without recovering the working bundle. Using a fixed backup name inside that
unique staging folder also makes uuid unnecessary.
---
Drafted by Codex, reviewed by me
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -126,6 +155,20 @@ def refresh(self) -> None:
raise AirflowException("Refreshing a specific version is not
supported")
with self.lock():
+ if self.archive_key:
+ try:
+ self._refresh_from_archive()
+ return
+ except Exception:
Review Comment:
If downloading or unpacking the archive fails, Airflow falls back to
downloading individual files from the configured S3 prefix - essentially an S3
folder.
But the PR allows a bucket to contain only the archive, with no individual
files in that folder. The fallback synchronizes this empty folder with the
local bundle and deletes anything that is no longer present in S3. That means
it can erase the entire working local bundle.
--
Drafted by Codex, reviewed and verified by me
##########
providers/amazon/docs/bundles/index.rst:
##########
@@ -45,3 +45,45 @@ Example of using the S3DagBundle:
}
}
]'
+
+Staging from a single archive object
+------------------------------------
+
+By default the bundle is staged by downloading every object under ``prefix``
one at a time. Each object
+costs a full request round-trip, so bundles made of many small files can take
a long time to stage — a cost
+paid by every component that stages the bundle, which with ephemeral workers
(e.g. KubernetesExecutor task
+pods) means on every task start.
+
+Setting the optional ``archive_key`` stages the bundle from a single
``.tar.gz`` object instead: one
+``HEAD`` request to detect changes (unchanged archives are not re-downloaded),
one ``GET`` to fetch it, then
+a local unpack and an atomic swap into place. If the archive cannot be fetched
or unpacked, staging
+automatically falls back to the per-object sync of ``prefix``.
+
+.. code-block:: bash
+
+ export AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST='[
+ {
+ "name": "my-s3-dags",
+ "classpath": "airflow.providers.amazon.aws.bundles.s3.S3DagBundle",
+ "kwargs": {
+ "aws_conn_id": "aws_default",
+ "bucket_name": "my-airflow-bucket",
+ "prefix": "dags/",
+ "archive_key": "bundle-archives/dags.tar.gz",
+ "refresh_interval": 60
+ }
+ }
+ ]'
+
+Publishing the archive is the responsibility of your deployment process. The
archive members must be laid out exactly as
+the objects under ``prefix``, so both staging strategies produce the same
local tree — for example, in the
+same CI job that syncs the Dags:
+
+.. code-block:: bash
+
+ tar -C ./dags -czf dags.tar.gz .
+ aws s3 cp dags.tar.gz s3://my-airflow-bucket/bundle-archives/dags.tar.gz
+
+.. note::
+ Keep the archive outside the ``prefix`` location, otherwise a ``aws s3
sync --delete`` of the Dag
Review Comment:
nit:
```suggestion
Keep the archive outside the ``prefix`` location, otherwise an ``aws s3
sync --delete`` of the Dag
```
--
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]
