使用监控信息中心内的关键指标监控环境健康状况和性能

Managed Airflow（第 3 代） | Managed Airflow（第 2 代） | Managed Airflow（旧版第 1 代）

本页介绍了如何使用监控信息中心内的关键指标来监控 Managed Airflow 环境的总体健康状况和性能。

简介

本教程重点介绍关键的 Managed Airflow 监控指标，这些指标可提供环境级健康状况和性能的良好概览。

Managed Airflow 提供了多种用于描述环境总体状态的指标。本教程中的监控指南基于 Managed Airflow 环境的监控信息中心上显示的指标。

在本教程中，您将了解一些关键指标，这些指标可作为环境性能和健康状况问题的主要指示器，同时您还将了解如何根据每个指标制定纠正措施，以保持环境健康。您还将为每个指标设置提醒规则，运行示例 DAG，并使用这些指标和提醒来优化环境的性能。

目标

目标：

• 了解“监控”信息中心内提供的关键性能和健康状况指标。
• 了解各项指标的基准。
• 了解如何根据指标采取纠正措施。
• 为每个关键指标创建提醒。
• 运行示例 DAG。
• 将提醒和 Monitoring 图表解读为纠正措施。
• 优化您的环境并评估其性能。

费用

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

• Managed Airflow（请参阅额外费用）
• 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

启用 Managed Airflow API（如果尚未启用）。

启用 API 所需的角色

如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您创建了项目，则可能已经通过 Owner 角色 (roles/owner) 拥有此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获取此权限。 了解如何授予角色。

启用 API

创建 Managed Airflow 环境

创建 Managed Airflow（第 2 代）环境。

在此过程的最后一步，您将 Cloud Composer v2 API Service Agent Extension (roles/composer.ServiceAgentV2Ext) 角色授予 Composer Service Agent 账号。Managed Airflow 使用此账号在您的 Google Cloud 项目中执行操作。

探索环境级健康状况和性能的关键指标

本教程重点介绍了一些关键指标，这些指标可让您大致了解环境的总体健康状况和性能。

Google Cloud 控制台中的监控信息中心包含各种指标和图表，可用于监控环境中的趋势，并发现 Airflow 组件和 Managed Airflow 资源的问题。

每个 Managed Airflow 环境都有自己的监控信息中心。

熟悉以下关键指标，并在“监控”信息中心内找到每个指标：

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

   转到“环境”

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

3. 转到监控标签页。

4. 选择概览部分，在信息中心内找到环境概览项，然后查看环境健康情况（Airflow 监控 DAG）指标。

   • 此时间轴显示了 Managed Airflow 环境的健康状况。环境健康状况栏的绿色表示环境健康状况良好，红色表示环境健康状况不佳。

   • Managed Airflow 每隔几分钟就会执行一个名为 airflow_monitoring 的活跃度 DAG。如果活性 DAG 运行成功完成，健康状况为 True。如果活跃性 DAG 运行失败（例如，由于 Pod 驱逐、外部进程终止或维护），则运行状况为 False。

5. 选择 SQL 数据库部分，在信息中心内找到数据库健康状况项，然后查看数据库健康状况指标。

   • 此时间轴显示了与您环境的 Cloud SQL 实例的连接状态。绿色的“数据库健康状况”栏表示连接正常，而红色表示连接失败。

   • Airflow 监控 Pod 会定期对数据库执行 ping 操作，如果可以建立连接，则报告健康状况为 True；否则，报告健康状况为 False。

6. 在数据库健康状况项中，查看数据库 CPU 使用率和数据库内存使用率指标。

   • “数据库 CPU 使用量”图表显示了您环境中的 Cloud SQL 数据库实例的 CPU 核心使用量与可用的数据库 CPU 总限额。

   • “数据库内存用量”图表显示了环境中的 Cloud SQL 数据库实例的内存用量与数据库内存总可用限额的对比情况。

7. 选择调度器部分，在信息中心内找到调度器检测信号项，然后观察调度器检测信号指标。

   • 此时间轴显示了 Airflow 调度程序的健康状况。 检查是否存在红色区域，以识别 Airflow 调度程序问题。如果您的环境中有多个调度器，只要至少有一个调度器在响应，心跳状态就是正常。

   • 如果上次收到心跳的时间距离当前时间超过 30 秒（默认值），则调度器会被视为不健康。

