ephraimbuddy commented on code in PR #73015:
URL: https://github.com/apache/airflow/pull/73015#discussion_r4405071497
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -140,6 +170,60 @@ 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))
Review Comment:
The marker stores the ETag from the HEAD above, but `download_file` fetches
whatever is current at GET time, so if the archive is replaced in between, the
marker can describe a different object than the one unpacked. Streaming
`get_object(..., IfMatch=etag)` to the file closes the gap (`download_file`
doesn't accept `IfMatch`). It also drops the HEAD that `download_file` sends by
default, so a changed archive costs 2 HEAD + 1 GET today, not the "one HEAD,
one GET" in the docs.
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -140,6 +170,60 @@ 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")
Review Comment:
I don't think `filter` is available for python 3.11.0-3
##########
providers/amazon/src/airflow/providers/amazon/aws/bundles/s3.py:
##########
@@ -140,6 +170,60 @@ 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-"))
Review Comment:
If the process is killed mid-staging (OOM, SIGKILL), the `finally` never
runs and a full copy of the Dags stays in the shared bundle storage root. Can
we put the bundle name in the prefix (e.g. `f".{self.name}-archive-staging-"`)
and remove leftovers under `self.lock()` before creating a new one? The lock is
per bundle, so without the name a cleanup could hit another bundle's
in-progress staging.
--
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]
