调试 DAG 内存不足和存储空间不足问题

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

注意 ：此页面尚未针对 Cloud Composer 3 进行修改，显示的内容适用于 Cloud Composer 2。

本教程介绍了使用日志和环境监控功能在 Cloud Composer 中调试失败的 Airflow DAG 的步骤，以及诊断与工作器资源相关的问题（例如工作器内存或存储空间不足）。

简介

本教程重点介绍与资源相关的问题，以演示调试 DAG 的方法。

缺少分配的工作器资源会导致 DAG 失败。如果 Airflow 任务运行时内存或存储空间不足，您可能会看到 Airflow 异常，例如：

WARNING airflow.exceptions.AirflowException: Task received SIGTERM signal
INFO - Marking task as FAILED.

或

Task exited with return code Negsignal.SIGKILL

在这种情况下，一般建议增加 Airflow 工作器资源或减少每个工作器的任务数量。不过，由于 Airflow 异常可能很笼统，因此可能很难确定导致问题的具体资源。

本教程介绍了如何通过调试因工作器内存和存储空间不足而失败的两个示例 DAG，诊断 DAG 失败的原因并确定导致问题的资源类型。

目标

• 运行因以下原因而失败的示例 DAG：

  • 工作器内存不足
  • 工作器存储空间不足

• 诊断失败原因

• 增加分配的工作器资源

• 使用新的资源限制测试 DAG

费用

本教程使用 Google Cloud的以下可计费组件：

• Cloud Composer（请参阅额外费用）
• Cloud Monitoring

完成本教程后，您可以删除所创建的资源以避免继续计费。如需了解详情，请参阅清理。

准备工作

本部分介绍了在开始本教程之前需要执行的操作。

创建和配置项目

在本教程中，您需要一个 Google Cloud 项目。按如下方式配置项目：

1. 在 Google Cloud 控制台中，选择或创建项目：

   前往“项目选择器”

2. 确保您的项目已启用结算功能。 了解如何检查项目是否已启用结算功能。

3. 确保您的 Google Cloud 项目用户具有以下角色，以便创建必要的资源：

   • Environment and Storage Object Administrator (roles/composer.environmentAndStorageObjectAdmin)
   • Compute Admin (roles/compute.admin)
   • Monitoring Editor (roles/monitoring.editor)

为您的项目启用 API

Enable the Cloud Composer API.

Enable the API

创建 Cloud Composer 环境

创建 Cloud Composer 2 环境。

重要提示： 请务必使用“小型环境预设”，以便在使用本教程中稍后提供的示例时重现错误。这是通过 Google Cloud 控制台创建环境时的默认预设。

在创建环境的过程中，您需要向 Composer Service Agent 账号授予 Cloud Composer v2 API Service Agent Extension (roles/composer.ServiceAgentV2Ext) 角色。Cloud Composer 使用此账号在您的 Google Cloud 项目中执行操作。

查看工作器资源限制

查看您环境中的 Airflow 工作器资源限制：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 转到环境配置标签页。

4. 依次前往资源 > 工作负载配置 > Worker。

5. 检查值是否为 0.5 vCPU、1.875 GB 内存和 1 GB 存储空间。 这些是您将在本教程的后续步骤中使用 Airflow 工作器资源限制。

示例：诊断内存不足问题

将以下示例 DAG 上传到您在前面步骤中创建的环境。在本教程中，此 DAG 名为 create_list_with_many_strings。

此 DAG 包含一个任务，该任务会执行以下步骤：

1. 创建一个空列表 s。
2. 运行一个循环，将 More 字符串附加到列表。
3. 输出列表消耗的内存量，并在每 1 分钟的迭代中等待 1 秒。

警告 ：此 DAG 会模拟内存不足问题，这可能会干扰您环境中的操作。

import time

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import sys
from datetime import timedelta

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 0,
    'retry_delay': timedelta(minutes=10)
}

dag = DAG(
    'create_list_with_many_strings',
    default_args=default_args,
    schedule_interval=None)


def consume():
    s = []
    for i in range(120):
        for j in range(1000000):
            s.append("More")
        print(f"i={i}; size={sys.getsizeof(s) / (1000**3)}GB")
        time.sleep(1)


t1 = PythonOperator(
    task_id='task0',
    python_callable=consume,
    dag=dag,
    depends_on_past=False,
    retries=0
)

触发示例 DAG