8. 选择 DAG 统计信息部分，在信息中心内找到已终止的僵尸任务项，然后观察已终止的僵尸任务指标。

   • 此图表显示了在较小的时间窗口内终止的僵尸任务数量。僵尸任务通常是由 Airflow 进程的外部终止（例如当任务的进程被终止时）引起的。

   • Airflow 调度器会定期终止僵尸任务，这会反映在此图表中。

9. 选择工作器部分，在信息中心内找到工作器容器重启项，然后观察工作器容器重启指标。

   • 图表显示了各个工作器容器的总重启次数。容器重启次数过多可能会影响服务的可用性，或影响将该服务用作依赖项的下游其他服务的可用性。

注意： 您可以在 Cloud Monitoring 中查看每项指标，以便更详细地了解它们。

了解关键指标的基准和可能的修正措施

重要提示： 时间轴指标可提供各个环境组件的健康状况概览；不过，红色区域并不一定表示某个组件不健康。有时，总体为绿色的时间轴中出现的小块红色区域可能是由外部因素造成的，例如环境集群中的维护操作。使用相应的监控图表和日志来调查潜在原因，分析不同指标之间的相关性，并根据确定的根本原因进行修正。

下表介绍了可能表明存在问题的基准值，并提供了您可能采取的纠正措施来解决这些问题。

• 环境健康情况（Airflow 监控 DAG）

  • 在 4 小时的时间窗口内，成功率低于 90%

  • 失败可能意味着 Pod 被逐出或工作器终止，因为环境过载或出现故障。环境健康时间轴上的红色区域通常与各个环境组件的其他健康栏中的红色区域相关。通过查看 Monitoring 信息中心中的其他指标来确定根本原因。

• 数据库健康状况

  • 在 4 小时的时间窗口内，成功率低于 95%

  • 失败意味着与 Airflow 数据库的连接存在问题，这可能是数据库崩溃或停机所致，因为数据库过载（例如，CPU 或内存使用率过高，或者连接到数据库时的延迟时间较长）。这些症状最常由次优 DAG 引起，例如当 DAG 使用许多全局定义的 Airflow 或环境变量时。通过查看 SQL 数据库资源使用情况指标来确定根本原因。您还可以检查调度程序日志，了解是否存在与数据库连接相关的错误。

• 数据库 CPU 和内存用量

  • 在 12 小时的时间窗口内，平均 CPU 或内存用量超过 80%

  • 数据库可能过载。分析 DAG 运行与数据库 CPU 或内存用量峰值之间的相关性。

    • 您可以通过更高效的 DAG（其中包含优化的运行查询和连接）来减少数据库负载，也可以通过在一段时间内更均匀地分配负载来减少数据库负载。

    • 或者，您可以为数据库分配更多 CPU 或内存。 数据库资源受环境的环境大小属性控制，您必须将环境扩容到更大的规模。

• 调度程序检测信号

  • 在 4 小时的时间窗口内，成功率低于 90%

  • 为调度程序分配更多资源，或将调度程序的数量从 1 个增加到 2 个（推荐）。

• 已终止的僵尸任务

  • 每 24 小时内有多个僵尸任务

  • 僵尸任务最常见的原因是环境集群中的 CPU 或内存资源不足。查看工作器资源使用情况图表，并为工作器分配更多资源，或增加僵尸任务超时时间，以便调度器在将任务视为僵尸任务之前等待更长时间。

• 工作器容器重启

  • 每 24 小时重启一次以上

  • 最常见的原因是工作器内存或存储空间不足。查看工作器的资源消耗情况，并为工作器分配更多内存或存储空间。如果资源不足不是原因，请参阅排查工作器重启事件，并使用日志记录查询来发现工作器重启的原因。

创建通知渠道

按照创建通知渠道中所述的说明创建电子邮件通知渠道。

注意： 为实现冗余，建议您创建多种类型的通知渠道。例如，除了电子邮件之外，您还可以设置 Slack 通知渠道。

如需详细了解通知渠道，请参阅管理通知渠道。

创建提醒政策

根据本教程前面部分提供的基准创建提醒政策，以持续监控指标的值，并在这些指标违反条件时接收通知。

控制台

您可以为 Monitoring 信息中心中显示的每个指标设置提醒，只需点击相应项目一角的铃铛图标即可：

图 1. 为监控信息中心内显示的指标创建提醒（点击可放大）

