需求背景

以小时为周期不停地上报事件到事件平台,事件平台如果在连续2个周期 没有检测到上报的事件,就会发送告警给事件的相关责任人.

问题的难点在于如何检测连续周期内无数据?

如上图,2 点和 3 点,都没有上报数据,说明连续两个周期存在无数据上报.

解决方案

本文采用 redis 的有序集合实现此算法.

1. 假设

先做一些假设

• 事件需要有类型属性,每种类型的事件有唯一一个上报渠道 ID
  ,参数为 data_id,每个上报渠道只有一种事件类型.
• 上报的每个事件都有一个hash值,参数为 hash_value,相同 hash 值的所有事件当做一条时间序列.连续周期无数据检测针对的是一条时间序列进行处理.

2. 记录事件的hash 值到 redis 集合

事件平台每收到一个事件,就将事件的 hash 值hash_value 记录在 redis 的 set 中,示例代码如下

from django.core.cache import cache as redis_cachedata_id = "9fffdf1c-6feb-11ee-a73d-acde48001122"
hash_value = "7dcfd8f2-6feb-11ee-a73d-acde48001122"redis_key = data_id
redis_client = redis_cache.client.get_client()
redis_client.sadd(redis_key, hash_value)

3. 记录事件的上报时间到 redis 有序集合

事件平台每收到一个事件,就将上报时间戳作为 score,记录到 redis 的 sorted set 中,示例代码如下

from django.core.cache import cache as redis_cachedata_id = "9fffdf1c-6feb-11ee-a73d-acde48001122"
hash_value = "7dcfd8f2-6feb-11ee-a73d-acde48001122"# 假设事件的值为
event = {"hash_value": hash_value,"data_id": data_id,"data": {"key1": "value1","key2": "value2"}
}redis_key = f"{data_id}:{hash_value}"
score = int(time.time())redis_client = redis_cache.client.get_client()
redis_client.zadd(redis_key, {value: score})

使用有序集合,就可以构造出一个分数为时间戳的时间序列.

4. 检测无数据周期

需要开启一个定时任务,这里假设5 分钟检测一次时间序列.每次检测指定时间范围内事件的数量.

from datetime import timezone
from django.core.cache import cache as redis_cache
from django.utils import timezonedata_id = "9fffdf1c-6feb-11ee-a73d-acde48001122"
hash_value = "7dcfd8f2-6feb-11ee-a73d-acde48001122"# 检测周期 1 小时
interval_seconds = 3600
# 无数据周期数
cycle_number = 2
# 计算无数据检查时间范围
period_end_time = timezone.now()
period_start_time = period_end_time - timedelta(seconds=interval_seconds * cycle_number)
# 计算分数区间
period_start_score = int(period_start_time.timestamp())
period_end_score = int(period_end_time.timestamp())# 获得周期范围内的事件数量
redis_key = f"{data_id}:{hash_value}"
redis_client = redis_cache.client.get_client()
count = redis_client.zcount(redis_key, period_start_score, period_end_score)

如果count为 0,说明在连续周期内无数据.

5. 删除历史数据

由于是周期性事件上报,时间久了 set 和 sorted set 的值就会越来越大,导致 redis 服务器内存告警.所以我们需要定期清理过期的数据.

from django.core.cache import cache as redis_cachedata_id = "9fffdf1c-6feb-11ee-a73d-acde48001122"redis_client = redis_cache.client.get_client()# 获取所有的事件
redis_key = data_id
hash_value_set = redis_client.smembers(redis_key)# 指定删除的时间区间
min_score = ""
max_score = ""# 根据指定的时间取件删除有序集合中过期的数据
for hash_value in hash_value_set:redis_key = f"{data_id}:{hash_value}"redis_client.zremrangebyscore(redis_key, min_score, max_score)