触发示例 DAG create_list_with_many_strings：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在 Airflow Web 服务器列中，点击与您的环境对应的 Airflow 链接。

3. 在 Airflow 网页界面中的 DAG 页面上，在您的 DAG 的链接列中，点击触发 Dag (Trigger Dag) 按钮。

4. 点击触发器。

5. 在 DAG 页面上，点击您触发的任务，然后查看输出日志，确保 DAG 已开始运行。

任务运行时，输出日志会输出 DAG 正在使用的内存大小（以 GB 为单位）。

几分钟后，该任务将因超出 Airflow 工作器内存限制 (1.875 GB) 而失败。

注意： 如果您同时触发超出内存限制的 DAG 和任何其他示例 DAG，您会发现，如果其中一个任务超出内存限制并失败，其他正在运行的任务也会失败，即使它们未消耗内存也是如此。

诊断失败的 DAG

如果您在失败时运行了多个任务，不妨考虑只运行一个任务，并诊断相应时间段内的资源压力，以确定哪些任务会导致资源压力，以及您需要增加哪些资源。

查看 Airflow 任务日志

请注意，create_list_with_many_strings DAG 中的任务的状态为 Failed。

查看任务的日志。您将看到以下日志条目：

```none
{local_task_job.py:102} INFO - Task exited with return code
Negsignal.SIGKILL
```

`Netsignal.SIGKILL` might be an indication of your task using more memory
than the Airflow worker is allocated. The system sends
the `Negsignal.SIGKILL` signal to avoid further memory consumption.

查看工作负载

查看工作负载，检查任务的负载是否会导致运行 Pod 的节点超出内存用量限制：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 转到环境配置标签页。

4. 依次选择资源 > GKE 集群 > 工作负载，然后点击查看集群工作负载。

5. 检查某些工作负载 pod 的状态是否类似于以下状态：

   Error with exit code 137 and 1 more issue.
   ContainerStatusUnknown with exit code 137 and 1 more issue
   

   Exit code 137 表示容器或 Pod 尝试使用的内存超出允许的上限。系统会终止该进程以防止内存用量过高。

查看环境运行状况和资源消耗监控

查看环境健康状况和资源消耗监控：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 前往监控标签页，然后选择概览。

4. 在环境概览面板中，找到环境健康情况（Airflow 监控 DAG）图表。其中包含一个红色区域，对应于日志开始输出错误的时间。

5. 选择工作器，然后找到工作器内存总用量图表。 请注意，在任务运行时，内存用量行出现了峰值。

图 1. 工作器内存总用量图表（点击可放大）

即使图表中的内存用量线未达到上限，在诊断失败原因时，您也只需考虑可分配的内存，而图表中的内存上限线表示可用内存总量（包括 GKE 预留的容量）。

在此示例中，工作器内存限制设置为 1.875 GB。GKE 会预留前 4 GiB 内存的 25%。GKE 还会在每个节点上额外预留 100 MiB 内存以用于 kubelet 逐出，即 eviction-threshold。

可分配内存按以下方式计算：

ALLOCATABLE = CAPACITY - RESERVED - EVICTION-THRESHOLD

如果内存限制为 1.875 GB，则实际可分配的内存为：

1.75 GiB (1.875GB) - 0.44 (25% GiB reserved) - 0.1 = 1.21 GiB (~1.3 GB).

将此实际限制附加到内存用量图表后，您会发现任务的内存用量激增达到了实际内存限制，因此可以得出结论，任务因工作器内存不足而失败。

注意： 如果内存用量增长过快，内存用量线可能无法正确捕获实际资源用量。airflow_monitoring DAG 是对每个环境进行的活跃性健康检查，用于填充 environment/healthy 监控指标。借助此 DAG，您可以使用 Cloud Monitoring 功能（例如指标图表），或在环境出现运行状况不佳时设置提醒。该脚本每 5 分钟运行一次并收集指标，这意味着它可能无法捕获运行期间的资源用量。

提高工作器内存上限

注意 ：更改工作器的 CPU 或内存会重启工作器，这可能会影响正在运行的任务。我们建议在没有运行 DAG 时执行此操作。

分配额外的工作器内存，以便示例 DAG 成功运行：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 转到环境配置标签页。

4. 找到资源 > 工作负载配置，然后点击修改。

5. 在工作器部分的内存字段中，为 Airflow 工作器指定新的内存限制。在本教程中，请使用 3 GB。

6. 保存更改，并等待几分钟，让 Airflow 工作器重启。