1. 在 Monitoring 信息中心内找到要监控的每个指标，然后点击相应指标项角落中的铃铛图标。系统会打开创建提醒政策页面。

   重要提示： 系统会自动应用指标名称、环境名称和位置过滤条件，但部分指标的提醒政策需要额外的过滤条件和不同的指标名称。请参阅各个指标的提醒政策配置中的“过滤条件”和“指标名称”字段，以验证是否应用了正确的过滤条件。

2. 在转换数据部分中：

   1. 按照指标的提醒政策配置中的说明，配置在每个时序中部分。

   2. 点击下一步，然后按照指标的提醒政策配置中的说明配置配置提醒触发器部分。

3. 点击下一步。

4. 配置通知。展开通知渠道菜单，然后选择您在上一步中创建的通知渠道。

5. 点击确定。

6. 在为提醒政策命名部分中，填写提醒政策名称字段。为每个指标使用描述性名称。使用“为提醒政策命名”值，如指标的提醒政策配置中所述。

7. 点击下一步。

8. 查看提醒政策，然后点击创建政策。

环境健康情况（Airflow 监控 DAG）指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 正常
• API：composer.googleapis.com/environment/healthy

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 > 每个时序内：

  • 滚动窗口：自定义
  • 自定义值：4
  • 自定义单位：小时
  • 滚动窗口函数：比例 (true)

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：低于阈值
  • 阈值：90
  • 条件名称：环境健康状况

• 配置通知并最终确定提醒：

  • 为提醒政策命名：Airflow 环境健康状况

数据库健康状况指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 数据库运行正常
• API：composer.googleapis.com/environment/database_health

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 > 每个时序内：

  • 滚动窗口：自定义
  • 自定义值：4
  • 自定义单位：小时
  • 滚动窗口函数：比例 (true)

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：低于阈值
  • 阈值：95
  • 条件名称：数据库健康状况条件

• 配置通知并最终确定提醒：

  • 将提醒政策命名为：Airflow 数据库健康状况

数据库 CPU 使用量指标 - 提醒政策配置

注意： 为简单起见，在为“数据库内存用量”和“数据库 CPU 用量”指标设置提醒时，请使用指标名称数据库内存利用率 (composer.googleapis.com/environment/database/memory/utilization) 和数据库 CPU 利用率 (composer.googleapis.com/environment/database/cpu/utilization)，而不是监控已用字节数 (composer.googleapis.com/environment/database/memory/bytes_used) 和已用时间 (composer.googleapis.com/environment/database/cpu/usage_time) 的默认指标名称。“数据库内存利用率”和“数据库 CPU 利用率”指标会自动监控已用 CPU 和内存与可用 CPU 和内存的百分比，这有助于避免在配置条件触发器时进行不必要的计算。

• 指标名称：Cloud Composer 环境 - 数据库 CPU 利用率
• API：composer.googleapis.com/environment/database/cpu/utilization

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 > 每个时序内：

  • 滚动窗口：自定义
  • 自定义值：12
  • 自定义单位：小时
  • 滚动窗口函数：平均值

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：高于阈值
  • 阈值：80
  • 条件名称：数据库 CPU 使用率条件

• 配置通知并最终确定提醒：

  • 将提醒政策命名为：Airflow 数据库 CPU 使用率

数据库内存用量指标 - 提醒政策配置

注意： 为简单起见，在为“数据库内存用量”和“数据库 CPU 用量”指标设置提醒时，请使用指标名称数据库内存利用率 (composer.googleapis.com/environment/database/memory/utilization) 和数据库 CPU 利用率 (composer.googleapis.com/environment/database/cpu/utilization)，而不是监控已用字节数 (composer.googleapis.com/environment/database/memory/bytes_used) 和已用时间 (composer.googleapis.com/environment/database/cpu/usage_time) 的默认指标名称。“数据库内存利用率”和“数据库 CPU 利用率”指标会自动监控已用 CPU 和内存与可用 CPU 和内存的百分比，这有助于避免在配置条件触发器时进行不必要的计算。

• 指标名称：Cloud Composer 环境 - 数据库内存利用率
• API：composer.googleapis.com/environment/database/memory/utilization

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 > 每个时序内：

  • 滚动窗口：自定义
  • 自定义值：12
  • 自定义单位：小时
  • 滚动窗口函数：平均值

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：高于阈值
  • 阈值：80
  • 条件名称：数据库内存用量条件

