elastalert.py 文件源码

def find_pending_aggregate_alert(self, rule, aggregation_key_value=None):
        query = {'filter': {'bool': {'must': [{'term': {'rule_name': rule['name']}},
                                              {'range': {'alert_time': {'gt': ts_now()}}},
                                              {'not': {'exists': {'field': 'aggregate_id'}}},
                                              {'term': {'alert_sent': 'false'}}]}},
                 'sort': {'alert_time': {'order': 'desc'}}}
        if aggregation_key_value:
            query['filter']['bool']['must'].append({'term': {'aggregate_key': aggregation_key_value}})

        if not self.writeback_es:
            self.writeback_es = elasticsearch_client(self.conf)
        try:
            res = self.writeback_es.search(index=self.writeback_index,
                                           doc_type='elastalert',
                                           body=query,
                                           size=1)
            if len(res['hits']['hits']) == 0:
                return None
        except (KeyError, ElasticsearchException) as e:
            self.handle_error("Error searching for pending aggregated matches: %s" % (e), {'rule_name': rule['name']})
            return None

        return res['hits']['hits'][0]