From fa28b95cad539590c850038dba4cee892251906c Mon Sep 17 00:00:00 2001 From: PoAn Yang Date: Fri, 24 Jul 2026 23:43:16 +0900 Subject: [PATCH] Support setting an S3 object ACL for remote task logs Signed-off-by: PoAn Yang --- .../amazon/docs/logging/s3-task-handler.rst | 20 +++++++ providers/amazon/provider.yaml | 15 ++++++ .../amazon/aws/log/s3_task_handler.py | 7 +++ .../providers/amazon/get_provider_info.py | 7 +++ .../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 577a1d3f38fcf..9870f4ea554d4 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 `_ 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 0121c5b5ee2ad..7de1f43a8c22e 100644 --- a/providers/amazon/provider.yaml +++ b/providers/amazon/provider.yaml @@ -1166,6 +1166,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.33.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 6ef82f301da4b..68a88b679eb33 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 @@ def from_config(cls) -> S3RemoteLogIO: "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 @@ def write( 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 @@ def __init__( 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 5314709a5df68..a5f2c447b8e1d 100644 --- a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py +++ b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py @@ -1279,6 +1279,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 full control over the uploaded logs.\n\nWhen unset, no ACL is sent and the bucket's default object ownership applies.\n", + "type": "string", + "version_added": "9.33.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 0576756c1965f..5fcbedf8dda0d 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 @@ def test_resolve_remote_task_log_uses_provider_dispatch_not_local_settings(self, 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 @@ def test_write_with_encryption(self): 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.