Indexer daemon should retry when ES times out
@@ -10,6 +10,7 @@ from kombu.mixins import ConsumerMixin
from elasticsearch import Elasticsearch
from elasticsearch import helpers
+from elasticsearch.exceptions import ConnectionTimeout
from raven.contrib.django.raven_compat.models import client
@@ -25,6 +26,9 @@ class SearchIndexer(ConsumerMixin):
MAX_CHUNK_BYTES = 10 * 1024 ** 2 # 10 megs
+ TIMEOUT_INTERVAL = 10 # seconds
+ TIMEOUT_RETRIES = 10
+
def __init__(self, connection, index, url=None, max_size=5000, prefetch_count=7500):
self._model_queues = {}
self._pool = ThreadPoolExecutor(max_workers=len(settings.INDEXABLE_MODELS) + 1)
@@ -122,14 +126,19 @@ class SearchIndexer(ConsumerMixin):
try:
while not self.should_stop:
msgs = []
+ actions = self._actions(250, msgs)
+ tries = 0
+
+ while not self.should_stop:
stream = helpers.streaming_bulk(
self.es_client,
- self._actions(250, msgs),
+ actions,
max_chunk_bytes=self.MAX_CHUNK_BYTES,
raise_on_error=False,
)
start = time.time()
+ try:
for (ok, resp), msg in zip(stream, msgs):
if not ok and not (resp.get('delete') and resp['delete']['status'] == 404):
raise ValueError(ok, resp, msg)
@@ -141,7 +150,14 @@ class SearchIndexer(ConsumerMixin):
logger.info('%r: Indexed %d documents in %.02fs', self, len(msgs), time.time() - start)
else:
logger.debug('%r: Recieved no messages for %.02fs', self, time.time() - start)
- msgs.clear()
+ break
+ except ConnectionTimeout:
+ if tries >= self.TIMEOUT_RETRIES:
+ raise
+ tries += 1
+ logger.warning('Connection to elasticsearch timed out. Trying again after %s sec...', self.TIMEOUT_INTERVAL)
+ time.sleep(self.TIMEOUT_INTERVAL)
+ continue
except Exception as e:
client.captureException()
logger.exception('%r: _index_loop encountered an unexpected error', self)
