event_processor/esClient.py at master · xOpsTech/event_processor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import logging
from elasticsearch import Elasticsearch, NotFoundError

ELASTICSEARCH_HOST = 'localhost'
ELASTICSEARCH_PORT = 9200


class EsClient(object):
    def __init__(self, host=ELASTICSEARCH_HOST):
        self.es = Elasticsearch(host.split(','), timeout=20)
        self.logger = logging.getLogger(__name__)

    def read_index_data(self, index, doc_type, id):
        try:
            res = self.es.get(index=index, doc_type=doc_type, id=id)
            self.logger.info('action: query elasticsearch | index: %s | id: %s | status: successful', index, id)
            return res['_source']
        except NotFoundError, e:
            self.logger.error(
                'action: query elasticsearch | index: %s | id: %s | status: unsuccessful | reason: index not found',
                index, id)
            self.logger.error(e)
        except Exception, e:
            self.logger.error('action: query elasticsearch | index: %s | id: %s | status: unsuccessful', index, id)
            self.logger.exception(e)

    def create_index_data(self, index, doc_type, id, body):
        try:
            res = self.es.index(index=index, doc_type=doc_type, id=id, body=body)
            self.logger.info('action: write to elasticsearch | index: %s | id: %s | status: successful', index, id)
            print 'indexing doc', id
            return res['created']
        except Exception, e:
            self.logger.error('action: write to elasticsearch | index: %s | id: %s | status: unsuccessful', index, id)
            self.logger.exception(e)

    def search_index_data(self, index, query):
        try:
            res = self.es.search(index=index, body=query)
            self.logger.info('action: search elasticsearch | index: %s | query: %s | hits: %d | status: successful',
                             index, query, res['hits']['total'])
            return res['hits']['hits']
        except Exception, e:
            self.logger.error('action: search elasticsearch | index: %s | query: %s | status: unsuccessful', index,
                              query)
            self.logger.exception(e)

    def delete_index_data(self, index, doc_type, id):
        try:
            self.es.delete(index=index, doc_type=doc_type, id=id)
            self.logger.info('action: delete elasticsearch | index: %s | id: %s | status: successful', index, id)
        except Exception, e:
            self.logger.error('action: delete elasticsearch | index: %s | id: %s | status: unsuccessful', index, id)
            self.logger.exception(e)

    def update_index_data(self, index, doc_type, id, body):
        try:
            self.es.update(index=index, doc_type=doc_type, id=id, body=body)
            self.logger.info('action: update elasticsearch | index: %s | id: %s | status: successful', index, id)
            print 'updated doc', id
            return True
        except Exception, e:
            self.logger.error('action: update elasticsearch | index: %s | id: %s | status: unsuccessful', index, id)
            self.logger.exception(e)

    def get_alert_by_event_id(self, index, doc_type, doc_id):
        return self.read_index_data(index=index, doc_type=doc_type, id=doc_id)

    def update_event(self, index, doc_type, body, alert_from_db):
        self.update_index_data(
            index=index, doc_type=doc_type, id=body['eventId'], body={
                "doc": {
                    "timestampUpdated": body['timestampUpdated'],
                    "severity": body['severity'],
                    "title": body['title'],
                    "trigger": body['trigger'],
                    "sourceEventsCount": int(alert_from_db.get('sourceEventsCount', 0)) + 1
                }
            })