【实战ES】实战 Elasticsearch：快速上手与深度实践-5.1.2基于Painless脚本的日志告警

👉 点击关注不迷路
👉 点击关注不迷路
👉 点击关注不迷路

5.1.2 基于Painless脚本的日志告警深度实践

• 是 Elasticsearch 中一种用于编写脚本的编程语言和执行环境，专为高性能、简单和安全的脚本执行设计。
  • 高性能： Painless 经过了优化，能够在 Elasticsearch 中高效执行。它可以充分利用 Elasticsearch 的底层架构和资源，实现快速的数据处理和操作。
  • 与 Elasticsearch 深度集成： Painless 与 Elasticsearch 紧密结合，可以直接访问和操作 Elasticsearch 中的数据、文档、索引等对象。它能够方便地与 Elasticsearch 的各种功能和 API 进行交互，实现复杂的业务逻辑。
• Painless脚本在Elastic Watcher中的执行流程

1. 核心原理与技术优势

1.1 Painless脚本特性对比

1.2 告警触发机制原理

// 伪代码：Watcher执行引擎工作流程
public class AlertEngine {
    void executeWatch(Watch watch) {
        Trigger trigger = watch.trigger();  // 调度触发器
        Condition condition = watch.condition(); // Painless脚本条件
        List<Action> actions = watch.actions(); // 告警动作
        
        while (trigger.hasNextExecution()) {
            if (condition.execute(ctx)) {  // 执行脚本判断
                for (Action action : actions) {
                    action.execute(ctx);    // 触发告警动作
                }
            }
        }
    }
}

2. 告警规则配置实战

2.1 基础告警模板

// 创建一个名为 error_rate_alert 的 Watcher 监控任务
PUT _watcher/watch/error_rate_alert
{
  // 定义监控任务的触发条件
  "trigger": {
    // 使用定时调度触发
    "schedule": {
      // 每隔 1 分钟触发一次
      "interval": "1m"
    }
  },
  // 定义监控任务的数据输入方式
  "input": {
    // 使用搜索作为输入方式
    "search": {
      "request": {
        // 指定要搜索的索引，这里使用通配符匹配所有以 logs- 开头的索引
        "indices": ["logs-*"],
        "body": {
          // 设置搜索结果返回的文档数量为 0，因为这里只关注聚合结果
          "size": 0,
          // 定义搜索查询条件
          "query": {
            // 使用布尔查询组合多个过滤条件
            "bool": {
              "filter": [
                // 过滤出 @timestamp 字段在过去 1 分钟内的文档
                { "range": { "@timestamp": { "gte": "now-1m/m" }}},
                // 过滤出 log.level 字段值为 "error" 的文档
                { "term": { "log.level": "error" }}
              ]
            }
          },
          // 定义聚合操作
          "aggs": {
            // 定义一个名为 error_count 的聚合，用于统计 log.level 字段的数量
            "error_count": { "value_count": { "field": "log.level" }}
          }
        }
      }
    }
  },
  // 定义监控任务的触发条件判断逻辑
  "condition": {
    // 使用脚本作为条件判断方式
    "script": {
      // Painless 脚本的源代码
      "source": """
        // 如果搜索结果中的错误数量大于设定的阈值
        if (ctx.payload.aggregations.error_count.value > params.threshold) {
          // 返回 true，表示条件满足
          return true;
        }
        // 否则返回 false，表示条件不满足
        return false;
      """,
      // 传递给脚本的参数
      "params": {
        // 设定错误数量的阈值为 50
        "threshold": 50
      }
    }
  },
  // 定义当条件满足时要执行的操作
  "actions": {
    // 定义一个名为 send_email 的操作
    "send_email": {
      // 使用电子邮件作为操作类型
      "email": {
        // 指定邮件的收件人
        "to": ["ops@example.com"],
        // 定义邮件的主题，使用上下文变量动态生成
        "subject": "错误日志告警：{{ctx.watch_id}}",
        // 定义邮件的正文内容，使用上下文变量动态生成
        "body": """
          检测时间：{{ctx.execution_time}}
          错误数量：{{ctx.payload.aggregations.error_count.value}}
          触发阈值：{{ctx.condition.params.threshold}}
        """
      }
    }
  }
}

2.2 多维度告警规则矩阵

3. 高级脚本开发技巧

3.1 时间序列分析函数

