def get_alarmed_ips()

in elkserver/docker/redelk-base/redelkinstalldata/scripts/modules/alarm_httptraffic/module.py [0:0]

• 25 lines of code
• 4 McCabe index (conditional complexity)

    def get_alarmed_ips(self):  # pylint: disable=no-self-use
        """Returns all previous IPs that have been alarmed already"""
        es_query = {
            "sort": [{"@timestamp": {"order": "desc"}}],
            "query": {
                "bool": {
                    "filter": [
                        {"range": {"@timestamp": {"gte": "now-1y"}}},
                        {"match": {"tags": info["submodule"]}},
                    ]
                }
            },
        }
        res = raw_search(es_query, index="redirtraffic-*")
        if res is None:
            alarmed_hits = []
        else:
            alarmed_hits = res["hits"]["hits"]

        # Created a dict grouped by IP address (from source.ip)
        ips = {}
        for alarmed_hit in alarmed_hits:
            # pylint: disable=invalid-name
            ip = get_value("_source.source.ip", alarmed_hit)
            if ip in ips:
                ips[ip].append(alarmed_hit)
            else:
                ips[ip] = [alarmed_hit]

        return ips