由于误报病例,周日早上放松elastalert
Relax elastalert on Sunday morning, due to false positive cases
我在 elastalert 中有一条规则,如果两小时内没有付款,我会发送通知。
我还有一个匹配增强功能,每晚从 0:00 到 8:00 上午删除这些通知:
from elastalert.enhancements import BaseEnhancement, DropMatchException
import datetime
import time
import sys
def datetime_from_utc_to_local(utc_datetime):
now_timestamp = time.time()
offset = datetime.datetime.fromtimestamp(now_timestamp) - datetime.datetime.utcfromtimestamp(now_timestamp)
return utc_datetime + offset
class DropFrom00To06(BaseEnhancement):
def process(self, match):
dateformat = "%Y-%m-%dT%H:%M:%S"
exceptional_dateformat = "%Y-%m-%dT%H:%M:%SZ"
timestamp = match['@timestamp'].split(".")[0]
try:
timestamp = datetime.datetime.strptime(timestamp, dateformat)
except ValueError:
timestamp = datetime.datetime.strptime(timestamp, exceptional_dateformat)
except:
print("Unexpected error:", sys.exc_info()[0])
raise
timestamp = datetime_from_utc_to_local(timestamp)
timePart = timestamp.time()
if timePart >= datetime.time(00, 00) and timePart <= datetime.time(8, 00):
raise DropMatchException()
但现在我还想为周日早上(人们大部分时间都在睡觉)添加一个 "relaxation",并从 0:00 到凌晨 10:00 引发 DropMatchException。
我该怎么做?
解决方法是这样的:
from elastalert.enhancements import BaseEnhancement, DropMatchException
import datetime
import time
import sys
def datetime_from_utc_to_local(utc_datetime):
now_timestamp = time.time()
offset = datetime.datetime.fromtimestamp(now_timestamp) - datetime.datetime.utcfromtimestamp(now_timestamp)
return utc_datetime + offset
class DropFrom00To06(BaseEnhancement):
def process(self, match):
dateformat = "%Y-%m-%dT%H:%M:%S"
exceptional_dateformat = "%Y-%m-%dT%H:%M:%SZ"
timestamp = match['@timestamp'].split(".")[0]
try:
timestamp = datetime.datetime.strptime(timestamp, dateformat)
except ValueError:
timestamp = datetime.datetime.strptime(timestamp, exceptional_dateformat)
except:
print("Unexpected error:", sys.exc_info()[0])
raise
timestamp = datetime_from_utc_to_local(timestamp)
timePart = timestamp.time()
d = timestamp.date()
day = d.weekday()
elif day == 6 and timePart >= datetime.time(00, 00) and timePart <= datetime.time(10, 00):
raise DropMatchException()
elif timePart >= datetime.time(00, 00) and timePart <= datetime.time(8, 00):
raise DropMatchException()
我在 elastalert 中有一条规则,如果两小时内没有付款,我会发送通知。 我还有一个匹配增强功能,每晚从 0:00 到 8:00 上午删除这些通知:
from elastalert.enhancements import BaseEnhancement, DropMatchException
import datetime
import time
import sys
def datetime_from_utc_to_local(utc_datetime):
now_timestamp = time.time()
offset = datetime.datetime.fromtimestamp(now_timestamp) - datetime.datetime.utcfromtimestamp(now_timestamp)
return utc_datetime + offset
class DropFrom00To06(BaseEnhancement):
def process(self, match):
dateformat = "%Y-%m-%dT%H:%M:%S"
exceptional_dateformat = "%Y-%m-%dT%H:%M:%SZ"
timestamp = match['@timestamp'].split(".")[0]
try:
timestamp = datetime.datetime.strptime(timestamp, dateformat)
except ValueError:
timestamp = datetime.datetime.strptime(timestamp, exceptional_dateformat)
except:
print("Unexpected error:", sys.exc_info()[0])
raise
timestamp = datetime_from_utc_to_local(timestamp)
timePart = timestamp.time()
if timePart >= datetime.time(00, 00) and timePart <= datetime.time(8, 00):
raise DropMatchException()
但现在我还想为周日早上(人们大部分时间都在睡觉)添加一个 "relaxation",并从 0:00 到凌晨 10:00 引发 DropMatchException。 我该怎么做?
解决方法是这样的:
from elastalert.enhancements import BaseEnhancement, DropMatchException
import datetime
import time
import sys
def datetime_from_utc_to_local(utc_datetime):
now_timestamp = time.time()
offset = datetime.datetime.fromtimestamp(now_timestamp) - datetime.datetime.utcfromtimestamp(now_timestamp)
return utc_datetime + offset
class DropFrom00To06(BaseEnhancement):
def process(self, match):
dateformat = "%Y-%m-%dT%H:%M:%S"
exceptional_dateformat = "%Y-%m-%dT%H:%M:%SZ"
timestamp = match['@timestamp'].split(".")[0]
try:
timestamp = datetime.datetime.strptime(timestamp, dateformat)
except ValueError:
timestamp = datetime.datetime.strptime(timestamp, exceptional_dateformat)
except:
print("Unexpected error:", sys.exc_info()[0])
raise
timestamp = datetime_from_utc_to_local(timestamp)
timePart = timestamp.time()
d = timestamp.date()
day = d.weekday()
elif day == 6 and timePart >= datetime.time(00, 00) and timePart <= datetime.time(10, 00):
raise DropMatchException()
elif timePart >= datetime.time(00, 00) and timePart <= datetime.time(8, 00):
raise DropMatchException()