Python中的日志解析:在给定的时间段内出现超过N个错误时发出警报?

2024-03-29 14:01:47 发布

您现在位置:Python中文网/ 问答频道 /正文

我有一个日志文件,里面有一小时的数据。你知道吗

如果在该日志的任何5分钟内记录的错误超过N个,我将生成警报。你知道吗

我不想做的是指定5分钟的时间段(例如00-05、06-10等)并对它们进行迭代,因为如果N=10,我在04处有8个错误,在07处有8个错误,它们将被视为在两个单独的bucket中,并且不会生成警报。你知道吗

我想我可以迭代60次,每次前进1分钟,从那一点开始考虑5分钟的bucket,但是有没有更优雅或更高效的方法呢?你知道吗


Tags: 文件数据方法bucket错误记录警报小时
2条回答

我决定采纳depperm's comment中的建议(我希望它是作为答案而不是评论提交的,这样我就可以将它标记为已接受)。你知道吗

它看起来像这样:

error_queue = []
max_errors = 3
for log_line in log_lines:
    log_ts = get_timestamp(log_line)
    if contains_error(log_line):
        error_queue.append(log_ts)
        interval_start = log_ts - datetime.timedelta(minutes=5)
        try:
            threshold = error_queue[-max_errors]
        except IndexError:
            continue
        if threshold and threshold >= interval_start:
            raise Exception

我将在错误列表上使用一个滑动窗口(请参见Rolling or sliding window iterator in Python),然后检查每个迭代是否第一个和最后一个条目在5分钟内

示例(来自Rolling or sliding window iterator in Python):

from collections import deque

def window(seq, n=2):
    it = iter(seq)
    win = deque((next(it, None) for _ in xrange(n)), maxlen=n)
    yield win
    append = win.append
    for e in it:
        append(e)
        yield win

for w in window(errors, 10):
    # if (w[-1]['timestamp'] - w[0]['timestamp']) > 60*5:
    #     error

相关问题 更多 >