# -*- coding: utf-8 -*- # Enable absolute import so we can import the elasticsearch client library from __future__ import absolute_import import os import hashlib import json import time import sys import traceback from elasticsearch import Elasticsearch, NotFoundError as ElasticsearchNotFoundError from elasticsearch.helpers import bulk, reindex from distribution import app from distribution.utils import get_worst_status class ElasticBackend(object): indices = { app.config.get('ELASTICSEARCH_FEED_INDEX_NAME', 'distribution_feed'): { 'filename': 'managers/resources/elastic/index_feed.json', 'real_index_name': None }, app.config.get('ELASTICSEARCH_TAXONOMY_INDEX_NAME', 'distribution_taxonomy'): { 'filename': 'managers/resources/elastic/index_taxonomy.json', 'real_index_name': None }, app.config.get('ELASTICSEARCH_AUTH_INDEX_NAME', 'distribution_auth'): { 'filename': 'managers/resources/elastic/index_apiauth.json', 'real_index_name': None }, app.config.get('ELASTICSEARCH_SITEMAP_INDEX_NAME', 'distribution_sitemap'): { 'filename': 'managers/resources/elastic/index_sitemap.json', 'real_index_name': None } } @property def client(self): if self._client is None: self.init() return self._client def __init__(self): self._client = None def init(self): app.logger.info('Initializing Elasticsearch manager') self._client = Elasticsearch(app.config['ELASTICSEARCH_HOSTS'], **app.config.get('ELASTICSEARCH_CONFIG', {})) self._init_indices() def _get_index_filename(self, index): return self.indices[index]['filename'] def _init_indices(self): for key in self.indices.keys(): self._init_index(key) def _set_real_index_name(self, index, index_real_name): self.indices[index]['real_index_name'] = index_real_name def _init_index(self, index): """ Initializes the Elasticsearch index. Distribution is going to create an index for each version of the settings/mappings specified in the managers/resources/elastic/index.json file. Then we specify an alias pointing to the most up-to-date index When we initially create the index, we also create an alias for it that we are going to use for indexing and querying. The index name is generated using the format "distribution_feed_", where "" is the sha256 digest for the index.json content, for example: distribution_feed_47re894v11nd74987r8142x1v2c3xz1weq87r8. Then the alias named "distribution_feed" will point to this index. At client initialization time, we look for the existence of the index whose name matches the current settings digest (prepend with the prefix), if we cannot find such index we can conclude that index.json file changed since the latest index creation. In this case, we create a new index with the updated settings and mappings, re-index the data and adjust the alias to point the new index. """ index_filename = self._get_index_filename(index) with open(os.path.join(app.config['APP_ROOT'], index_filename), 'r') as index_file: index_file_content = index_file.read() index_settings_hash = hashlib.sha256(index_file_content).hexdigest() real_index_name = '{}_{}'.format(index, index_settings_hash) self._set_real_index_name(index, '{}_{}'.format(index, index_settings_hash)) if not self.client.indices.exists(real_index_name): app.logger.info("Index for settings with hash '{}' doesn't exists, creating" .format(index_settings_hash)) self.client.indices.create(real_index_name, json.loads(index_file_content)) self._wait_for_ready_cluster(real_index_name) if self.client.indices.exists_alias(name=index): app.logger.info("Reindexing all data from old index") # Re-index data from old index to the new index # # More information: # https://www.elastic.co/guide/en/elasticsearch/guide/current/reindex.html # https://www.elastic.co/guide/en/elasticsearch/guide/current/scan-scroll.html reindex(self.client, index, real_index_name) try: # Delete alias and all associated indices. Also ensures # that no index named "distribution_feed" exists before # creating the alias self.client.indices.delete(index) except ElasticsearchNotFoundError: pass if not self.client.indices.exists(index): app.logger.info("Creating index alias") self.client.indices.put_alias(real_index_name, index) current_pointed_indices = self.client.indices.get_alias(index).keys() if current_pointed_indices != [unicode(real_index_name)]: msg = ('The {} alias is pointing to wrong indices: {}. It must point only to {}, p' 'lease check the cluster status') msg = msg.format(index, ', '.join(current_pointed_indices), real_index_name) app.logger.warning(msg) app.logger.info("Using {} index".format(real_index_name)) def _wait_for_ready_cluster(self, index_name): while self.client.cluster.health(index_name)['status'] == 'red': time.sleep(0.5) def clear(self): # pragma: no cover for index in self.indices.keys(): self.clear_index(index) def clear_index(self, index): # pragma: no cover try: self.client.indices.delete(index) except ElasticsearchNotFoundError: pass self._init_index(index) def _get_indices_names(self): return self.indices.keys() def _get_real_indices_names(self): real_indices_names = [] for value in self.indices.values(): real_indices_names.append(value['real_index_name']) return real_indices_names def _get_alias_status(self, index): current_pointed_indices = self.client.indices.get_alias(index).keys() real_index_name = self.indices[index]['real_index_name'] health = { 'status': 'green', 'pointed_indices': current_pointed_indices, 'message': None } if current_pointed_indices != [real_index_name]: health['status'] = 'red' message = 'The {} alias is pointing to wrong indices. It must point only to {}'.format( index, real_index_name) health['message'] = message return health def health(self): health = { 'class': self.__class__.__module__ + '.' + self.__class__.__name__, 'status': 'green', 'error': {}, 'current_index_name': self._get_real_indices_names(), 'alias': {}, 'alias_health': {}, 'current_index_health': {} } for index, index_data in self.indices.items(): health['alias'][index] = {} health['alias_health'][index] = {} health['current_index_health'][index] = {} try: health['alias'][index] = self._get_alias_status(index) health['alias_health'][index] = self.client.cluster.health(index) health['current_index_health'][index] = self.client.cluster.health( index_data['real_index_name']) except Exception as exc: health['error'][index] = { 'message': str(exc), 'stacktrace': traceback.extract_tb(sys.exc_info()[2]) } health['alias'][index].setdefault('status', 'red') health['alias_health'][index].setdefault('status', 'red') health['current_index_health'][index].setdefault('status', 'red') try: health['cluster_health'] = self.client.cluster.health() except Exception as exc: health['error']['_cluster'] = { 'message': str(exc), 'stacktrace': traceback.extract_tb(sys.exc_info()[2]) } health.setdefault('cluster_health', {'status': 'red'}) statuses = [health['cluster_health']['status']] for alias in health['alias'].values(): statuses.append(alias['status']) health['status'] = get_worst_status(statuses) return health def bulk(self, actions): """ Runs all given actions and log if errors are found :return: (len(correct_actions), errors) """ result = bulk(self.client, actions, raise_on_error=False) if len(result[1]) > 0: app.logger.warning( '{} of {} actions failed during a bulk operation: {}' .format(len(result[1]), len(actions), result[1])) return result