<commit_before>from copy import copy
from functools import wraps
import logging
from couchdbkit.exceptions import ResourceNotFound
from elasticsearch import Elasticsearch
from psycopg2._psycopg import InterfaceError
from datetime import datetime, timedelta
import hashlib
import traceback
import math
import time
from rawes.elastic_exception import ElasticException

from requests import ConnectionError
import simplejson
import rawes
import sys

from dimagi.utils.decorators.memoized import memoized
from dimagi.utils.couch import LockManager
from pillow_retry.models import PillowError
from pillowtop.checkpoints.manager import PillowCheckpoint
from pillowtop.checkpoints.util import get_machine_id, construct_checkpoint_doc_id_from_name
from pillowtop.const import CHECKPOINT_FREQUENCY
from pillowtop.couchdb import CachedCouchDB

from django import db
from pillowtop.dao.couch import CouchDocumentStore
from pillowtop.feed.couch import CouchChangeFeed
from pillowtop.logger import pillow_logging
from pillowtop.pillow.interface import PillowBase
try:
    from corehq.util.soft_assert import soft_assert
    _assert = soft_assert(to='@'.join(['czue', 'dimagi.com']), fail_if_debug=True)
except ImportError:
    # hack for dependency resolution if corehq not available
    _assert = lambda assertion, message: None


WAIT_HEARTBEAT = 10000
CHANGES_TIMEOUT = 60000
RETRY_INTERVAL = 2  # seconds, exponentially increasing
MAX_RETRIES = 4  # exponential factor threshold for alerts


INDEX_REINDEX_SETTINGS = {"index": {"refresh_interval": "900s",
                                    "merge.policy.merge_factor": 20,
                                    "store.throttle.max_bytes_per_sec": "1mb",
                                    "store.throttle.type": "merge",
                                    "number_of_replicas": "0"}
}
INDEX_STANDARD_SETTINGS = {"index": {"refresh_interval": "1s",
                                     "merge.policy.merge_factor": 10,
                                     "store.throttle.max_bytes_per_sec": "5mb",
                                     "store.throttle.type": "node",
                                     "number_of_replicas": "0"}
}


class PillowtopIndexingError(Exception):
    pass


class PillowtopNetworkError(Exception):
    pass


def ms_from_timedelta(td):
    """
    Given a timedelta object, returns a float representing milliseconds
    """
    return (td.seconds * 1000) + (td.microseconds / 1000.0)


def lock_manager(obj):
    if isinstance(obj, LockManager):
        return obj
    else:
        return LockManager(obj, None)


