[大数据] 物联网设备数据流转之告警信息推送：TDengine-alert

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> 物联网设备数据流转之告警信息推送：TDengine-alert -> 正文阅读

[大数据]物联网设备数据流转之告警信息推送：TDengine-alert

背景

前面关于 TDengine 的使用，我们只涉及到 Server 端与 Client 端，除此之外，官方还有一个报警模块，用以根据用户定义的规则实现近实时的报警监测。

从开始到现在，我们一直在喊口号： 人至践则无敌 ，事实上，在这个小项目中任何一个功能都对应一个实际真实存在的需求。既然是监测物联网设备的数据，那么我们的目的肯定是要看看设备的数据是不是有异常，这就对应于这样一个需求：

我们实际项目中通常要对某项指标设置一个阈值，超过阈值要进行告警推送。

官方文档的具体实践是这样：结合 Prometheus 的 AlertManager 实现报警推送；
这篇文章里就不涉及Prometheus 的 AlertManager了，直接将 TDengine 产生的报警信息推送到我们自己编写的 SpringBoot 的 Web 服务回调，并完成 WebSocket 推送、告警邮件推送。

Note: 整理代码时对 TDengine 进行过一次升级， TDengine （2.2.0.0–>2.4.0.16），发现从官网上找不到 TDengine-alert 相关的信息了，而且 GitHub 源码仓库关于 alert 的代码也半年多没更新了，不了解是什么情况。先用旧版的试试，竟然可以~我猜测应该是这样的： TDengine-alert 是相对独立的一个模块，再加上其将告警信息直接转发至 Prometheus 的 AlertManager 上， TDengine-alert 的功能应该已经没什么可更新的了。。

场景

在一系列的监测电压、电流、温度的时序数据中，当近3分钟之内的平均温度超过50℃时，进行告警。

Note：虽然使用平均值，但这并不是一个好的实践，这里仅以平均值作为一个示例。

一个古老的笑话：一位统计学家跳进平均深度只有25厘米的湖中，然后差点被淹死……

TDengine的报警模块

下载安装

# 下载
[root@iot1 ~]# cd /usr/local/
[root@iot1 local]# wget https://www.taosdata.com/assets-download/TDengine-alert-2.2.0.2-Linux-x64.tar.gz

# 解压安装
[root@iot1 local]# tar -xvf TDengine-alert-2.2.0.2-Linux-x64.tar.gz

配置文件

[root@iot1 local]# cd TDengine-alert

# 养成好习惯，备份配置文件
[root@iot1 TDengine-alert]# cp alert.cfg alert.cfg.bk

# 查看配置文件的默认内容
[root@iot1 TDengine-alert]# cat alert.cfg
{
    "port": 8100,
    "database": "file:alert.db",
    "tdengine": "root:taosdata@/tcp(127.0.0.1:0)/",
    "log": {
        "level": "debug",
        "path": ""
    },
    "receivers": {
        "alertManager": "http://127.0.0.1:9093/api/v1/alerts",
        "console": true
    }
}
# 配置选项的含义如下：
#     port：报警模块支持使用 restful API 对规则进行管理，这个参数用于配置 http 服务的侦听端口。
#     database：报警模块将规则保存到了一个 sqlite 数据库中，这个参数用于指定数据库文件的路径（不需要提前创建这个文件，如果它不存在，程序会自动创建它）。
#     tdengine：TDEngine 的连接信息，一般来说，数据库信息应该在报警规则中指定，所以这里 不 应包含这一部分信息。
#     log > level：日志的记录级别，可选 production 或 debug。
#     log > path：日志文件的路径。
#     receivers > alertManager：报警模块会将报警推送到 AlertManager，在这里指定 AlertManager 的接收地址。

# 可以启动啦，这里是前台启动，方便查看控制台的日志；生产环境可以用：nohup ./alert cfg alert.cfg &
[root@iot1 TDengine-alert]# ./alert -cfg alert.cfg

报警规则

上一步中虽然启动了报警模块，但是我们还没有配置报警规则，这个报警规则部分是需要我们自定义的。我这里新建了个 rules 目录，然后在其中编写 JSON 格式的报警规则，其实就是编写我们熟悉的 SQL 语句，在对应指标满足某种条件时触发的一个规则。

新建的报警规则文件的路径： /usr/local/TDengine-alert/rules/sensor_rule.json ，注意： json 文件中不可以有注释！

{
  "name": "temperatureTooHigh",
  "sql": "select avg(temperature) as avgTemperature from iot.power where ts > now - 3m group by sn",
  "expr": "avgTemperature > 50",
  "for": "0",
  "period": "20s",
  "labels": {
    "ruleName": "temperatureTooHigh"
  },
  "annotations": {
    "summary": "avg temperature of rule {{$labels.ruleName}} of device {{$values.sn}} is too high, its average temperature is {{$values.avgTemperature}} ℃"
  }
}

其中字段含义如下：

name：用于为规则指定一个唯一的名字。
sql：从 TDEngine 中查询数据时使用的 sql 语句，查询结果中的列将被后续计算使用，所以，如果使用了聚合函数，请为这一列指定一个别名。您可能已经注意到，本例中，这条语句和本文开头的那条完全相同。
expr：一个计算结果为布尔型的表达式，支持算数运算、逻辑运算，并且内置了部分函数，也可以引用查询结果中的列。 当表达式计算结果为 true 时，进入报警状态。
period：规则的检查周期，默认1分钟，而在我们的例子中，是每20秒检查一次有没有超过温度阈值。
for: 一个时间长度，当布尔表达式的计算结果为 true 的持续时间超过这个选项时，才会触发报警。默认为0，表示只要计算结果为 true，就触发报警。
labels：人为指定的标签列表，标签可以在生成报警信息时引用。特别的，如果 sql 中包含 group by 子句，则所有用于分组的字段会被自动加入这个标签列表中。
annotations：用于定义报警信息，使用 go template 语法，其中，可以通过 $labels.<label name> 引用标签，也可以通过 $values.<column name> 引用查询结果中的列。