// 计算滑动窗口平均值
// 从 Elasticsearch Watcher 上下文的负载（payload）中提取一系列命中结果（hits）
// 假设 ctx 是 Elasticsearch Watcher 脚本的上下文对象
// ctx.payload.series.hits.hits 表示一系列的文档命中结果
// 将这些命中结果转换为一个流（Stream），以便后续进行处理
double[] values = ctx.payload.series.hits.hits.stream()
    // 对流中的每个命中结果（hit）进行映射操作
    // 从每个命中结果的源数据（_source）中提取 response_time 字段的值
    // 并将其转换为 double 类型
    .mapToDouble(hit -> hit._source.response_time)
    // 将映射后的 double 类型值收集到一个数组中
    .toArray();

// 创建一个 MovingAverage 对象，用于计算移动平均值
// 第一个参数 30 表示移动平均的窗口大小，即计算最近 30 个值的平均值
// 第二个参数 values 是之前提取的 response_time 数组
// 第三个参数 "simple" 表示使用简单移动平均算法
MovingAverage movingAvgObj = new MovingAverage(30, values, "simple");
// 从 MovingAverage 对象中获取计算得到的移动平均值
double movingAvg = movingAvgObj.value;

// 检查计算得到的移动平均值是否超过了预设的阈值
// params 是传递给脚本的参数对象，threshold 是预设的阈值
if (movingAvg > params.threshold) {
    // 如果移动平均值超过了阈值，返回 true，表示满足条件
    return true;
}

3.2 自定义预警级别

// 多级告警判断逻辑
long errorCount = ctx.payload.aggregations.errors.value;
long warnCount = ctx.payload.aggregations.warns.value;

if (errorCount > 100) {
    ctx.alert_level = "CRITICAL";
} else if (errorCount > 50 || warnCount > 200) {
    ctx.alert_level = "WARNING";
} else {
    return false;
}

return true;

4. 性能优化策略

4.1 脚本执行参数调优

# Elasticsearch配置
watcher:
  execution:
    thread_pool:
      size: 8               # 默认4
      queue_size: 1000      # 默认100
  indices:
    buffer_size: 50%        # JVM堆内存占比

# Watcher模板参数
{
  "settings": {
    "script.max_compilations_rate": "1000/1m",
    "script.cache.max_size": 1000
  }
}

4.2 性能测试数据对比

5. 企业级告警方案

5.1 电商大促监控案例

// 创建一个名为 black_friday_alert 的 Elasticsearch Watcher 监控任务
PUT _watcher/watch/black_friday_alert
{
    // 定义监控任务的触发规则
    "trigger": {
        // 采用定时调度触发方式
        "schedule": {
            // 每 10 秒触发一次监控任务
            "interval": "10s"
        }
    },
    // 定义监控任务的数据输入来源
    "input": {
        // 使用搜索查询作为数据输入方式
        "search": {
            "request": {
                // 指定要搜索的索引，这里使用通配符匹配所有以 logs - ecommerce - 开头的索引
                "indices": ["logs-ecommerce-*"],
                "body": {
                    // 定义搜索的查询条件
                    "query": {
                        // 筛选出 @timestamp 字段值在当前时间往前推 10 秒内的文档
                        "range": {
                            "@timestamp": {
                                "gte": "now-10s"
                            }
                        }
                    },
                    // 定义聚合操作，用于对搜索结果进行统计分析
                    "aggs": {
                        // 计算 error_count 字段的平均值，将结果命名为 error_rate
                        "error_rate": {
                            "avg": {
                                "field": "error_count"
                            }
                        },
                        // 计算 response_time 字段的 99 分位数，将结果命名为 slow_api
                        "slow_api": {
                            "percentiles": {
                                "field": "response_time",
                                // 指定要计算的分位数值为 99
                                "percents": [99]
                            }
                        }
                    }
                }
            }
        }
    },
    // 定义监控任务的触发条件判断逻辑
    "condition": {
        // 使用脚本作为条件判断方式
        "script": {
            // 编写 Painless 脚本逻辑
            "source": """
                // 从搜索结果的聚合数据中提取 error_rate 的值，并存储为 double 类型的变量
                double errorRate = ctx.payload.aggregations.error_rate.value;
                // 从搜索结果的聚合数据中提取 slow_api 聚合下 99 分位数值，并存储为 double 类型的变量
                double p99 = ctx.payload.aggregations.slow_api.values['99.0'];
                
                // 判断是否满足触发告警的基本条件
                if (errorRate > 0.1 || p99 > 5000) {
                    // 根据更严格的条件判断告警的严重程度
                    ctx.severity = (errorRate > 0.3 || p99 > 10000) ? 'CRITICAL' : 'WARNING';
                    // 满足基本触发条件，返回 true 表示触发告警
                    return true;
                }
                // 不满足基本触发条件，返回 false 表示不触发告警
                return false;
            """
        }
    },
    // 定义当触发条件满足时要执行的操作
    "actions": {
        // 定义一个名为 slack_alert 的操作
        "slack_alert": {
            // 使用 Webhook 方式执行操作
            "webhook": {
                // 指定 Slack Webhook 的 URL，用于发送消息到 Slack 频道
                "url": "https://hooks.slack.com/services/TXXXXXX/BXXXXXX",
                // 定义要发送到 Slack 的消息体
                "body": """
                    {
                        "text": "【{{ctx.severity}}】系统异常告警",
                        "blocks": [
                            {
                                "type": "section",
                                "text": {
                                    "type": "mrkdwn",
                                    "text": "*错误率*: {{ctx.payload.aggregations.error_rate.value}}\n*P99延迟*: {{ctx.payload.aggregations.slow_api.values['99.0']}}ms"
                                }
                            }
                        ]
                    }
                """
            }
        }
    }
}