class BasicPillow(PillowBase):
    """
    BasicPillow is actually a CouchPillow. PillowBase defines the actual interface.
    """
    checkpoint_frequency = CHECKPOINT_FREQUENCY
    couch_filter = None  # string for filter if needed
    extra_args = {}  # filter args if needed
    document_class = None  # couchdbkit Document class
    _couch_db = None
    include_docs = True
    use_locking = False

    def __init__(self, couch_db=None, document_class=None, checkpoint=None, change_feed=None):
        if document_class:
            self.document_class = document_class

        self._couch_db = couch_db
        self._checkpoint = checkpoint
        self._change_feed = change_feed

        if self.use_locking:
            # document_class must be a CouchDocLockableMixIn
            assert hasattr(self.document_class, 'get_obj_lock_by_id')

    def get_couch_db(self):
        if self._couch_db is None:
            self._couch_db = self.get_default_couch_db()
        return self._couch_db

    def set_couch_db(self, couch_db):
        self._couch_db = couch_db

    def get_default_couch_db(self):
        return self.document_class.get_db() if self.document_class else None

    @property
    def couch_db(self):
        _assert(False, 'People should not be using the couch_db properties!')
        return self.get_couch_db()

    @couch_db.setter
    def couch_db(self, value):
        _assert(False, 'People should not be using the couch_db properties!')
        self._couch_db = value

    @property
    def document_store(self):
        return CouchDocumentStore(self.get_couch_db())

    @property
    def checkpoint(self):
        if self._checkpoint is None:
            self._checkpoint = self._get_default_checkpoint()
        return self._checkpoint

    def _get_default_checkpoint(self):
        return PillowCheckpoint(
            self.document_store,
            construct_checkpoint_doc_id_from_name(self.get_name()),
        )

    def get_change_feed(self):
        if self._change_feed is None:
            self._change_feed = self._get_default_change_feed()
        return self._change_feed

    def _get_default_change_feed(self):
        return CouchChangeFeed(
            couch_db=self.get_couch_db(),
            couch_filter=self.couch_filter,
            include_docs=self.include_docs,
            extra_couch_view_params=self.extra_args
        )

    @memoized
    def get_name(self):
        return self.get_legacy_name()

    @classmethod
    def get_legacy_name(cls):
        return "%s.%s.%s" % (cls._get_base_name(), cls.__name__, get_machine_id())

    @classmethod
    def _get_base_name(cls):
        return cls.__module__

    def processor(self, change, context):
        """
        Parent processsor for a pillow class - this should not be overridden.
        This workflow is made for the situation where 1 change yields 1 transport/transaction
        """
        self.process_change(change)

    def fire_change_processed_event(self, change, context):
        if context.changes_seen % self.checkpoint_frequency == 0 and context.do_set_checkpoint:
            self.set_checkpoint(change)

    def process_change(self, change, is_retry_attempt=False):
        try:
            with lock_manager(self.change_trigger(change)) as t:
                if t is not None:
                    tr = self.change_transform(t)
                    if tr is not None:
                        self.change_transport(tr)
        except Exception, ex:
            if not is_retry_attempt:
                self._handle_pillow_error(change, ex)
            else:
                raise

    def _handle_pillow_error(self, change, exception):
        try:
            # This breaks the module boundary by using a show function defined in commcare-hq
            # but it was decided that it wasn't worth the effort to maintain the separation.
            meta = self.get_couch_db().show('domain/domain_date', change['id'])
        except ResourceNotFound:
            # Show function does not exist
            meta = None
        error = PillowError.get_or_create(change, self, change_meta=meta)
        error.add_attempt(exception, sys.exc_info()[2])
        error.save()
        pillow_logging.exception(
            "[%s] Error on change: %s, %s. Logged as: %s" % (
                self.get_name(),
                change['id'],
                exception,
                error.id
            )
        )

    def change_trigger(self, changes_dict):
        """
        Step one of pillowtop process
        For a given _changes indicator, the changes dict (the id, _rev) is sent here.

        Note, a couch _changes line is: {'changes': [], 'id': 'guid',  'seq': <int>}
        a 'deleted': True might be there too

        whereas a doc_dict is _id
        Should return a doc_dict
        """
        if changes_dict.get('deleted', False):
            # override deleted behavior on consumers that deal with deletions
            return None
        id = changes_dict['id']
        if self.use_locking:
            lock = self.document_class.get_obj_lock_by_id(id)
            lock.acquire()
            return LockManager(self.get_couch_db().open_doc(id), lock)
        elif changes_dict.get('doc', None) is not None:
            return changes_dict['doc']
        else:
            return self.get_couch_db().open_doc(id)

    def change_transform(self, doc_dict):
        """
        Step two of the pillowtop processor:
        Process/transform doc_dict if needed - by default, return the doc_dict passed.
        """
        return doc_dict

    def change_transport(self, doc_dict):
        """
        Step three of the pillowtop processor:
        Finish transport of doc if needed. Your subclass should implement this
        """
        raise NotImplementedError(
            "Error, this pillowtop subclass has not been configured to do anything!")


PYTHONPILLOW_CHUNK_SIZE = 250
PYTHONPILLOW_CHECKPOINT_FREQUENCY = CHECKPOINT_FREQUENCY * 10
PYTHONPILLOW_MAX_WAIT_TIME = 60


