Elasticsearch使用及踩坑

from elasticsearch import Elasticsearch, helpers
from elasticsearch.exceptions import NotFoundError
import abc
import os
from config.conf import Config
import logging
import contextlib
import fcntl


class ElasticsearchClient(object):
    def __new__(cls):
        if not hasattr(cls, 'instance'):
            cls.instance = super(ElasticsearchClient, cls).__new__(cls)
        return cls.instance

    def __init__(self):
        es_uri = os.environ.get('ELASTICSEARCH_URI')
        self.es = Elasticsearch(hosts=es_uri.split(','))

    def get_client(self):
        return self.es


class SearchBase(metaclass=abc.ABCMeta):
    alias = None
    build_lock_file = None

    def __init__(self, _type='default'):
        self.es = ElasticsearchClient().get_client()
        self._type = _type
        if self.alias is None:
            self.alias = self.index_prefix
        if not self.build_lock_file:
            self.build_lock_file = os.path.join(Config.LOGFILE_PATH, f'{self.index_prefix}.lock')

    @property
    @abc.abstractmethod
    def index_prefix(self) -> str:
        """索引名前缀"""

    @property
    @abc.abstractmethod
    def mappings(self) -> dict:
        """
        mappings, eg.
        {
            'default': {
                "properties": {
                    "test": {
                        "type": "keyword"
                    }
                }
            }
        }
        """

    @abc.abstractmethod
    def build_index(self, index):
        """建立索引"""

    @contextlib.contextmanager
    def lock_build(self):
        f = open(self.build_lock_file, 'a')
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            yield False
            f.close()
            logger.warning(f'{self.index_prefix} build is locked')
        else:
            yield True
            f.close()
            os.remove(self.build_lock_file)

    def rebuild_index(self):
        with self.lock_build() as flag:
            if flag is False:
                logger.warning('this index already locked! maybe other thread is doing this')
                return
            new_index = self.create_index(no_alias=True)
            self.build_index(new_index)
            indices = self.es.indices.get(self.index_prefix + '*')
            old_indices = [index for index in indices.keys() if index != new_index]
            remove_actions = [{"remove": {"index": index, "alias": self.alias}} for index in old_indices]
            self.es.indices.update_aliases({
                "actions": [
                    {"add": {"index": new_index, "alias": self.alias}},
                    *remove_actions
                ]
            })
            for index in old_indices:
                self.es.indices.delete(index)

    def create_index(self, no_alias=False):
        import random
        import string
        body = {
            "mappings": self.mappings,
        }
        if not no_alias:
            body['aliases'] = {
                self.alias: {}
            }

        if not no_alias and self.es.indices.exists_alias(self.alias):
            raise Exception(f'alias [{self.alias}] already exists')

        for _ in range(3):
            index = self.index_prefix + '-' + ''.join(random.sample(string.ascii_lowercase + string.digits, 4))
            if not self.es.indices.exists(index):
                self.es.indices.create(index, body=body)
                return index
        else:
            raise Exception('create index fail')

    def delete_index(self):
        try:
            self.es.indices.delete(self.index_prefix)
        except NotFoundError:
            pass
        except Exception as e:
            raise e

    def create(self, _id, body, **kwargs):
        return self.es.create(self.alias, _id, body, doc_type=self._type, **kwargs)

    def get(self, _id, **kwargs):
        try:
            return self.es.get(self.alias, _id, doc_type=self._type, **kwargs)
        except:
            return None

    def update(self, _id, body, **kwargs):
        return self.es.update(self.alias, _id, body, doc_type=self._type, **kwargs)

    def delete(self, _id, body, **kwargs):
        return self.es.delete(self.alias, _id, body, doc_type=self._type, **kwargs)

    def bulk(self, body, **kwargs):
        return self.es.bulk(body, index=self.alias, doc_type=self._type, **kwargs)

    def count(self, body=None, **kwargs):
        return self.es.count(body=body, index=self.alias, doc_type=self._type, **kwargs)

    def search(self, body=None, **kwargs):
        return self.es.search(body=body, index=self.alias, doc_type=self._type, **kwargs)

    def bulk_action(self, actions):
        return helpers.bulk(self.es, actions, request_timeout=120)