5.2 实施效果统计

6. 安全与稳定性保障

6.1 权限控制矩阵

6.2 故障熔断机制

// 异常处理脚本示例
try {
    def result = someComplexCalculation();
    return result > threshold;
} catch (CircuitBreakingException e) {
    ctx.metadata.put("circuit_breaker", "open");
    return false; // 主动抑制告警
} catch (Exception e) {
    logger.error("Script execution failed", e);
    throw e;
}

7. 最佳实践指南

• 1. 脚本版本管理

  # 存储脚本到集群状态
  # 该请求用于将自定义的 Painless 脚本存储到 Elasticsearch 集群状态中，方便后续引用
  POST _scripts/alert_logic_v1
  {
      "script": {
          // 指定脚本使用的语言为 Painless，Painless 是 Elasticsearch 内置的脚本语言
          "lang": "painless",
          // 脚本的具体内容，这里的逻辑是判断 ctx.payload.hits.total 是否大于传入的阈值
          // ctx 是 Elasticsearch 脚本中的上下文对象，payload 通常包含搜索结果等信息
          // hits.total 表示搜索命中的文档总数
          // params 是传递给脚本的参数对象，用于动态传入阈值
          "source": "ctx.payload.hits.total > params.threshold"
      }
  }
  
  # 引用存储脚本
  # 此部分是在某个配置（如 Watcher 的条件配置）中引用之前存储的脚本
  "condition": {
      "script": {
          // 指定要引用的脚本的 ID，即之前存储脚本时使用的名称 alert_logic_v1
          "id": "alert_logic_v1",
          // 传递给脚本的参数，这里设置阈值为 100
          // 脚本在执行时会使用这个参数来进行条件判断
          "params": {
              "threshold": 100
          }
      }
  }
  

• 2. 监控告警系统自身

// 创建一个名为 watcher_health 的 Elasticsearch Watcher 监控任务
PUT _watcher/watch/watcher_health
{
    // 定义监控任务的触发规则
    "trigger": {
        // 采用定时调度的触发方式
        "schedule": {
            // 每 5 分钟触发一次该监控任务
            "interval": "5m"
        }
    },
    // 定义监控任务的数据输入方式
    "input": {
        // 使用搜索查询来获取数据，这里省略了具体的搜索请求体，用 ... 表示
        "search": {
            ...
        }
    },
    // 定义监控任务的触发条件判断逻辑
    "condition": {
        // 使用脚本作为触发条件的判断方式
        "script": {
            // 编写 Painless 脚本逻辑
            "source": """
                // 从搜索结果的聚合数据中提取名为 watcher_stats 的聚合统计信息
                // ctx.payload 包含了搜索查询的结果负载，aggregations 是其中的聚合结果部分
                def stats = ctx.payload.aggregations.watcher_stats;
                // 判断是否满足触发告警的条件
                // stats.avg_execution_time 表示平均执行时间
                // stats.failure_rate 表示失败率
                if (stats.avg_execution_time > 5000 || 
                    stats.failure_rate > 0.1) {
                    // 如果平均执行时间超过 5000 或者失败率超过 0.1（即 10%）
                    // 则返回 true，表示触发该监控任务对应的后续操作
                    return true;
                }
                // 如果不满足上述条件，则默认返回 false，不触发后续操作
                // 这里虽然没有显式的 return false，但 Painless 脚本在没有其他返回语句时，最后一行语句的执行结果会作为返回值，
                // 由于 if 条件不满足时没有额外的返回，所以会隐式返回 false
            """
        }
    }
}

附录：常用调试工具

关键注意事项：

1. 避免在脚本中进行全量数据遍历
2. 重要业务脚本需进行单元测试
3. 定期清理历史执行记录
4. 生产环境禁用动态脚本