amazon-s3 - 如何使用 localstack s3 端点以编程方式设置 Airflow 1.10 日志记录？

Question

为了将 Airflow 日志设置到 localstack s3 存储桶，对于本地和 kubernetes 开发环境，我正在关注 airflow documentation for logging to s3 .提供一点背景信息，localstack是一个本地 AWS 云堆栈，具有包括本地运行的 s3 在内的 AWS 服务。

我在 Airflow 容器中添加了以下环境变量 similar to this other stack overflow post试图登录到我的本地 s3 存储桶。这是我添加到 docker-compose.yaml 的内容对于所有 Airflow 容器:

       - AIRFLOW__CORE__REMOTE_LOGGING=True
       - AIRFLOW__CORE__REMOTE_BASE_LOG_FOLDER=s3://local-airflow-logs
       - AIRFLOW__CORE__REMOTE_LOG_CONN_ID=MyS3Conn
       - AIRFLOW__CORE__ENCRYPT_S3_LOGS=False

我还将我的 localstack s3 凭证添加到 airflow.cfg

[MyS3Conn]
aws_access_key_id = foo
aws_secret_access_key = bar
aws_default_region = us-east-1
host = http://localstack:4572    # s3 port. not sure if this is right place for it 

此外，我已经安装了 apache-airflow[hooks] 和 apache-airflow[s3]，但根据 documentation 不清楚真正需要哪一个。 .

我已按照 a previous stack overflow post 中的步骤操作尝试验证 S3Hook 是否可以写入我的 localstack s3 实例:

from airflow.hooks import S3Hook
s3 = S3Hook(aws_conn_id='MyS3Conn')
s3.load_string('test','test',bucket_name='local-airflow-logs')

但我得到 botocore.exceptions.NoCredentialsError: Unable to locate credentials .

在 /admin/connection/edit 下向 Airflow 控制台添加凭据后如图:

这是新的异常(exception)， botocore.exceptions.ClientError: An error occurred (InvalidAccessKeyId) when calling the PutObject operation: The AWS Access Key Id you provided does not exist in our records.被退回。 Other people have encountered this same issue它可能与网络有关。

无论如何，需要编程设置，而不是手动设置。

我能够使用独立的 Python 脚本(使用 boto 显式输入 AWS 凭证)访问存储桶，但它需要作为 Airflow 的一部分工作。

是否有正确的方法为 S3Hook 设置主机/端口/凭据？通过添加 MyS3Conn至 airflow.cfg ?

基于 airflow s3 hooks source code ，似乎 Airflow 可能尚不支持自定义 s3 URL。但是，基于 Airflow aws_hook source code (父)似乎应该可以设置endpoint_url 包括端口，它应该从 airflow.cfg 读取。 .

我可以单独使用 boto 在 localstack 中检查和写入我的 s3 存储桶。另外， curl http://localstack:4572/local-mochi-airflow-logs从 Airflow 容器返回桶的内容。和 aws --endpoint-url=http://localhost:4572 s3 ls返回 Could not connect to the endpoint URL: "http://localhost:4572/" .

从在 docker 中运行的 Airflow 登录到 localstack s3 存储桶可能还需要哪些其他步骤，使用自动设置，这是否甚至受支持？

Answer 1

我认为你应该使用 localhost不是 localstack对于端点，例如host = http://localhost:4572 .

在 Airflow 1.10 中，您可以在每个连接的基础上覆盖端点，但不幸的是，它一次仅支持一个端点，因此您需要为使用该连接的所有 AWS Hook 更改它。要覆盖它，请编辑相关连接并在“额外”字段中输入:

{"host": "http://localhost:4572"}

我相信这会解决它吗？