StashCache/tools.py at master · OSGConnect/StashCache · 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
import time

from elasticsearch import Elasticsearch, exceptions as es_exceptions
from elasticsearch import helpers

def get_es_connection():
    """
    establishes es connection.
    """
    print("make sure we are connected right...")
    try:
        es_conn = Elasticsearch([{'host': 'cl-analytics.mwt2.org', 'port': 9200}])
        print("connected OK!")
    except es_exceptions.ConnectionError as error:
        print('ConnectionError in GetESConnection: ', error)
    except:
        print('Something seriously wrong happened.')
    else:
        return es_conn

    time.sleep(70)
    get_es_connection()


def bulk_index(data, es_conn=None, thread_name=''):
    """
    sends the data to ES for indexing.
    if successful returns True.
    """
    success = False
    if es_conn is None:
        es_conn = get_es_connection()
    try:
        res = helpers.bulk(es_conn, data, raise_on_exception=True, request_timeout=60)
        print(thread_name, "inserted:", res[0], 'errors:', res[1])
        success = True
    except es_exceptions.ConnectionError as error:
        print('ConnectionError ', error)
    except es_exceptions.TransportError as error:
        print('TransportError ', error)
    except helpers.BulkIndexError as error:
        print(error[0])
      # for i in error[1]:
      # print(i)
    except:
        print('Something seriously wrong happened.')
    return success