This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ff5a90c6f79 Support setting an S3 object ACL for remote task logs
(#70394)
ff5a90c6f79 is described below
commit ff5a90c6f79ec2ff141085c938150d4b4668e188
Author: PoAn Yang <[email protected]>
AuthorDate: Sun Aug 2 01:34:31 2026 +0900
Support setting an S3 object ACL for remote task logs (#70394)
* Support setting an S3 object ACL for remote task logs
Signed-off-by: PoAn Yang <[email protected]>
* Record the S3 log ACL option against the release that ships it
Amazon provider 9.33.0 is already tagged and released, so it cannot
contain this option.
---------
Signed-off-by: PoAn Yang <[email protected]>
Co-authored-by: Jarek Potiuk <[email protected]>
---
providers/amazon/docs/logging/s3-task-handler.rst | 20 ++++++++
providers/amazon/provider.yaml | 15 ++++++
.../providers/amazon/aws/log/s3_task_handler.py | 7 +++
.../airflow/providers/amazon/get_provider_info.py | 7 +++
.../unit/amazon/aws/log/test_s3_task_handler.py | 53 ++++++++++++++++++++++
5 files changed, 102 insertions(+)
diff --git a/providers/amazon/docs/logging/s3-task-handler.rst
b/providers/amazon/docs/logging/s3-task-handler.rst
index 577a1d3f38f..9870f4ea554 100644
--- a/providers/amazon/docs/logging/s3-task-handler.rst
+++ b/providers/amazon/docs/logging/s3-task-handler.rst
@@ -43,6 +43,26 @@ To enable this feature, ``airflow.cfg`` must be configured
as follows:
In the above example, Airflow will try to use
``S3Hook(aws_conn_id='my_s3_conn')``.
+Setting an object ACL for cross-account logging
+'''''''''''''''''''''''''''''''''''''''''''''''
+
+When Airflow runs under one AWS account but writes logs to a bucket owned by a
different
+account, S3 makes the writing account the owner of each uploaded log object.
As a result the
+bucket owner cannot read or manage the logs. Applying the
``bucket-owner-full-control`` ACL
+on upload grants the bucket owner full control over the log objects.
+
+Set the ACL with the ``[aws] s3_task_handler_acl_policy`` option:
+
+.. code-block:: ini
+
+ [aws]
+ # ACL applied to every task log object uploaded to S3, e.g. for
cross-account buckets.
+ s3_task_handler_acl_policy = bucket-owner-full-control
+
+When unset, no ACL is sent and the bucket's default object ownership applies.
The same value can
+also be supplied through ``[logging] remote_task_handler_kwargs``, for example
+``{"acl_policy": "bucket-owner-full-control"}``.
+
You can also use `LocalStack <https://localstack.cloud/>`_ to emulate Amazon
S3 locally.
To configure it, you must additionally set the endpoint url to point to your
local stack.
You can do this via the Connection Extra ``endpoint_url`` field.
diff --git a/providers/amazon/provider.yaml b/providers/amazon/provider.yaml
index 343398db658..fecd3a9b236 100644
--- a/providers/amazon/provider.yaml
+++ b/providers/amazon/provider.yaml
@@ -1169,6 +1169,21 @@ config:
version_added: 8.7.2
example:
airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize
default:
airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize_legacy
+ s3_task_handler_acl_policy:
+ description: |
+ The ACL applied to task log objects uploaded to S3 by the S3 remote
log handler,
+ for example ``bucket-owner-full-control``.
+
+ This is primarily useful for cross-account remote logging: when
Airflow runs under one AWS
+ account but writes logs to a bucket owned by another account, S3
makes the writing account
+ the object owner, so the bucket owner cannot read or manage the log
objects. Setting
+ ``bucket-owner-full-control`` grants the bucket owner full control
over the uploaded logs.
+
+ When unset, no ACL is sent and the bucket's default object ownership
applies.
+ type: string
+ version_added: 9.34.0
+ example: bucket-owner-full-control
+ default: ~
aws_batch_executor:
description: |
This section only applies if you are using the AwsBatchExecutor in
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
b/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
index 6ef82f301da..68a88b679eb 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/log/s3_task_handler.py
@@ -44,6 +44,7 @@ class S3RemoteLogIO(LoggingMixin): # noqa: D101
remote_base: str
base_log_folder: pathlib.Path = attrs.field(converter=pathlib.Path)
delete_local_copy: bool
+ acl_policy: str | None = None
processors = ()
@@ -68,6 +69,7 @@ class S3RemoteLogIO(LoggingMixin): # noqa: D101
"base_log_folder":
os.path.expanduser(conf.get_mandatory_value("logging", "base_log_folder")),
"remote_base": conf.get_mandatory_value("logging",
"remote_base_log_folder"),
"delete_local_copy": conf.getboolean("logging",
"delete_local_logs"),
+ "acl_policy": conf.get("aws", "s3_task_handler_acl_policy",
fallback=None) or None,
}
| io_kwargs,
)
@@ -161,6 +163,8 @@ class S3RemoteLogIO(LoggingMixin): # noqa: D101
extra_args = {}
if conf.getboolean("logging", "ENCRYPT_S3_LOGS"):
extra_args["ServerSideEncryption"] = "AES256"
+ if self.acl_policy:
+ extra_args["ACL"] = self.acl_policy
# Default to a single retry attempt because s3 upload failures are
# rare but occasionally occur. Multiple retry attempts are unlikely
@@ -235,6 +239,9 @@ class S3TaskHandler(FileTaskHandler, LoggingMixin):
delete_local_copy=kwargs.get(
"delete_local_copy", conf.getboolean("logging",
"delete_local_logs")
),
+ acl_policy=kwargs.get(
+ "acl_policy", conf.get("aws", "s3_task_handler_acl_policy",
fallback=None) or None
+ ),
)
def set_context(self, ti: TaskInstance, *, identifier: str | None = None)
-> None:
diff --git a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
index a3e609aa5a8..8e249e606fe 100644
--- a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
+++ b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
@@ -1283,6 +1283,13 @@ def get_provider_info():
"example":
"airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize",
"default":
"airflow.providers.amazon.aws.log.cloudwatch_task_handler.json_serialize_legacy",
},
+ "s3_task_handler_acl_policy": {
+ "description": "The ACL applied to task log objects
uploaded to S3 by the S3 remote log handler,\nfor example
``bucket-owner-full-control``.\n\nThis is primarily useful for cross-account
remote logging: when Airflow runs under one AWS\naccount but writes logs to a
bucket owned by another account, S3 makes the writing account\nthe object
owner, so the bucket owner cannot read or manage the log objects.
Setting\n``bucket-owner-full-control`` grants the bucket owner [...]
+ "type": "string",
+ "version_added": "9.34.0",
+ "example": "bucket-owner-full-control",
+ "default": None,
+ },
},
},
"aws_batch_executor": {
diff --git a/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
b/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
index 0576756c196..5fcbedf8dda 100644
--- a/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
+++ b/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
@@ -132,6 +132,50 @@ class TestS3RemoteLogIOFromConfig:
assert conn_id == "aws_default"
legacy_discover.assert_not_called()
+ @conf_vars(
+ {
+ ("logging", "remote_base_log_folder"):
"s3://bucket/remote/log/location",
+ ("aws", "s3_task_handler_acl_policy"): "bucket-owner-full-control",
+ }
+ )
+ def test_from_config_reads_acl_policy(self):
+ subject = S3RemoteLogIO.from_config()
+
+ assert subject.acl_policy == "bucket-owner-full-control"
+
+ @conf_vars({("logging", "remote_base_log_folder"):
"s3://bucket/remote/log/location"})
+ def test_from_config_acl_policy_defaults_to_none(self):
+ subject = S3RemoteLogIO.from_config()
+
+ assert subject.acl_policy is None
+
+ @conf_vars(
+ {
+ ("logging", "remote_base_log_folder"):
"s3://bucket/remote/log/location",
+ ("logging", "remote_task_handler_kwargs"): '{"acl_policy":
"bucket-owner-full-control"}',
+ }
+ )
+ def test_from_config_acl_policy_via_remote_task_handler_kwargs(self):
+ subject = S3RemoteLogIO.from_config()
+
+ assert subject.acl_policy == "bucket-owner-full-control"
+
+
+class TestS3TaskHandlerInit:
+ @conf_vars({("aws", "s3_task_handler_acl_policy"):
"bucket-owner-full-control"})
+ def test_init_reads_acl_policy_from_conf(self):
+ handler = S3TaskHandler("/tmp/local",
"s3://bucket/remote/log/location")
+
+ assert handler.io.acl_policy == "bucket-owner-full-control"
+
+ @conf_vars({("aws", "s3_task_handler_acl_policy"):
"bucket-owner-full-control"})
+ def test_init_acl_policy_kwarg_overrides_conf(self):
+ handler = S3TaskHandler(
+ "/tmp/local", "s3://bucket/remote/log/location",
acl_policy="bucket-owner-read"
+ )
+
+ assert handler.io.acl_policy == "bucket-owner-read"
+
@pytest.mark.db_test
class TestS3RemoteLogIO:
@@ -280,6 +324,15 @@ class TestS3RemoteLogIO:
body = boto3.resource("s3").Object("bucket",
self.remote_log_key).get()["Body"].read()
assert body == b"text"
+ def test_write_with_acl_policy(self):
+ self.subject.acl_policy = "bucket-owner-full-control"
+ conn = self.subject.hook.get_conn()
+ with mock.patch.object(conn, "put_object", wraps=conn.put_object) as
mock_put_object:
+ self.subject.write("text", self.remote_log_location)
+ assert mock_put_object.call_args.kwargs["ACL"] ==
"bucket-owner-full-control"
+ body = boto3.resource("s3").Object("bucket",
self.remote_log_key).get()["Body"].read()
+ assert body == b"text"
+
def test_upload_repeated_appends_no_duplication(self):
"""Simulate reschedule-mode sensor: each cycle appends to the local
log, then uploads.
