【Azure Event Hub】自定义告警(Alert Rule)用来提示Event Hub的消息incoming(生产)与outgoing(消费)的异常情况

问题描述

在使用Azure Service Bus的时候，我们可以根据Queue中目前存在的消息数来判断当前消息是否有积压的情况。

但是，在Event Hub中，因为所有消息都会被存留到预先设定的保留时间(默认是7天)， 所以无法通过消息数来判断当前的消息是否有积压或者是有多余重复消费。

当消费端出现异常情况，在没有incoming的情况下，还是存在大量的outgoing，这种情况如何来提前预警呢？是否可以通过设置告警规则来及时通知运维人员呢？

如下图这样Outgoing 大于 incoming的情况

 

 

问题解答

在Azure服务中，PaaS资源都可以根据指标数据来设置告警，但是这是基于该服务有这类型指标的情况。

 

如现在Event Hub中 Outgoing Message 和Incoming Message不匹配的情况，由于Event Hub自身的指标中，并没有一个指标表示 Outgoing 与 Incoming 之间的差距值。 如果需要知道两个指标间的差值，就需要从收集的Metrics指标中进行自定义查询语句，并根据结果进行阈值设定。

 

主要的步骤有：

1）收集Event Hub Metrics指标到Log A中

• 设置诊断日志，日志发送到Log Analytics Workspace中

详细步骤，可参考官网：https://docs.azure.cn/zh-cn/event-hubs/monitor-event-hubs#log-analytics

 

2）自定义查询语句，在Kusto中进行行列转换，绝对值转换

示例Kusto语句

AzureMetrics 
| where ResourceProvider =="MICROSOFT.EVENTHUB" 
| where MetricName =="EHOUTMSGS"  
| project TimeGenerated,EHOUTMSGS=MetricName, OutMsgTotal=Total
| join kind = leftouter
(
    AzureMetrics 
    | where ResourceProvider =="MICROSOFT.EVENTHUB" 
    | where MetricName =="EHINMSGS" 
    | project TimeGenerated,EHINMSGS=MetricName, InMsgTotal=Total
) on  $left.TimeGenerated ==  $right.TimeGenerated
| project  TimeGenerated,EHINMSGS,EHOUTMSGS,OutMsgTotal,InMsgTotal
|extend OutMsgTotal=case(isnull(OutMsgTotal),0.0,OutMsgTotal), InMsgTotal=case(isnull(InMsgTotal),0.0,InMsgTotal) 
| project TimeGenerated, AlertValue = abs(OutMsgTotal - InMsgTotal)

说明：

1： EHOUTMSGS 表示Event Hub Outgoing Messages 统计值

2： EHINMSGS表示 Event Hub Incoming Messages 统计值

3： 使用 Left Outer Join的方式，把 EHOUTMSGS 和 ENINMSGS 的值组合在一行中，并且通过 EXTEND  CASE  补齐数据，缺少的行用0代替。

4：AlertValue = abs(EHOUTMSGS - EHINMSGS)， 计算差值并取绝对值，然结果显示为正

 

3）设置自定义指标

• 选择 Signal Name 为 Custom Log search
• 输入第二步中的Kusto Query 语句
• 在Measurement选择 AlertVaule， Average， 并以5分钟为一个时间周期进行判断

4）设置阈值

在Alert Logic 部分中，这是阈值即可。这里也是触发Alert的关键点，需要根据实际情况来决定一个合理的值。

 

 

【结束】

 

附录：为什么pivot 语句中特意的添加了 输出格式的结构体呢？

| evaluate pivot(MetricName,sum(Total)):(TimeGenerated:datetime, EHOUTMSGS:long, EHINMSGS:long,AlertValue:long)

添加OutputSchema的目的就为了避免查询结构体中缺少了需要统计的字段值而报错（'project' operator: Failed to resolve scalar expression named 'EHINMSGS' I）。

如下效果对比：

 

附录二： PIVOT 示例

pivot 函数执行把多行数据转换为多列。此处通过MetricsName的值的不同而转换为 EHOUTMSGS 和 EHINMSGS 两列，并且根据TimeGenerated时间值进行Total统计。在同一个时间点的数据归并为一行数据。

PIVOT使用的前后对比效果：