• 配置通知并最终确定提醒：

  • 将提醒政策命名为：Airflow 数据库内存用量

调度器检测信号指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 调度程序心跳
• API：composer.googleapis.com/environment/scheduler_heartbeat_count

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 > 每个时序内：

  • 滚动窗口：自定义
  • 自定义值：4
  • 自定义单位：小时
  • 滚动窗口函数：count

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：低于阈值

  • 阈值：216

    注意： 空闲环境中的平均调度程序心跳计数为 240，因此在 4 小时窗口内成功率为 90% 等于 240 * 0.9 =
    1. 您可以在 Metrics Explorer 查询编辑器中运行可汇总值 _scheduler_heartbeat_count_mean 的查询，以获取此数字。

  • 条件名称：调度程序检测信号条件

• 配置通知并最终确定提醒：

  • 将提醒政策命名为：Airflow 调度程序检测信号

“已终止的僵尸任务”指标 - 提醒政策配置

• 指标名称：Cloud Composer 环境 - 已终止的僵尸任务
• API：composer.googleapis.com/environment/zombie_task_killed_count

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  

• 转换数据 > 每个时序内：

  • 滚动窗口：1 天
  • 滚动窗口函数：sum

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：高于阈值
  • 阈值：1
  • 条件名称：僵尸任务条件

• 配置通知并最终确定提醒：

  • 为提醒政策命名：Airflow 僵尸任务

工作器容器重启次数指标 - 提醒政策配置

注意： 对于工作器容器重启次数指标，您需要为要监控的 pod、工作器和集群设置额外的过滤条件。如需在默认过滤条件的基础上添加更多过滤条件，请点击添加过滤条件。

• 指标名称：Kubernetes 容器 - 重启次数
• API：kubernetes.io/container/restart_count

• 过滤条件：

  environment_name = [ENVIRONMENT_NAME]
  location = [CLUSTER_LOCATION]
  pod_name =~ airflow-worker-.*|airflow-k8s-worker-.*
  container_name =~ airflow-worker|base
  cluster_name = [CLUSTER_NAME]
  

  CLUSTER_NAME 是您环境的集群名称，可在 Google Cloud 控制台中的“环境配置”>“资源”>“GKE 集群”下找到。

• 转换数据 > 每个时序内：

  • 滚动窗口：1 天
  • 滚动窗口函数：速率

• 配置提醒触发器：

  • 条件类型：阈值
  • 提醒触发器：任何时序违反
  • 阈值位置：高于阈值
  • 阈值：1
  • 条件名称：工作器容器重启条件

• 配置通知并最终确定提醒：

  • 将提醒政策命名为：Airflow Worker Restarts

Terraform

运行一个 Terraform 脚本，该脚本会创建一个电子邮件通知渠道，并根据本教程中提供的关键指标各自的基准上传相应的提醒政策：

1. 将示例 Terraform 文件保存到本地计算机上。

2. 替换以下内容：

   • PROJECT_ID：您的项目的项目 ID。例如 example-project。
   • EMAIL_ADDRESS：在触发提醒时必须通知的电子邮件地址。
   • ENVIRONMENT_NAME：您的 Managed Airflow 环境的名称。 例如 example-composer-environment。
   • CLUSTER_NAME：您的环境集群名称，可在 Google Cloud 控制台中的“环境配置”>“资源”>“GKE 集群”下找到。

resource "google_monitoring_notification_channel" "basic" {
  project      = "PROJECT_ID"
  display_name = "Test Notification Channel"
  type         = "email"
  labels = {
    email_address = "EMAIL_ADDRESS"
  }
  # force_delete = false
}

resource "google_monitoring_alert_policy" "environment_health_metric" {
  project      = "PROJECT_ID"
  display_name = "Airflow Environment Health"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Environment health condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/healthy\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_LT"
      threshold_value = 0.9
      aggregations {
        alignment_period   = "14400s"
        per_series_aligner = "ALIGN_FRACTION_TRUE"
      }
    }
  }

}

