def get_alarmed_messages()

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

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

    def get_alarmed_messages(self):
        """Returns all previous messages 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="rtops-*")
        if res is None:
            alarmed_hits = []
        else:
            alarmed_hits = res["hits"]["hits"]

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

        return messages