class PythonPillow(BasicPillow):
    """
    A pillow that does filtering in python instead of couch.

    Useful because it will actually set checkpoints throughout even if there
    are no matched docs.

    In initial profiling this was also 2-3x faster than the couch-filtered
    version.

    Subclasses should override the python_filter function to perform python
    filtering.
    """
    process_deletions = False

    def __init__(self, document_class=None, chunk_size=PYTHONPILLOW_CHUNK_SIZE,
                 checkpoint_frequency=PYTHONPILLOW_CHECKPOINT_FREQUENCY,
                 couch_db=None, checkpoint=None, change_feed=None, preload_docs=True):
        """
        Use chunk_size = 0 to disable chunking
        """
        super(PythonPillow, self).__init__(
            document_class=document_class,
            checkpoint=checkpoint,
            couch_db=couch_db,
            change_feed=change_feed,
        )
        self.change_queue = []
        self.chunk_size = chunk_size
        self.use_chunking = chunk_size > 0
        self.checkpoint_frequency = checkpoint_frequency
        self.include_docs = not self.use_chunking
        self.last_processed_time = None
        self.preload_docs = preload_docs

    def get_default_couch_db(self):
        if self.document_class and self.use_chunking:
            return CachedCouchDB(self.document_class.get_db().uri, readonly=False)
        else:
            return super(PythonPillow, self).get_default_couch_db()

    def python_filter(self, change):
        """
        Should return True if the doc is to be processed by your pillow
        """
        return True

    def process_chunk(self):
        def _assert_change_has_id(change):
            if 'id' not in change:
                _assert(False, "expected 'id' in change, but wasn't found! change is: {}".format(
                    simplejson.dumps(change)
                ))
                return False
            return True

        changes_to_process = filter(_assert_change_has_id, self.change_queue)
        if self.preload_docs:
            self.get_couch_db().bulk_load([change['id'] for change in changes_to_process],
                                     purge_existing=True)
        for change in changes_to_process:
            if self.preload_docs:
                doc = self.get_couch_db().open_doc(change['id'], check_main=False)
                change.document = doc

            # a valid change is either a non-preload situation or a valid doc + a filter match
            valid_change = (not self.preload_docs or change.document) and self.python_filter(change)
            valid_deletion = self.process_deletions and change.get('deleted', None)
            if valid_change or valid_deletion:
                try:
                    self.process_change(change)
                except Exception:
                    logging.exception('something went wrong processing change %s (%s)' %
                                      (change.get('seq', None), change['id']))

        # reset the queue after we've processed this chunk
        self.change_queue = []
        self.last_processed_time = datetime.utcnow()

    @property
    def queue_full(self):
        return len(self.change_queue) > self.chunk_size

    @property
    def wait_expired(self):
        if not self.last_processed_time:
            return False

        wait_time = datetime.utcnow() - self.last_processed_time
        return wait_time > timedelta(seconds=PYTHONPILLOW_MAX_WAIT_TIME)

    def processor(self, change, context):
        if self.use_chunking:
            self.change_queue.append(change)
            if self.queue_full or self.wait_expired:
                self.process_chunk()
        elif self.python_filter(change) or (change.get('deleted', None) and self.process_deletions):
            self.process_change(change)
        if context.changes_seen % self.checkpoint_frequency == 0 and context.do_set_checkpoint:
            # if using chunking make sure we never allow the checkpoint to get in
            # front of the chunks
            if self.use_chunking:
                self.process_chunk()
            self.set_checkpoint(change)

    def fire_change_processed_event(self, change, context):
        # todo: should fix this so that the checkpointing happens here.
        pass

    def run(self):
        self.change_queue = []
        self.last_processed_time = datetime.utcnow()
        super(PythonPillow, self).run()


def send_to_elasticsearch(path, es_getter, name, data=None, retries=MAX_RETRIES,
        except_on_failure=False, update=False, delete=False):
    """
    More fault tolerant es.put method
    """
    data = data if data is not None else {}
    current_tries = 0
    while current_tries < retries:
        try:
            if delete:
                res = es_getter().delete(path=path)
            elif update:
                params = {'retry_on_conflict': 2}
                res = es_getter().post("%s/_update" % path, data={"doc": data}, params=params)
            else:
                res = es_getter().put(path, data=data)
            break
        except ConnectionError, ex:
            current_tries += 1
            pillow_logging.error("[%s] put_robust error %s attempt %d/%d" % (
                name, ex, current_tries, retries))
            time.sleep(math.pow(RETRY_INTERVAL, current_tries))

            if current_tries == retries:
                message = "[%s] Max retry error on %s" % (name, path)
                if except_on_failure:
                    raise PillowtopIndexingError(message)
                else:
                    pillow_logging.error(message)
                res = {}

    if res.get('status', 0) == 400:
        error_message = "Pillowtop put_robust error [%s]:\n%s\n\tpath: %s\n\t%s" % (
            name,
            res.get('error', "No error message"),
            path,
            data.keys())

        if except_on_failure:
            raise PillowtopIndexingError(error_message)
        else:
            pillow_logging.error(error_message)
    return res