使用新内存限制测试 DAG

再次触发 create_list_with_many_strings DAG，并等待其运行完毕。

1. 在 DAG 运行的输出日志中，您会看到 Marking task as SUCCESS，并且任务的状态将显示为“成功”。

2. 查看 Monitoring（监控）标签页中的 Environment overview（环境概览）部分，确保没有红色区域。

3. 点击工作器部分，然后找到工作器内存总用量图表。您会看到，内存上限线反映了内存上限的变化，而内存用量线远低于实际的可分配内存上限。

示例：诊断存储空间不足问题

在此步骤中，您将上传两个会创建大型文件的 DAG。第一个 DAG 会创建一个大文件。第二个 DAG 会创建一个大文件并模拟长时间运行的操作。

这两个 DAG 中的文件大小都超出了 Airflow 工作器的默认存储空间限制（1 GB），但第二个 DAG 包含一个额外的等待任务，用于人为延长其时长。

在后续步骤中，您将调查这两个 DAG 的行为差异。

注意： 本教程中的 DAG 会创建本地文件。在生产环境中，请避免在本地文件系统中存储任何文件或配置，因为下一个任务可能会在无法访问该文件的其他 Airflow 工作器上运行。

上传会创建大型文件的 DAG

将以下示例 DAG 上传到您在前面步骤中创建的环境。在本教程中，此 DAG 名为 create_large_txt_file_print_logs。

此 DAG 包含一个任务，该任务会执行以下步骤：

1. 将 1.5 GB 的 localfile.txt 文件写入 Airflow 工作器存储空间。
2. 使用 Python os 模块输出创建的文件的大小。
3. 每 1 分钟输出一次 DAG 运行时长。

警告： 此 DAG 会模拟存储空间不足问题，这可能会干扰您环境中的操作。

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 0,
    'retry_delay': timedelta(minutes=10)
}

dag = DAG(
    'create_large_txt_file_print_logs',
    default_args=default_args,
    schedule_interval=None)


def consume():
    size = 1000**2  # bytes in 1 MB
    amount = 100

    def create_file():
        print(f"Start creating a huge file")
        with open("localfile.txt", "ab") as f:
            for j in range(15):
                f.write(os.urandom(amount) * size)
        print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")

    create_file()
    print("Success!")


t1 = PythonOperator(
    task_id='create_huge_file',
    python_callable=consume,
    dag=dag,
    depends_on_past=False,
    retries=0)

上传在长时间运行的操作中创建大文件的 DAG

如需模拟长时间运行的 DAG 并调查任务时长对结束状态的影响，请将第二个示例 DAG 上传到您的环境。在本教程中，此 DAG 名为 long_running_create_large_txt_file_print_logs。

此 DAG 包含一个任务，该任务会执行以下步骤：

1. 将 1.5 GB 的 localfile.txt 文件写入 Airflow 工作器存储空间。
2. 使用 Python os 模块输出创建的文件的大小。
3. 等待 1 小时 15 分钟，以模拟执行文件操作（例如从文件中读取）所需的时间。
4. 每 1 分钟输出一次 DAG 运行时长。

警告： 此 DAG 会模拟存储空间不足问题，这可能会干扰您环境中的操作。

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
import os
from datetime import timedelta
import time

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 0,
    'retry_delay': timedelta(minutes=10)
}

dag = DAG(
    'long_running_create_large_txt_file_print_logs',
    default_args=default_args,
    schedule_interval=None)


def consume():
    size = 1000**2  # bytes in 1 MB
    amount = 100

    def create_file():
        print(f"Start creating a huge file")
        with open("localfile.txt", "ab") as f:
            for j in range(15):
                f.write(os.urandom(amount) * size)
        print("localfile.txt size:", os.stat("localfile.txt").st_size / (1000**3), "GB")

    create_file()
    for k in range(75):
        time.sleep(60)
        print(f"{k+1} minute")

    print("Success!")


t1 = PythonOperator(
    task_id='create_huge_file',
    python_callable=consume,
    dag=dag,
    depends_on_past=False,
    retries=0)

触发示例 DAG

触发第一个 DAG create_large_txt_file_print_logs：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在 Airflow Web 服务器列中，点击与您的环境对应的 Airflow 链接。

3. 在 Airflow 网页界面中的 DAG 页面上，在您的 DAG 的链接列中，点击触发 Dag (Trigger Dag) 按钮。

4. 点击触发器。