resource "google_monitoring_alert_policy" "database_health_metric" {
  project      = "PROJECT_ID"
  display_name = "Airflow Database Health"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Database health condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/database_health\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_LT"
      threshold_value = 0.95
      aggregations {
        alignment_period   = "14400s"
        per_series_aligner = "ALIGN_FRACTION_TRUE"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_database_cpu_usage" {
  project      = "PROJECT_ID"
  display_name = "Airflow Database CPU Usage"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Database CPU usage condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/database/cpu/utilization\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 80
      aggregations {
        alignment_period   = "43200s"
        per_series_aligner = "ALIGN_MEAN"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_database_memory_usage" {
  project      = "PROJECT_ID"
  display_name = "Airflow Database Memory Usage"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Database memory usage condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/database/memory/utilization\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 80
      aggregations {
        alignment_period   = "43200s"
        per_series_aligner = "ALIGN_MEAN"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_scheduler_heartbeat" {
  project      = "PROJECT_ID"
  display_name = "Airflow Scheduler Heartbeat"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Scheduler heartbeat condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/scheduler_heartbeat_count\" AND resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_LT"
      threshold_value = 216 // Threshold is 90% of the average for composer.googleapis.com/environment/scheduler_heartbeat_count metric in an idle environment
      aggregations {
        alignment_period   = "14400s"
        per_series_aligner = "ALIGN_COUNT"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_zombie_task" {
  project      = "PROJECT_ID"
  display_name = "Airflow Zombie Tasks"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Zombie tasks condition"
    condition_threshold {
      filter     = "resource.type = \"cloud_composer_environment\" AND metric.type=\"composer.googleapis.com/environment/zombie_task_killed_count\" AND  resource.label.environment_name=\"ENVIRONMENT_NAME\""
      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 1
      aggregations {
        alignment_period   = "86400s"
        per_series_aligner = "ALIGN_SUM"
      }
    }
  }
}

resource "google_monitoring_alert_policy" "alert_worker_restarts" {
  project      = "PROJECT_ID"
  display_name = "Airflow Worker Restarts"
  combiner     = "OR"
  notification_channels = [google_monitoring_notification_channel.basic.name] // To manually add a notification channel add it with the syntax "projects/[PROJECT_ID]/notificationChannels/[CHANNEL_ID]"
  conditions {
    display_name = "Worker container restarts condition"
    condition_threshold {
      filter     = "resource.type = \"k8s_container\" AND (resource.labels.cluster_name = \"CLUSTER_NAME\" AND resource.labels.container_name = monitoring.regex.full_match(\"airflow-worker|base\") AND resource.labels.pod_name = monitoring.regex.full_match(\"airflow-worker-.*|airflow-k8s-worker-.*\")) AND metric.type = \"kubernetes.io/container/restart_count\""

      duration   = "60s"
      comparison = "COMPARISON_GT"
      threshold_value = 1
      aggregations {
        alignment_period   = "86400s"
        per_series_aligner = "ALIGN_RATE"
      }
    }
  }
}

测试提醒政策

本部分介绍了如何测试创建的提醒政策并解读结果。

上传示例 DAG

本教程中提供的示例 DAG memory_consumption_dag.py 模拟了高强度的工作器内存利用率。该 DAG 包含 4 个任务，每个任务都会将数据写入一个示例字符串，消耗 380 MB 的内存。示例 DAG 已安排为每 2 分钟运行一次，在您将其上传到 Composer 环境后，它会自动开始运行。

警告： 完成本教程后，请务必删除或暂停此 DAG，否则您的环境可能会消耗大量资源，从而产生额外的费用。

将以下示例 DAG 上传到您在之前步骤中创建的环境：

from datetime import datetime
import sys
import time

from airflow import DAG
from airflow.operators.python import PythonOperator


def ram_function():
    data = ""
    start = time.time()
    for i in range(38):
        data += "a" * 10 * 1000**2
        time.sleep(0.2)
        print(f"{i}, {round(time.time() - start, 4)}, {sys.getsizeof(data) / (1000 ** 3)}")
    print(f"Size={sys.getsizeof(data) / (1000 ** 3)}GB")
    time.sleep(30 - (time.time() - start))
    print(f"Complete in {round(time.time() - start, 2)} seconds!")


with DAG(
    dag_id="memory_consumption_dag",
    start_date=datetime(2023, 1, 1, 1, 1, 1),
    schedule="1/2 * * * *",
    catchup=False,
) as dag:
    for i in range(4):
        PythonOperator(
            task_id=f"task_{i+1}",
            python_callable=ram_function,
            retries=0,
            dag=dag,
        )

解读 Monitoring 中的提醒和指标

等待示例 DAG 开始运行约 10 分钟后，评估测试结果：

1. 检查您的电子邮件邮箱，验证您是否收到了来自Google Cloud Alerting 的通知，该通知的主题行以 [ALERT] 开头。此消息的内容包含提醒政策突发事件的详细信息。

2. 点击电子邮件通知中的查看突发事件按钮。系统会将您重定向到 Metrics Explorer。查看提醒突发事件的详细信息：

   

   图 2. 提醒事件的详细信息（点击可放大）

   突发事件指标图表表明，您创建的指标超过了阈值 1，这意味着 Airflow 检测到并终止了 1 个以上的僵尸任务。

3. 在 Managed Airflow 环境中，前往监控标签页，打开 DAG 统计信息部分，然后找到已终止的僵尸任务图表：

   

   图 3. 僵尸任务图（点击可放大）

   该图表显示，在运行示例 DAG 的前 10 分钟内，Airflow 终止了大约 20 个僵尸任务。

4. 根据基准和纠正措施，僵尸任务最常见的原因是工作器内存或 CPU 不足。通过分析工作器资源利用率，确定僵尸任务的根本原因。

   在监控信息中心内打开“工作器”部分，然后查看工作器的 CPU 和内存用量指标：

   

   图 4. 工作器 CPU 和内存用量指标（点击可放大）

   “工作器 CPU 总用量”图表表明，工作器 CPU 用量始终低于可用总限额的 50%，因此可用 CPU 充足。“工作器总内存用量”图表显示，运行示例 DAG 导致内存用量达到可分配内存上限，该上限几乎等于图表上显示的总内存上限的 75%（GKE 会预留前 4 GiB 内存的 25%，并在每个节点上额外预留 100 MiB 内存，用于处理 Pod 逐出）。

   由此可以得出结论，工作器缺少成功运行示例 DAG 所需的内存资源。

优化环境并评估其性能

根据对工作器资源利用率的分析，您需要为 DAG 中的所有任务成功执行而为工作器分配更多内存。

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

1. 在 Composer 环境中，打开 DAG 标签页，点击示例 DAG (memory_consumption_dag) 的名称，然后点击暂停 DAG。

2. 分配额外的工作器内存：

   1. 在“环境配置”标签页中，找到资源 > 工作负载配置，然后点击修改。

   2. 在 Worker 项中，增加内存限制。在本教程中，请使用 3.25 GB。

   3. 保存更改，并等待工作器重启（可能需要几分钟时间）。

3. 打开“DAG”标签页，点击示例 DAG 的名称 (memory_consumption_dag)，然后点击解除 DAG 的暂停状态。

前往监控，并验证在更新工作器资源限制后是否未出现新的僵尸任务：

图 5. 更改内存限制后的僵尸任务图（点击可放大）

注意： 建议长时间观察环境性能。例如，在运行许多 DAG 的生产环境中，我们建议在 24 小时后仔细检查是否没有新提醒。

摘要

在本教程中，您了解了关键的环境级健康状况和性能指标，学习了如何为每个指标设置提醒政策，以及如何将每个指标解读为纠正措施。然后，您运行了一个示例 DAG，在提醒和监控图表的帮助下确定了环境健康问题的根本原因，并通过为工作器分配更多内存来优化环境。不过，建议您先优化 DAG，以减少工作器资源消耗，因为资源无法增加到超出某个阈值。

清理

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

删除项目

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

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

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

1. 在 Google Cloud 控制台中，前往管理资源页面。

   转到“管理资源”

2. 在项目列表中，选择要删除的项目，然后点击删除。
3. 在对话框中输入项目 ID，然后点击关闭以删除项目。

删除各个资源

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

控制台

1. 删除 Managed Airflow 环境。在此过程中，您还可以删除环境的存储桶。
2. 删除您在 Cloud Monitoring 中创建的每项提醒政策。

Terraform

1. 确保您的 Terraform 脚本不包含项目仍需要的资源条目。例如，您可能希望保留一些已启用的 API 和仍分配的 IAM 权限（如果您已将此类定义添加到 Terraform 脚本中）。
2. 运行 terraform destroy。
3. 手动删除环境的存储桶。Managed Airflow 不会自动删除该文件。您可以通过 Google Cloud 控制台或 Google Cloud CLI 执行此操作。

后续步骤

• 优化环境
• 扩缩环境
• 管理环境标签并细分环境费用