class AliasedElasticPillow(BasicPillow):
    """
    This pillow class defines it as being alias-able. That is, when you query it, you use an
    Alias to access it.

    This could be for varying reasons -  to make an index by certain metadata into its own index
    for performance/separation reasons

    Or, for our initial use case, needing to version/update the index mappings on the fly with
    minimal disruption.


    The workflow for altering an AliasedElasticPillow is that if you know you made a change, the
    pillow will create a new Index with a new md5sum as its suffix. Once it's finished indexing,
    you will need to flip the alias over to it.
    """
    es_host = ""
    es_port = ""
    es_index = ""
    es_type = ""
    es_alias = ''
    es_meta = {}
    es_timeout = 3  # in seconds
    default_mapping = None  # the default elasticsearch mapping to use for this
    bulk = False
    online = True  # online=False is for in memory (no ES) connectivity for testing purposes or for admin operations

    # Note - we allow for for existence because we do not care - we want the ES
    # index to always have the latest version of the case based upon ALL changes done to it.
    allow_updates = True

    def __init__(self, create_index=True, online=True, **kwargs):
        """
        create_index if the index doesn't exist on the ES cluster
        """
        super(AliasedElasticPillow, self).__init__(**kwargs)
        # online=False is used in unit tests
        self.online = online
        index_exists = self.index_exists()
        if create_index and not index_exists:
            self.create_index()
        if self.online:
            pillow_logging.info("Pillowtop [%s] Initializing mapping in ES" % self.get_name())
            self.initialize_mapping_if_necessary()
        else:
            pillow_logging.info("Pillowtop [%s] Started with no mapping from server in memory testing mode" % self.get_name())

    def initialize_mapping_if_necessary(self):
        """
        Initializes the elasticsearch mapping for this pillow if it is not found.
        """
        mapping = self.get_index_mapping()
        if not mapping:
            pillow_logging.info("Initializing elasticsearch mapping for [%s]" % self.es_type)
            mapping = copy(self.default_mapping)
            mapping['_meta']['created'] = datetime.isoformat(datetime.utcnow())
            mapping_res = self.set_mapping(self.es_type, {self.es_type: mapping})
            if mapping_res.get('ok', False) and mapping_res.get('acknowledged', False):
                # API confirms OK, trust it.
                pillow_logging.info("Mapping set: [%s] %s" % (self.es_type, mapping_res))
        else:
            pillow_logging.info("Elasticsearch mapping for [%s] was already present." % self.es_type)

    def index_exists(self):
        if not self.online:
            # If offline, just say the index is there and proceed along
            return True

        es = self.get_es()
        try:
            res = es.head(self.es_index)
        except ElasticException as e:
            if e.status_code == 404:
                return False
            else:
                raise
        return res

    def get_doc_path(self, doc_id):
        return "%s/%s/%s" % (self.es_index, self.es_type, doc_id)

    def update_settings(self, settings_dict):
        return self.send_robust("%s/_settings" % self.es_index, data=settings_dict)

    def set_index_reindex_settings(self):
        """
        Set a more optimized setting setup for fast reindexing
        """
        return self.update_settings(INDEX_REINDEX_SETTINGS)

    def set_index_normal_settings(self):
        """
        Normal indexing configuration
        """
        return self.update_settings(INDEX_STANDARD_SETTINGS)

    def get_index_mapping(self):
        es = self.get_es()
        return es.get('%s/_mapping' % self.es_index).get(self.es_index, {})

    def set_mapping(self, type_string, mapping):
        if self.online:
            return self.send_robust("%s/%s/_mapping" % (self.es_index, type_string), data=mapping)
        else:<commit_msg>use new elasticsearch api for index_exists<commit_after>from copy import copy