5. 在 DAG 页面上，点击您触发的任务，然后查看输出日志，确保 DAG 已开始运行。

6. 等待您使用 create_large_txt_file_print_logs DAG 创建的任务完成。这可能需要几分钟的时间。

   注意 ：您可能会注意到，在任务完成后，已使用的 Airflow 工作器存储空间已重置为零。Cloud Composer 依赖于 Google Kubernetes Engine Autopilot，后者会提供必要的资源。Cloud Composer 中的自动扩缩基于指标，并通过 GKE Pod 管理在给定配额内控制资源分配。

7. 在 DAG 页面上，点击 DAG 运行作业。您会看到任务的状态为 Success，即使存储空间用量已超出上限也是如此。

查看任务 Airflow 日志：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

   1. 在环境列表中，点击您的环境名称。环境详情页面会打开。

   2. 前往日志标签页，然后依次选择所有日志 > Airflow 日志 > 工作器 > 在日志浏览器中查看。

   3. 按类型过滤日志：仅显示错误消息。

在日志中，您会看到类似如下内容的消息：

Worker: warm shutdown (Main Process)

或

A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.

这些日志表明，由于所用存储空间超出限制并在 1 小时内被驱逐，因此 pod 启动了“温行关闭”流程。不过，DAG 运行并未失败，因为它在 Kubernetes 终止宽限期内完成，本教程将对此进行进一步说明。

为了说明终止宽限期的概念，请查看第二个示例 DAG long_running_create_large_txt_file_print_logs 的结果。

触发第二个 DAG long_running_create_large_txt_file_print_logs：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在 Airflow Web 服务器列中，点击与您的环境对应的 Airflow 链接。

3. 在 Airflow 网页界面中的 DAG 页面上，在您的 DAG 的链接列中，点击触发 Dag (Trigger Dag) 按钮。

4. 点击触发器。

5. 在 DAG 页面上，点击您触发的任务，然后查看输出日志，确保 DAG 已开始运行。

6. 等待 long_running_create_large_txt_file_print_logs DAG 运行失败。此过程大约需要 1 小时。

查看 DAG 运行结果：

1. 在 DAG 页面上，点击 long_running_create_large_txt_file_print_logs DAG 运行作业。您会看到该任务的状态为 Failed，并且运行时长恰好为 1 小时 5 分钟，这比任务的等待时长（1 小时 15 分钟）短。

2. 查看任务的日志。当 DAG 在 Airflow 工作器的容器中创建 localfile.txt 文件后，日志会输出 DAG 开始等待的消息，并且每 1 分钟在任务日志中输出运行时长。在此示例中，DAG 会输出 localfile.txt size: 日志，并且 localfile.txt 文件的大小将为 1.5 GB。

写入 Airflow 工作器容器的文件超出存储限制后，DAG 运行应会失败。不过，该任务不会立即失败，而是会一直运行，直到其时长达到 1 小时 5 分钟。之所以会出现这种情况，是因为 Kubernetes 不会立即终止任务，而是会让其继续运行，以便在 1 小时内恢复，这段时间称为“终止宽限期”。节点耗尽资源后，Kubernetes 不会立即终止 Pod，而是会妥善处理终止，以尽可能减少对最终用户的影响。

终止宽限期有助于用户在任务失败后恢复文件，但在诊断 DAG 时可能会造成混淆。当 Airflow 工作器存储空间超出上限时，结束任务状态取决于 DAG 运行时长：

• 如果 DAG 运行超出工作器存储空间上限，但在 1 小时内完成，则任务会在终止宽限期内完成，并显示 Success 状态。不过，Kubernetes 会终止 Pod，并且写入的文件会立即从容器中删除。

• 如果 DAG 超出工作器存储空间上限且运行时间超过 1 小时，则 DAG 会继续运行 1 小时，并且在 Kubernetes 移除 Pod 并 Airflow 将任务标记为 Failed 之前，其存储空间用量可能会超出存储空间上限数千倍。

注意 ：如果您同时触发超出存储空间限制的某个示例 DAG 和任何其他示例 DAG，您会发现，如果其中一个任务超出存储空间限制并失败，任何其他正在运行的任务也会失败，即使它们不需要存储空间也是如此。

诊断失败的 DAG

如果您在失败时运行了多个任务，不妨考虑只运行一个任务，并诊断相应时间段内的资源压力，以确定哪些任务会导致资源压力，以及您需要增加哪些资源。

