ElastAlert 自定义 Alerter 开发指南：从零编写自己的告警器插件

告警异常检测【免费下载链接】elastalertEasy Flexible Alerting With ElasticSearch项目地址https://gitcode.com/gh_mirrors/el/elastalert点击查看免费下载本指南基于 ElastAlert 官方配方文档 docs/source/recipes/adding_alerts.rst完整讲解如何通过继承Alerter基类编写自定义告警器Alerter并将其接入规则配置。读完本文你将掌握 ElastAlert 的告警器抽象模型、两个核心方法alert与get_info的实现约定、pipeline跨告警器数据传递机制并能独立完成一个把告警写入本地文件的可用插件与配套规则配置。告警器是什么ElastAlert 的告警抽象层ElastAlert 将规则命中match与通知动作解耦规则类型如 frequency、spike、change负责在 Elasticsearch 中查找匹配事件而告警器Alerter负责把匹配结果转换为具体的通知动作——发送邮件、创建 JIRA 工单、推送 Slack 消息等。从源码结构看所有内置告警器都继承自位于 elastalert/alerts.py 的基类Alerterclass Alerter(object): Base class for types of alerts. :param rule: The rule configuration. required_options frozenset([]) def __init__(self, rule): self.rule rule self.pipeline None self.resolve_rule_references(self.rule) def alert(self, match): raise NotImplementedError() def get_info(self): return {type: Unknown}一个自定义告警器本质上就是Alerter的子类ElastAlert 在启动时实例化它之后周期性把匹配数据通过alert方法传给它执行动作再通过get_info把告警元信息写回 Elasticsearch。官方文档给出的最小骨架如下class AwesomeNewAlerter(Alerter): required_options set([some_config_option]) def alert(self, matches): ... def get_info(self): ...自定义告警器通过模块路径 类名的字符串形式在规则文件中被引用格式为module.file.AlertNamemodule是 Python 模块名file是包含Alerter子类的 Python 文件名AlertName即类名。ElastAlert 会据此执行from module.file import AlertName完成加载。三个关键成员required_options、rule 与 pipeline官方文档明确了自定义告警器生命周期中最重要的三个成员属性self.required_options一个包含必选配置项名称的集合。若规则配置中缺少其中任意一项ElastAlert 将拒绝实例化该告警器并抛出异常。这为你的插件提供了一道声明式的配置校验。self.rule规则的完整配置字典。告警器特有的全部配置项都应放在规则文件中并通过self.rule访问。Alerter.__init__在赋值self.rule后还会调用resolve_rule_references支持以$option$形式引用规则中其他顶层配置项见 elastalert/alerts.py。self.pipeline用于在多个告警器之间传递信息的共享字典。每次告警触发时ElastAlert 会新建一个空的 pipeline 对象然后按规则文件中定义告警器的顺序依次调用各告警器先执行的告警器可以向 pipeline 写入数据供后执行的告警器读取。pipeline 的经典实例是 JIRA 与 Email 的协作在 elastalert/alerts.py 中JiraAlerter创建工单后会把jira_ticket与jira_server写入self.pipeline随后EmailAlerter在构造邮件正文时检查 pipelineelastalert/alerts.py若存在工单号则把http://server/browse/TICKET链接追加进邮件。pipeline 的创建与分发逻辑位于 elastalert/elastalert.py 的send_alert方法中# Alert.pipeline is a single object shared between every alerter # This allows alerters to pass objects and data between themselves alert_pipeline {alert_time: alert_time} for alert in rule[alert]: alert.pipeline alert_pipeline try: alert.alert(matches) except EAException as e: ...注意pipeline 初始自带alert_time键所有告警器共享同一个pipeline 对象实例因此先写后读的顺序依赖要求你在规则文件中合理安排告警器的声明顺序。alert(self, matches)告警动作的入口alert是 ElastAlert 触发告警时调用的方法matches是一个字典对象的列表每个字典包含一次命中事件的相关信息。需要注意两点列表而非单元素当规则设置了aggregation聚合选项时同一轮触发可能携带多条匹配因此必须用循环逐个处理。可读性格式化官方文档建议通过self.rule[type].get_match_str(match, self.rule)获取匹配事件的人类可读字符串在实际代码中更常见的是使用BasicMatchString工具类同样位于 elastalert/alerts.py它会把规则名、自定义alert_text、top_events_*计数以及匹配字段按键名排序输出为整齐的文本。alert的异常处理约定也很重要若该方法抛出异常ElastAlert 会捕获它将本次告警标记为**未发送unsent**并保存起来等待后续重试。在 elastalert/elastalert.py 中EAException会被handle_error记录alert_sent保持False同时alert_exception会随告警信息一起写回 Elasticsearch 的elastalert索引。get_info(self)回写 Elasticsearch 的元信息get_info在告警发送后被调用用于收集本次告警的元信息并写入 Elasticsearch。它必须返回一个字典该字典会被直接上传应包含告警的类型、收件人、参数等有诊断价值的信息。从 elastalert/elastalert.py 的get_alert_body可以看到get_info()的返回值被写入写入回索引文档的alert_info字段body { match_body: match, rule_name: rule[name], alert_info: rule[alert][0].get_info() if not self.debug else {}, alert_sent: alert_sent, alert_time: alert_time }也就是说你可以在get_info中返回如{type: Awesome Alerter, output_file: /tmp/alerts.log}这样的结构之后便能在elastalert索引中按alert_info.type、alert_info.output_file检索每一次告警的投递情况。内置告警器也遵循该约定例如DebugAlerter返回{type: debug}、StompAlerter返回{type: stomp}见 elastalert/alerts.py。动手实践编写一个本地文件告警器下面完整复现官方教程创建一个把告警追加写入本地文件的告警器。第一步创建模块目录在 ElastAlert 根目录下创建可被 Python 导入的模块包$ mkdir elastalert_modules $ cd elastalert_modules $ touch __init__.py__init__.py使该目录成为 Python 包。由于 ElastAlert 通过 elastalert/util.py 的get_module加载告警器内部执行sys.path.append(os.getcwd())后__import__模块目录必须位于可导入的位置例如 ElastAlert 的启动目录下。第二步编写告警器类在elastalert_modules/my_alerts.py中写入from elastalert.alerts import Alerter, BasicMatchString class AwesomeNewAlerter(Alerter): # By setting required_options to a set of strings # You can ensure that the rule config file specifies all # of the options. Otherwise, ElastAlert will throw an exception # when trying to load the rule. required_options set([output_file_path]) # Alert is called def alert(self, matches): # Matches is a list of match dictionaries. # It contains more than one match when the alert has # the aggregation option set for match in matches: # Config options can be accessed with self.rule with open(self.rule[output_file_path], a) as output_file: # basic_match_string will transform the match into the default # human readable string format match_string str(BasicMatchString(self.rule, match)) output_file.write(match_string) # get_info is called after an alert is sent to get data that is written back # to Elasticsearch in the field alert_info # It should return a dict of information relevant to what the alert does def get_info(self): return {type: Awesome Alerter, output_file: self.rule[output_file_path]}逐段解读required_options set([output_file_path])声明必填配置规则中缺少output_file_path时加载失败并抛异常。alert中遍历matches兼容 aggregation 聚合的多匹配通过self.rule[output_file_path]读取配置项用BasicMatchString生成默认人类可读文本后追加写入文件。get_info返回告警类型与目标文件路径便于在 Elasticsearch 中追溯。第三步在规则中引用在规则配置文件中指定告警alert: elastalert_modules.my_alerts.AwesomeNewAlerter output_file_path: /tmp/alerts.logElastAlert 将执行from elastalert_modules.my_alerts import AwesomeNewAlerter来导入该类因此该目录必须位于可被当作 Python 模块导入的位置。加载机制背后的源码细节为了让你理解自定义告警器为何能生效这里补充加载链路的关键实现模块解析elastalert/util.py 的get_module按最后一个点号拆分module_path与module_class先__import__模块再getattr取类任何ImportError、AttributeError、ValueError都会包装为EAException。类校验elastalert/loaders.py 的create_alert先查内置映射表alerts_mapping查不到则走get_module加载外部类随后用issubclass(alert_class, alerts.Alerter)强制校验非Alerter子类会报Alert module %s is not a subclass of Alerter。这意味着你自定义的类必须继承Alerter。必选项校验missing_options (rule[type].required_options | alert_class.required_options) - frozenset(alert_config or [])把规则类型与告警器的必选项合并后与已提供配置取差集缺项即抛EAException(Missing required option(s): ...)——这正是required_options声明式校验的实现位置elastalert/loaders.py。别名与顺序内置告警器支持email、jira、slack等短别名同时 elastalert/loaders.py 定义了alerts_order偏序jira在email之前保证先建工单、再在邮件中引用工单链接。自定义告警器未在alerts_order中时默认顺序值为1。配置合并规则中的alert字段既可以是字符串也可以是{alertType: {key: data}}的字典形式normalize_configelastalert/loaders.py会把字典形式合并进规则配置副本让告警器通过self.rule访问到这些键。此外tests/alerts_test.py 中大量针对EmailAlerter、JiraAlerter等内置告警器的用例构造 rule 字典、调用alert([{...}])、断言 SMTP mock 调用为你编写自定义告警器的单元测试提供了现成范式。小结与进阶建议编写自定义告警器只需三件事继承Alerter、实现alert(matches)、实现get_info()并通过module.file.AlertName字符串在规则中引用。善用required_options做配置校验通过self.rule读取自定义配置项利用self.pipeline与其他告警器协作注意声明顺序决定执行顺序。alert抛异常会导致告警标记为未发送并被保存重试get_info的返回值会进入elastalert索引的alert_info字段建议包含类型、接收方、参数等可诊断信息。进阶方向参考 example_rules 下的示例规则将自定义告警器与aggregation、top_count_keys、alert_text_args等特性组合实现更丰富的通知内容与聚合报表。相关源码与测试告警器基类与内置实现 elastalert/alerts.py加载与校验 elastalert/loaders.py发送与回写流程 elastalert/elastalert.py模块导入工具 elastalert/util.py告警器测试用例 tests/alerts_test.py。赞分享告警异常检测【免费下载链接】elastalertEasy Flexible Alerting With ElasticSearch项目地址https://gitcode.com/gh_mirrors/el/elastalert点击查看免费下载相关推荐从零编写自己的安全规则Deepsec 自定义 Matcher 插件开发实战从零编写自己的安全规则Deepsec 自定义 Matcher 插件开发实战 Deepsec 是一款由 AI 编码智能体驱动的安全漏洞扫描工具能在你的代码库中应用安全漏洞扫描人工智能AI AgentElastAlert 自定义规则开发从 YAML 配置到 Python 插件编写ElastAlert 自定义规则开发从 YAML 配置到 Python 插件编写 ElastAlert 是一款基于 ElasticSearch 的灵活告警工具告警异常检测umi插件开发指南从零开始编写自定义插件umi插件开发指南从零开始编写自定义插件 前言 你是否曾经在使用umi框架时遇到过这样的困境想要扩展项目的功能却发现官方提供的配置选项无法满足需求或者想前端Web框架CLI构建工具上一篇dio与Flutter A2 Hosting LoginA2 Hosting登录请求下一篇告别重复请求Dio请求合并高级指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考