from functools import wraps
import logging
from couchdbkit.exceptions import ResourceNotFound
from elasticsearch import Elasticsearch
from psycopg2._psycopg import InterfaceError
from datetime import datetime, timedelta
import hashlib
import traceback
import math
import time
from rawes.elastic_exception import ElasticException

from requests import ConnectionError
import simplejson
import rawes
import sys

from dimagi.utils.decorators.memoized import memoized
from dimagi.utils.couch import LockManager
from pillow_retry.models import PillowError
from pillowtop.checkpoints.manager import PillowCheckpoint
from pillowtop.checkpoints.util import get_machine_id, construct_checkpoint_doc_id_from_name
from pillowtop.const import CHECKPOINT_FREQUENCY
from pillowtop.couchdb import CachedCouchDB

from django import db
from pillowtop.dao.couch import CouchDocumentStore
from pillowtop.feed.couch import CouchChangeFeed
from pillowtop.logger import pillow_logging
from pillowtop.pillow.interface import PillowBase
try:
    from corehq.util.soft_assert import soft_assert
    _assert = soft_assert(to='@'.join(['czue', 'dim<EMAIL>']), fail_if_debug=True)
except ImportError:
    # hack for dependency resolution if corehq not available
    _assert = lambda assertion, message: None


WAIT_HEARTBEAT = 10000
CHANGES_TIMEOUT = 60000
RETRY_INTERVAL = 2  # seconds, exponentially increasing
MAX_RETRIES = 4  # exponential factor threshold for alerts


INDEX_REINDEX_SETTINGS = {"index": {"refresh_interval": "900s",
                                    "merge.policy.merge_factor": 20,
                                    "store.throttle.max_bytes_per_sec": "1mb",
                                    "store.throttle.type": "merge",
                                    "number_of_replicas": "0"}
}
INDEX_STANDARD_SETTINGS = {"index": {"refresh_interval": "1s",
                                     "merge.policy.merge_factor": 10,
                                     "store.throttle.max_bytes_per_sec": "5mb",
                                     "store.throttle.type": "node",
                                     "number_of_replicas": "0"}
}


class PillowtopIndexingError(Exception):
    pass


class PillowtopNetworkError(Exception):
    pass


def ms_from_timedelta(td):
    """
    Given a timedelta object, returns a float representing milliseconds
    """
    return (td.seconds * 1000) + (td.microseconds / 1000.0)


def lock_manager(obj):
    if isinstance(obj, LockManager):
        return obj
    else:
        return LockManager(obj, None)


