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

Question

为了尝试将气流记录设置到 localstack s3 存储桶，对于本地和 kubernetes 开发环境，我正在关注 airflow documentation for logging to s3。提供一些背景信息，localstack 是一个本地 AWS 云堆栈，其中包含在本地运行的包括 s3 在内的 AWS 服务。

我将以下环境变量添加到我的气流容器similar to this other stack overflow post 以尝试登录到我的本地 s3 存储桶。这是我为所有气流容器添加到 docker-compose.yaml 的内容：

       - 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 下的气流控制台后，如图所示： 这是新的异常，返回 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 凭证）访问存储桶，但它需要作为气流的一部分工作。

是否有适当的方法通过将MyS3Conn 添加到airflow.cfg 来为S3Hook 设置主机/端口/凭据？

根据airflow s3 hooks source code，似乎气流可能还不支持自定义 s3 URL。但是，根据气流aws_hook source code（父级），似乎应该可以设置endpoint_url，包括端口，并且应该从airflow.cfg读取。

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

可能还需要哪些其他步骤才能从 docker 中运行的气流中记录到 localstack s3 存储桶，并使用自动设置，甚至还支持吗？

Answer 1

我认为您应该使用localhost 而不是localstack 作为端点，例如host = http://localhost:4572.

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

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

我相信这会解决它？

Answer 2

我通过参考this guide 设法完成了这项工作。基本上，您需要使用 Connection 类创建一个连接并传递您需要的凭据，在我的情况下，我需要 AWS_SESSION_TOKEN, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, REGION_NAME 来完成这项工作。将此函数用作 PythonOperator 中的 python_callable，这应该是 DAG 的第一部分。

import os
import json
from airflow.models.connection import Connection
from airflow.exceptions import AirflowFailException

def _create_connection(**context):
    """
    Sets the connection information about the environment using the Connection
    class instead of doing it manually in the Airflow UI
    """
    AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
    AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
    AWS_SESSION_TOKEN = os.getenv("AWS_SESSION_TOKEN")
    REGION_NAME = os.getenv("REGION_NAME")
    credentials = [
        AWS_SESSION_TOKEN,
        AWS_ACCESS_KEY_ID,
        AWS_SECRET_ACCESS_KEY,
        REGION_NAME,
    ]
    if not credentials or any(not credential for credential in credentials):
        raise AirflowFailException("Environment variables were not passed")

    extras = json.dumps(
        dict(
            aws_session_token=AWS_SESSION_TOKEN,
            aws_access_key_id=AWS_ACCESS_KEY_ID,
            aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
            region_name=REGION_NAME,
        ),
    )
    try:
        Connection(
            conn_id="s3_con",
            conn_type="S3",
            extra=extras,
        )
    except Exception as e:
        raise AirflowFailException(
            f"Error creating connection to Airflow :{e!r}",
        )