查看第二个 DAG long_running_create_large_txt_file_print_logs 的任务日志：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 前往日志标签页，然后依次选择所有日志 > Airflow 日志 > 工作器 > 在日志浏览器中查看。

4. 按类型过滤日志：仅显示错误消息。

在日志中，您会看到类似如下内容的消息：

Container storage usage of worker reached 155.7% of the limit.

This likely means that the total size of local files generated by your DAGs is
close to the storage limit of worker.

You may need to decrease the storage usage or increase the worker storage limit
in your Cloud Composer environment configuration.

或

Pod storage usage of worker reached 140.2% of the limit.
A worker pod was evicted at 2023-12-01T12:30:05Z with message: Pod ephemeral
local storage usage exceeds the total limit of containers 1023Mi.

This eviction likely means that the total size of dags and plugins folders plus
local files generated by your DAGs exceeds the storage limit of worker.

Please decrease the storage usage or increase the worker storage limit in your
Cloud Composer environment configuration.

这些消息表示，随着任务的进行，当 DAG 生成的文件大小超出工作器存储空间限制并开始进入终止宽限期时，Airflow 日志开始输出错误。在终止宽限期内，存储空间用量并未恢复到上限，这导致在终止宽限期结束后 pod 被驱逐。

查看环境健康状况和资源消耗监控：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 前往监控标签页，然后选择概览。

4. 在环境概览面板中，找到环境健康情况（Airflow 监控 DAG）图表。其中包含一个红色区域，对应于日志开始输出错误的时间。

5. 选择工作器，然后找到工作器磁盘总用量图表。请注意，在任务运行时，磁盘用量线条出现了峰值，并超出了磁盘限制线条。

图 2. 工作器磁盘总用量图表（点击可放大）

提高 Worker 存储空间上限

注意 ：更改工作器存储空间会重启 Airflow 工作器，这可能会影响正在运行的任务。我们建议在没有运行 DAG 时执行此操作。

分配额外的 Airflow 工作器存储空间，以便示例 DAG 成功运行：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击您的环境名称。环境详情页面会打开。

3. 转到环境配置标签页。

4. 找到资源 > 工作负载配置，然后点击修改。

5. 在工作器部分的存储空间字段中，为 Airflow 工作器指定新的存储空间限制。在本教程中，将其设置为 2 GB。

6. 保存更改，并等待几分钟，让 Airflow 工作器重启。

使用新的存储空间上限测试 DAG

再次触发 long_running_create_large_txt_file_print_logs DAG，然后等待 1 小时 15 分钟，直到其运行完毕。

1. 在 DAG 运行的输出日志中，您会看到 Marking task as SUCCESS，并且任务的状态将显示为“成功”，时长为 1 小时 15 分钟，即等待时间（在 DAG 代码中设置）。

2. 查看 Monitoring（监控）标签页中的 Environment overview（环境概览）部分，确保没有红色区域。

3. 点击工作器部分，然后找到工作器磁盘总用量图表。您会看到，“磁盘上限”行反映了存储空间上限的变化，而“磁盘用量”行则在允许范围内。

摘要

在本教程中，您通过调试因工作器内存和存储空间不足而失败的两个示例 DAG，诊断了 DAG 失败的原因，并确定了造成压力的资源类型。然后，您向工作器分配了更多内存和存储空间，并成功运行了 DAG。不过，建议您先优化 DAG（工作流），以减少工作器资源消耗，因为资源无法超出特定阈值。

清理

为避免因本教程中使用的资源导致您的 Google Cloud 账号产生费用，请删除包含这些资源的项目，或者保留项目但删除各个资源。

删除项目

注意：删除项目会产生以下影响：

• 项目中的所有内容都会被删除。如果您将现有项目用于本文档中的任务，则删除该项目后，还将删除您已在该项目中完成的任何其他工作。
• 自定义项目 ID 丢失。创建此项目时，您可能创建了要在将来使用的自定义项目 ID。要保留使用该项目 ID 的网址（例如 appspot.com 网址），请删除项目内的所选资源，而不是删除整个项目。

如果您打算探索多个架构、教程或快速入门，则重复使用项目可以帮助您避免超出项目配额限制。

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

删除各个资源

如果您打算探索多个教程和快速入门，重复使用项目可以帮助您避免超出项目配额上限。

删除 Cloud Composer 环境。在此过程中，您还会删除环境的存储分区。

后续步骤

• 扩缩环境
• DAG 问题排查