将上述的规则翻译成人话：一旦监测到近3分钟之内的平均温度超过50℃时，就进行告警，每隔20s检查一次；触发规则后，使用summary中的内容组合前面定义的变量为一条完整的报警信息。

# 添加或修改规则
[root@iot1 rules]# curl -d '@sensor_rule.json' http://localhost:8100/api/update-rule

# 查看已有规则
[root@iot1 rules]# curl http://localhost:8100/api/list-rule
[{"name":"temperatureTooHigh","state":0,"sql":"select avg(temperature) as avgTemperature from iot.power where ts \u003e now - 3m group by sn","for":"0s","period":"20s","expr":"avgTemperature \u003e 50","labels":{"ruleName":"temperatureTooHigh"},"annotations":{"summary":"avg temperature of rule {{$labels.ruleName}} of device {{$values.sn}} is too high, its average temperature is {{$values.avgTemperature}} ℃"}}]

以上， TDengine 的报警模块便成功运行了，一旦有符合规则的数据，则会触发报警，报警会推送到设置的 receivers.alertManager URL上。现在我们还需要这个 Web 服务

自定义服务

下面，我们将抛弃 Prometheus 的黄金搭档 AlertManager 。让 TDengine 生成的报警信息不经过 AlertManager 而直接推送到我们定义的 SpringBoot 项目的 Web 回调地址上。

告警实体

新建 com.heartsuit.alert 包，里面再创建 AlertMessage 类。

AlertMessage.java

@Data
public class AlertMessage {
    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS", timezone = "GMT+8")
    private Date startsAt;

//    时间字符串：0001-01-01T00:00:00Z，如何解析？？
//    @JsonFormat(pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS", timezone = "GMT+8")
//    private Timestamp endsAt;

    private Map<String, Object> values;
    private Map<String, String> labels;
    private Map<String, String> annotations;
}

告警实体

在 com.heartsuit.common.event 包，里面再创建 AlertEvent 类（继承了 Spring 的 ApplicationEvent 类）。

AlertEvent.java

@Getter
public class AlertEvent extends ApplicationEvent {
    private AlertMessage alertMessage;

    /**
     * Create a new {@code ApplicationEvent}.
     *
     * @param source the object on which the event initially occurred or with
     *               which the event is associated (never {@code null})
     */
    public AlertEvent(Object source) {
        super(source);
    }

    public AlertEvent(Object source, AlertMessage alertMessage) {
        super(source);
        this.alertMessage = alertMessage;
    }
}

WebHook转发

在 com.heartsuit.alert 包下创建 AlertWebhook 类。后面 TDengine-alert 配置文件中的 receivers.alertManager URL就配置为我们这个回调地址： http://10.1.17.109:8000/alert/webhook 。

AlertWebhook.java

@RestController
@Slf4j
@RequestMapping("/alert")
public class AlertWebhook {
    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @PostMapping("/webhook")
    public String webhook(@RequestBody List<AlertMessage> req) {
        log.info(req.toString());

        if (req.size() > 0) {
            for (AlertMessage msg : req) {
                AlertEvent alertEvent = new AlertEvent(this, msg);
                applicationEventPublisher.publishEvent(alertEvent);
            }
        }
        return req.toString();
    }
}

事件监听与消费

AlertEventListener4Alert.java

@Component
@Slf4j
public class AlertEventListener4Alert {
    private static final String STANDARD_FORMAT = "yyyy-MM-dd HH:mm:ss";

    @EventListener
    public void handleEvent(AlertEvent event) throws JsonProcessingException {
        log.info("Alert Message received in WebSocket: {}", event.getAlertMessage());

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setTimeZone(TimeZone.getTimeZone("GMT+8"));
        objectMapper.setDateFormat(new SimpleDateFormat(STANDARD_FORMAT));
        String jsonAlert = objectMapper.writeValueAsString(event.getAlertMessage());
        WebSocketSession.sendMessage2All(jsonAlert);
    }
}

修改配置

将 TDengine-alert 的配置文件 alert.cfg 中的 receivers.alertManager 改为我们服务的地址即可：

完成以上操作后，项目结构如下：

模拟报警数据

在 taos 客户端执行插入语句前，我们先打开这几个服务窗口，方便观察效果：

前台启动TDengine-alert，用以观察日志；
我们的后端服务，观察控制台收到的告警信息；
我们的前端服务，看是不是可以正常弹出告警信息框；
一个taos客户端，用以插入测试数据；

# 在TDengine中模拟数据写入：近3分钟的五条数据，显然，每条数据的温度指标都高于我们设定的阈值：50℃
taos> insert into device0 values(now - 2m, 220, 5, 50) (now - 90s, 220, 5, 60) (now - 1m, 220, 5, 70) (now - 30s, 220, 5, 80) (now, 220, 5, 90);

最后，在我们的后端服务拿到警情信息后，还可以做进一步的处理，比如我这里就进行了：

WebSocket 服务端推送（注意：WebSocket前端代码不小心在之前实时设备数据推送时已经复制进去了。。）
邮件推送（并没有在当前的项目中实现，虽然在实际中告警信息邮件推送相对于WebSocket来说更实用，但是如果使用AlertManager，就可以直接进行邮件配置就可以了；而且自己发送邮件也比较简单，我之前也专门写过一篇文章：全栈开发之SpringBoot发送邮件，如果想集成到当前这个项目，那么可以自行实现。）