class BasicPillow(PillowBase):
    """
    BasicPillow is actually a CouchPillow. PillowBase defines the actual interface.
    """
    checkpoint_frequency = CHECKPOINT_FREQUENCY
    couch_filter = None  # string for filter if needed
    extra_args = {}  # filter args if needed
    document_class = None  # couchdbkit Document class
    _couch_db = None
    include_docs = True
    use_locking = False

    def __init__(self, couch_db=None, document_class=None, checkpoint=None, change_feed=None):
        if document_class:
            self.document_class = document_class

        self._couch_db = couch_db
        self._checkpoint = checkpoint
        self._change_feed = change_feed

        if self.use_locking:
            # document_class must be a CouchDocLockableMixIn
            assert hasattr(self.document_class, 'get_obj_lock_by_id')

    def get_couch_db(self):
        if self._couch_db is None:
            self._couch_db = self.get_default_couch_db()
        return self._couch_db

    def set_couch_db(self, couch_db):
        self._couch_db = couch_db

    def get_default_couch_db(self):
        return self.document_class.get_db() if self.document_class else None

    @property
    def couch_db(self):
        _assert(False, 'People should not be using the couch_db properties!')
        return self.get_couch_db()

    @couch_db.setter
    def couch_db(self, value):
        _assert(False, 'People should not be using the couch_db properties!')
        self._couch_db = value

    @property
    def document_store(self):
        return CouchDocumentStore(self.get_couch_db())

    @property
    def checkpoint(self):
        if self._checkpoint is None:
            self._checkpoint = self._get_default_checkpoint()
        return self._checkpoint

    def _get_default_checkpoint(self):
        return PillowCheckpoint(
            self.document_store,
            construct_checkpoint_doc_id_from_name(self.get_name()),
        )

    def get_change_feed(self):
        if self._change_feed is None:
            self._change_feed = self._get_default_change_feed()
        return self._change_feed

    def _get_default_change_feed(self):
        return CouchChangeFeed(
            couch_db=self.get_couch_db(),
            couch_filter=self.couch_filter,
            include_docs=self.include_docs,
            extra_couch_view_params=self.extra_args
        )

    @memoized
    def get_name(self):
        return self.get_legacy_name()

    @classmethod
    def get_legacy_name(cls):
        return "%s.%s.%s" % (cls._get_base_name(), cls.__name__, get_machine_id())

    @classmethod
    def _get_base_name(cls):
        return cls.__module__

    def processor(self, change, context):
        """
        Parent processsor for a pillow class - this should not be overridden.
        This workflow is made for the situation where 1 change yields 1 transport/transaction
        """
        self.process_change(change)

    def fire_change_processed_event(self, change, context):
        if context.changes_seen % self.checkpoint_frequency == 0 and context.do_set_checkpoint:
            self.set_checkpoint(change)

    def process_change(self, change, is_retry_attempt=False):
        try:
            with lock_manager(self.change_trigger(change)) as t:
                if t is not None:
                    tr = self.change_transform(t)
                    if tr is not None:
                        self.change_transport(tr)
        except Exception, ex:
            if not is_retry_attempt:
                self._handle_pillow_error(change, ex)
            else:
                raise

    def _handle_pillow_error(self, change, exception):
        try:
            # This breaks the module boundary by using a show function defined in commcare-hq
            # but it was decided that it wasn't worth the effort to maintain the separation.
            meta = self.get_couch_db().show('domain/domain_date', change['id'])
        except ResourceNotFound:
            # Show function does not exist
            meta = None
        error = PillowError.get_or_create(change, self, change_meta=meta)
        error.add_attempt(exception, sys.exc_info()[2])
        error.save()
        pillow_logging.exception(
            "[%s] Error on change: %s, %s. Logged as: %s" % (
                self.get_name(),
                change['id'],
                exception,
                error.id
            )
        )

    def change_trigger(self, changes_dict):
        """
        Step one of pillowtop process
        For a given _changes indicator, the changes dict (the id, _rev) is sent here.

        Note, a couch _changes line is: {'changes': [], 'id': 'guid',  'seq': <int>}
        a 'deleted': True might be there too

        whereas a doc_dict is _id
        Should return a doc_dict
        """
        if changes_dict.get('deleted', False):
            # override deleted behavior on consumers that deal with deletions
            return None
        id = changes_dict['id']
        if self.use_locking:
            lock = self.document_class.get_obj_lock_by_id(id)
            lock.acquire()
            return LockManager(self.get_couch_db().open_doc(id), lock)
        elif changes_dict.get('doc', None) is not None:
            return changes_dict['doc']
        else:
            return self.get_couch_db().open_doc(id)

    def change_transform(self, doc_dict):
        """
        Step two of the pillowtop processor:
        Process/transform doc_dict if needed - by default, return the doc_dict passed.
        """
        return doc_dict

    def change_transport(self, doc_dict):
        """
        Step three of the pillowtop processor:
        Finish transport of doc if needed. Your subclass should implement this
        """
        raise NotImplementedError(
            "Error, this pillowtop subclass has not been configured to do anything!")


PYTHONPILLOW_CHUNK_SIZE = 250
PYTHONPILLOW_CHECKPOINT_FREQUENCY = CHECKPOINT_FREQUENCY * 10
PYTHONPILLOW_MAX_WAIT_TIME = 60


