Search code examples
elasticsearchelastalert

Relax elastalert on Sunday morning, due to false positive cases


I have a rule in elastalert that send a notification if there are no payments in two hours. I also have a match enhancements that drops these notifications every night from 0:00 to 8:00 AM:

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()

But now I want to add also a "relaxation" for the Sunday morning (when people mostly sleeps) and raise a DropMatchException from 0:00 to 10:00 AM. How can I do this?


Solution

  • Solution is this:

    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()