将日志写入 Amazon S3¶

远程日志写入 Amazon S3 使用现有的 Airflow 连接来读取或写入日志。如果您没有正确设置连接，此过程将失败。

启用远程日志记录¶

要启用此功能，必须按如下方式配置 airflow.cfg

[logging]
# Airflow can store logs remotely in AWS S3. Users must supply a remote
# location URL (starting with either 's3://...') and an Airflow connection
# id that provides access to the storage location.
remote_logging = True
remote_base_log_folder = s3://my-bucket/path/to/logs
remote_log_conn_id = my_s3_conn
# Use server-side encryption for logs stored in S3
encrypt_s3_logs = False

在上面的示例中，Airflow 将尝试使用 S3Hook(aws_conn_id='my_s3_conn')。

您还可以使用 LocalStack 在本地模拟 Amazon S3。要配置它，您必须额外设置 endpoint url 以指向您的本地堆栈。您可以通过连接额外的 endpoint_url 字段来完成此操作。例如，{"endpoint_url": "http://localstack:4572"}

使用 AWS IRSA 启用 Amazon S3 的远程日志记录¶

IRSA 是一项功能，允许您将 IAM 角色分配给 Kubernetes 服务帐户。它的工作原理是利用 Kubernetes 功能，称为服务帐户令牌卷投影。当 Pod 配置了引用 IAM 角色的服务帐户时，Kubernetes API 服务器将在启动时调用集群的公共 OIDC 发现端点。当调用 AWS API 时，AWS SDK 会调用 sts:AssumeRoleWithWebIdentity。在验证令牌签名后，IAM 会将 Kubernetes 颁发的令牌交换为临时的 AWS 角色凭证。

为了从 Amazon EKS 访问 AWS 服务（例如，S3），建议使用 ServiceAccount 的 IAM 角色。在以下步骤中，您将学习如何创建具有 ServiceAccount 的新 IAM 角色，并将其与 Airflow WebServers 和 Worker（Kubernetes 执行器）一起使用。

步骤 1：为服务帐户创建 IAM 角色 (IRSA)¶

此步骤使用 eksctl 创建 IAM 角色和服务帐户。另外，请注意，此示例使用附加到 IAM 角色的具有完整 S3 权限的托管策略。这仅用于测试目的。我们强烈建议您创建一个受限的 S3 IAM 策略，并将其与 --attach-policy-arn 一起使用。

或者，您可以使用其他 IaC 工具，如 Terraform。有关使用 Terraform 部署 Airflow（包括 IRSA）的信息。请查看此示例链接。

通过提供所有必要的输入来执行以下命令。

eksctl create iamserviceaccount --cluster="<EKS_CLUSTER_ID>" --name="<SERVICE_ACCOUNT_NAME>" --namespace="<NAMESPACE>" --attach-policy-arn="<IAM_POLICY_ARN>" --approve``

带有示例输入的示例

eksctl create iamserviceaccount --cluster=airflow-eks-cluster --name=airflow-sa --namespace=airflow --attach-policy-arn=arn:aws:iam::aws:policy/AmazonS3FullAccess --approve

如果您创建自己的 IAM 策略（强烈建议这样做），它应包括以下权限。

s3:ListBucket（用于写入日志的 S3 存储桶）
s3:GetObject（用于写入日志的前缀下的所有对象）
s3:PutObject（用于写入日志的前缀下的所有对象）

步骤 2：使用服务帐户更新 Helm Chart values.yaml¶

此步骤使用 Airflow Helm Chart 部署。如果您使用 Helm Chart 部署 Airflow，则可以按如下所述修改 values.yaml。将步骤 1 创建的服务帐户（例如，airflow-sa）添加到 Helm Chart values.yaml 中的以下部分。我们正在使用现有的 serviceAccount，因此 create: false，并且现有名称为 name: airflow-sa。

workers:
  serviceAccount:
    create: false
    name: airflow-sa
    # Annotations are automatically added by **Step1** to serviceAccount. So, you dont need to mention the annotations. We have added this for information purpose
    annotations:
      eks.amazonaws.com/role-arn: <ENTER_IAM_ROLE_ARN_CREATED_BY_EKSCTL_COMMAND>

webserver:
  serviceAccount:
    create: false
    name: airflow-sa
    # Annotations are automatically added by **Step1** to serviceAccount. So, you dont need to mention the annotations. We have added this for information purpose
    annotations:
      eks.amazonaws.com/role-arn: <ENTER_IAM_ROLE_ARN_CREATED_BY_EKSCTL_COMMAND

config:
  logging:
    remote_logging: 'True'
    logging_level: 'INFO'
    remote_base_log_folder: 's3://<ENTER_YOUR_BUCKET_NAME>/<FOLDER_PATH' # Specify the S3 bucket used for logging
    remote_log_conn_id: 'aws_conn' # Notice that this name is used in Step3 for creating connections through Airflow UI
    delete_worker_pods: 'False'
    encrypt_s3_logs: 'True'

步骤 3：创建 Amazon Web Services 连接¶

通过上述配置，Webserver 和 Worker Pod 可以访问 Amazon S3 存储桶并写入日志，而无需使用任何访问密钥和秘密密钥或实例配置文件凭据。

使用 Airflow Web UI

在执行 DAG 之前，在 Airflow UI 下创建连接的最后一步。
- 使用 admin 凭据登录 Airflow Web UI，然后导航到 Admin -> Connections
- 为 Amazon Web Services 创建连接，然后选择图像中所示的选项（连接 ID 和连接类型）。
- 在 Extra 文本框中选择创建 S3 存储桶的正确区域。
使用 Airflow CLI

airflow connections add aws_conn --conn-uri aws://@/?region_name=eu-west-1

请注意，在 -conn-uri 参数中使用的 @ 通常分隔密码和主机，但在本例中，它符合使用的 uri 验证器。

步骤 4：验证日志¶

执行示例 DAG
验证 S3 存储桶中的日志
从 DAG 日志中验证 Airflow UI 中的日志