class PythonPillow(BasicPillow):
    """
    A pillow that does filtering in python instead of couch.

    Useful because it will actually set checkpoints throughout even if there
    are no matched docs.

    In initial profiling this was also 2-3x faster than the couch-filtered
    version.

    Subclasses should override the python_filter function to perform python
    filtering.
    """
    process_deletions = False

    def __init__(self, document_class=None, chunk_size=PYTHONPILLOW_CHUNK_SIZE,
                 checkpoint_frequency=PYTHONPILLOW_CHECKPOINT_FREQUENCY,
                 couch_db=None, checkpoint=None, change_feed=None, preload_docs=True):
        """
        Use chunk_size = 0 to disable chunking
        """
        super(PythonPillow, self).__init__(
            document_class=document_class,
            checkpoint=checkpoint,
            couch_db=couch_db,
            change_feed=change_feed,
        )
        self.change_queue = []
        self.chunk_size = chunk_size
        self.use_chunking = chunk_size > 0
        self.checkpoint_frequency = checkpoint_frequency
        self.include_docs = not self.use_chunking
        self.last_processed_time = None
        self.preload_docs = preload_docs

    def get_default_couch_db(self):
        if self.document_class and self.use_chunking:
            return CachedCouchDB(self.document_class.get_db().uri, readonly=False)
        else:
            return super(PythonPillow, self).get_default_couch_db()

    def python_filter(self, change):
        """
        Should return True if the doc is to be processed by your pillow
        """
        return True

    def process_chunk(self):
        def _assert_change_has_id(change):
            if 'id' not in change:
                _assert(False, "expected 'id' in change, but wasn't found! change is: {}".format(
                    simplejson.dumps(change)
                ))
                return False
            return True

        changes_to_process = filter(_assert_change_has_id, self.change_queue)
        if self.preload_docs:
            self.get_couch_db().bulk_load([change['id'] for change in changes_to_process],
                                     purge_existing=True)
        for change in changes_to_process:
            if self.preload_docs:
                doc = self.get_couch_db().open_doc(change['id'], check_main=False)
                change.document = doc

            # a valid change is either a non-preload situation or a valid doc + a filter match
            valid_change = (not self.preload_docs or change.document) and self.python_filter(change)
            valid_deletion = self.process_deletions and change.get('deleted', None)
            if valid_change or valid_deletion:
                try:
                    self.process_change(change)
                except Exception:
                    logging.exception('something went wrong processing change %s (%s)' %
                                      (change.get('seq', None), change['id']))

        # reset the queue after we've processed this chunk
        self.change_queue = []
        self.last_processed_time = datetime.utcnow()

    @property
    def queue_full(self):
        return len(self.change_queue) > self.chunk_size

    @property
    def wait_expired(self):
        if not self.last_processed_time:
            return False

        wait_time = datetime.utcnow() - self.last_processed_time
        return wait_time > timedelta(seconds=PYTHONPILLOW_MAX_WAIT_TIME)

    def processor(self, change, context):
        if self.use_chunking:
            self.change_queue.append(change)
            if self.queue_full or self.wait_expired:
                self.process_chunk()
        elif self.python_filter(change) or (change.get('deleted', None) and self.process_deletions):
            self.process_change(change)
        if context.changes_seen % self.checkpoint_frequency == 0 and context.do_set_checkpoint:
            # if using chunking make sure we never allow the checkpoint to get in
            # front of the chunks
            if self.use_chunking:
                self.process_chunk()
            self.set_checkpoint(change)

    def fire_change_processed_event(self, change, context):
        # todo: should fix this so that the checkpointing happens here.
        pass

    def run(self):
        self.change_queue = []
        self.last_processed_time = datetime.utcnow()
        super(PythonPillow, self).run()


def send_to_elasticsearch(path, es_getter, name, data=None, retries=MAX_RETRIES,
        except_on_failure=False, update=False, delete=False):
    """
    More fault tolerant es.put method
    """
    data = data if data is not None else {}
    current_tries = 0
    while current_tries < retries:
        try:
            if delete:
                res = es_getter().delete(path=path)
            elif update:
                params = {'retry_on_conflict': 2}
                res = es_getter().post("%s/_update" % path, data={"doc": data}, params=params)
            else:
                res = es_getter().put(path, data=data)
            break
        except ConnectionError, ex:
            current_tries += 1
            pillow_logging.error("[%s] put_robust error %s attempt %d/%d" % (
                name, ex, current_tries, retries))
            time.sleep(math.pow(RETRY_INTERVAL, current_tries))

            if current_tries == retries:
                message = "[%s] Max retry error on %s" % (name, path)
                if except_on_failure:
                    raise PillowtopIndexingError(message)
                else:
                    pillow_logging.error(message)
                res = {}

    if res.get('status', 0) == 400:
        error_message = "Pillowtop put_robust error [%s]:\n%s\n\tpath: %s\n\t%s" % (
            name,
            res.get('error', "No error message"),
            path,
            data.keys())

        if except_on_failure:
            raise PillowtopIndexingError(error_message)
        else:
            pillow_logging.error(error_message)
    return res


class AliasedElasticPillow(BasicPillow):
    """
    This pillow class defines it as being alias-able. That is, when you query it, you use an
    Alias to access it.

    This could be for varying reasons -  to make an index by certain metadata into its own index
    for performance/separation reasons

    Or, for our initial use case, needing to version/update the index mappings on the fly with
    minimal disruption.


    The workflow for altering an AliasedElasticPillow is that if you know you made a change, the
    pillow will create a new Index with a new md5sum as its suffix. Once it's finished indexing,
    you will need to flip the alias over to it.
    """
    es_host = ""
    es_port = ""
    es_index = ""
    es_type = ""
    es_alias = ''
    es_meta = {}
    es_timeout = 3  # in seconds
    default_mapping = None  # the default elasticsearch mapping to use for this
    bulk = False
    online = True  # online=False is for in memory (no ES) connectivity for testing purposes or for admin operations

    # Note - we allow for for existence because we do not care - we want the ES
    # index to always have the latest version of the case based upon ALL changes done to it.
    allow_updates = True

    def __init__(self, create_index=True, online=True, **kwargs):
        """
        create_index if the index doesn't exist on the ES cluster
        """
        super(AliasedElasticPillow, self).__init__(**kwargs)
        # online=False is used in unit tests
        self.online = online
        index_exists = self.index_exists()
        if create_index and not index_exists:
            self.create_index()
        if self.online:
            pillow_logging.info("Pillowtop [%s] Initializing mapping in ES" % self.get_name())
            self.initialize_mapping_if_necessary()
        else:
            pillow_logging.info("Pillowtop [%s] Started with no mapping from server in memory testing mode" % self.get_name())

    def initialize_mapping_if_necessary(self):
        """
        Initializes the elasticsearch mapping for this pillow if it is not found.
        """
        mapping = self.get_index_mapping()
        if not mapping:
            pillow_logging.info("Initializing elasticsearch mapping for [%s]" % self.es_type)
            mapping = copy(self.default_mapping)
            mapping['_meta']['created'] = datetime.isoformat(datetime.utcnow())
            mapping_res = self.set_mapping(self.es_type, {self.es_type: mapping})
            if mapping_res.get('ok', False) and mapping_res.get('acknowledged', False):
                # API confirms OK, trust it.
                pillow_logging.info("Mapping set: [%s] %s" % (self.es_type, mapping_res))
        else:
            pillow_logging.info("Elasticsearch mapping for [%s] was already present." % self.es_type)

    def index_exists(self):
        if not self.online:
            # If offline, just say the index is there and proceed along
            return True

        return self.get_es_new().indices.exists(self.es_index)

    def get_doc_path(self, doc_id):
        return "%s/%s/%s" % (self.es_index, self.es_type, doc_id)

    def update_settings(self, settings_dict):
        return self.send_robust("%s/_settings" % self.es_index, data=settings_dict)

    def set_index_reindex_settings(self):
        """
        Set a more optimized setting setup for fast reindexing
        """
        return self.update_settings(INDEX_REINDEX_SETTINGS)

    def set_index_normal_settings(self):
        """
        Normal indexing configuration
        """
        return self.update_settings(INDEX_STANDARD_SETTINGS)

    def get_index_mapping(self):
        es = self.get_es()
        return es.get('%s/_mapping' % self.es_index).get(self.es_index, {})

    def set_mapping(self, type_string, mapping):
        if self.online:
            return self.send_robust("%s/%s/_mapping" % (self.es_index, type_string), data=mapping)
        else:
{'fraction_non_alphanumeric': 0.0799179235729894, 'fraction_numerical': 0.003224431061130949, 'mean_word_length': 2.3602256446991405, 'pattern_counts': {'":': 26, '<': 8, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1267055', 'n_tokens_mistral': 10875, 'n_tokens_neox': 10469, 'n_words': 2892}