``rattail.datasync.config``

===========================

.. automodule:: rattail.datasync.config

   :members:

``rattail.datasync.daemon``

===========================

.. automodule:: rattail.datasync.daemon

   :members:

   config

   daemon

   watchers

``rattail.datasync.watchers``

=============================

.. automodule:: rattail.datasync.watchers

   :members:

#  Copyright © 2010-2022 Lance Edgar

This subpackage contains logic for the Rattail "datasync" daemon.

For general info see :doc:`rattail-manual:/data/sync/index`.

    This is normally started via command line, e.g.:

    .. code-block:: sh

       cd /srv/envs/poser

       bin/rattail -c app/datasync.conf datasync start

    .. note::

       Even though the naming implies a "proper" daemon, it will not

       actually daemonize itself.  For true daemon behavior, you

       should run this using a wrapper such as `supervisor`_.

       .. _supervisor: http://supervisord.org

        A separate thread is started for each watcher defined in

        config.  :func:`watch_for_changes()` is the target callable

        for each of these threads.

        Additionally, a separate thread is started for each consumer

        defined for the watcher.  :func:`consume_changes_perpetual()`

        is the target callable for each of these threads.

        Once all threads are started, this (main) thread loops forever

        (so as to stay alive, and hence keep child threads alive) but

        takes no further actions.

                    thread = Thread(target=consume_changes_perpetual, name=name, args=(self.config, consumer))

    Target for datasync watcher threads.

    This function will loop forever (barring an error) and

    periodically invoke the ``watcher`` object to check for any new

    changes, then pause before doing it again, etc.  It also tries to

    detect errors and handle them gracefully.

    The watcher itself is of course responsible for the mechanics of

    checking for new changes.  It also determines how frequently the

    checks should happen, etc.

    Each time the watcher finds/returns changes, this function will

    invoke :func:`record_or_process_changes()` with the result.

    :param config: Config object for the app.

    :param watcher: Reference to datasync watcher object.

    # let watcher do any setup it needs

    # outer loop - this should never terminate unless an unhandled

    # exception happens, or the daemon is stopped

        # reset for inner loop

        thisrun = app.make_utc(tzinfo=True)

        last_error_type = None

        # inner loop - this is for sake of retry

        while True:

                if last_error_type is not None and not isinstance(error, last_error_type):

                # remember which type of error this was; pause for next retry

                last_error_type = type(error)

                log.warning("attempt #%s of %s for watcher.get_changes() failed",

                            attempts, watcher.retry_attempts, exc_info=True)

                log.debug("pausing for %s seconds before making next attempt",

                          watcher.retry_delay)

            else: # watcher got changes okay (possibly empty set)

                # record new lastrun time

                try:

                    api.save_setting(None, lastrun_setting,

                                     lastrun.strftime(timefmt))

                except:

                    log.exception("failed to save lastrun time")

                    raise

                    log.debug("got %s changes from watcher", len(changes))

                    # record or process changes (depends on watcher)

                    try:

                        record_or_process_changes(config, watcher, changes)

                    except:

                        log.exception("failed to record/process changes")

                        raise

                    # prune changes if applicable (depends on watcher)

                    try:

                        prune_changes(config, watcher, changes)

                    except:

                        log.exception("failed to prune changes")

                        raise

                # break out of inner loop to reset the attempt count

                # for next grab

        # pause between successful change grabs

    This function is responsible for the "initial" handling of changes

    obtained from a watcher.  What this means will depend on the

    watcher, as follows:

    Most watchers are just that - their only job is to report new

    changes to datasync.  In this case this function will merely

    "record" the changes, by inserting them into the

    ``datasync_change`` queue table for further processing by the

    consumer(s).

    But some watchers "consume self" - meaning any changes they find,

    they also are responsible for processing.  In this case this

    function will "process" (consume) the changes directly, by

    invoking the watcher to do so.  These changes will *not* be added

    to the queue table for any other consumer(s) to process.

    :param changes: List of changes obtained from the watcher.

    # if watcher consumes itself, then it will process its own

    # changes.  note that there are no assumptions made about the

    # format or structure of these change objects.

        try:

            session = app.make_session()

        except:

            log.exception("failed to make session")

            raise

        except:

            log.exception("watcher failed to process its changes")

    # will record changes to consumer queue...

    # give all change stubs the same timestamp, to help identify them

    # as a "batch" of sorts, so consumers can process them as such.

    # (note, this is less important for identifiying a batch now that

    # we have batch_id, but is probably still helpful anyway)

    now = app.make_utc()

    # save change stub records to rattail database, for consumer

    # thread to find and process

    saved = 0

    try:

    except:

        log.exception("failed to make session for recording changes")

        raise

    try:

        # assign new/unique batch_id so that consumers can keep things

        # straight

    except:

        log.exception("failed to record changes")

        session.rollback()

        raise

    else:

    finally:

    log.debug("saved %s '%s' changes to datasync queue", saved, watcher.key)

    return True

def prune_changes(config, watcher, changes):

    if not watcher.prunes_changes:

        return

    try:

        # note that we only give it the keys for this

        pruned = watcher.prune_changes([c[0] for c in changes])

    except:

        log.exception("failed to prune changes")

        raise

    if pruned is not None:

        log.debug("watcher pruned %s changes", pruned)

def consume_changes_perpetual(config, consumer):

    Target for datasync consumer threads.

    This function will loop forever (barring an error) and

    periodically invoke the ``consumer`` object to process any changes

    in the queue, then pause before doing it again, etc.  It also

    tries to detect errors and handle them gracefully.

    This function is mostly just the loop itself; it calls

    :func:`consume_current_changes()` during each iteration.

    :param config: Config object for the app.

    :param consumer: Reference to datasync consumer object.

        try:

            result = consume_current_changes(config, consumer)

        except:

            log.exception("failed to consume current changes")

            raise

        if not result:

            # consumption failed, so exit the perma-loop (this thread

            # is now dead)

def consume_current_changes(config, consumer):

    The datasync queue table will be checked, and if it contains any

    changes applicable to the given consumer, then the consumer will

    be invoked to process the changes.

    If there are no applicable changes in the queue, this function

    will return without taking any real action.  But if there are

    changes, then it tries to be smart about processing them in the

    correct order, as follows:

    The changes are sorted by

    :attr:`~rattail.db.model.datasync.DataSyncChange.obtained` in

    order to determine the earliest timestamp.  Then it calls

    :func:`consume_changes_from()` with that timestamp.

    Once all changes with that timestamp have been processed

    (consumed), this function again looks for any applicable changes

    in the queue, sorting by timestamp and then calling

    :func:`consume_changes_from()` with earliest timestamp.

    This process repeats until there are no longer any changes in the

    queue which pertain to the given consumer.

    :param config: Config object for the app.

    :param consumer: Reference to datasync consumer object.

    try:

    except:

        log.exception("failed to make session for consuming changes")

        raise

    def get_first_change():

    try:

        first = get_first_change()

    except:

        log.exception("failed to get first change")

        session.close()

        return False

        try:

            if not consume_changes_from(config, session, consumer,

                                        first.obtained):

                error = True

        except:

            log.exception("failed to consume changes obtained at: %s",

                          first.obtained)

        if error:

        # fetch next 'obtained' timestamp

        try:

        except:

            log.exception("failed to get next-first change")

            break

    session.close()

def consume_changes_from(config, session, consumer, obtained):

    This fetches all changes from the datasync queue table, which

    correspond to the given consumer and which have an

    :attr:`~rattail.db.model.datasync.DataSyncChange.obtained` value

    matching the one specified.

    There are two possibilities here: either the matching changes are

    part of a "true" batch (i.e. they have a

    :attr:`~rattail.db.model.datasync.DataSyncChange.batch_id` value),

    or not.

    This function therefore first looks for changes which *do* have a

    ``batch_id``.  If found, it then sorts those changes by

    :attr:`~rattail.db.model.datasync.DataSyncChange.batch_sequence`

    to be sure they are processed in the correct order.

    If none of the changes have a ``batch_id`` then this function does

    not sort the changes in any way; they will be processed in

    (presumably) random order.

    In any case, regardless of ``batch_id``, at this point the

    function has identified a set of changes to be processed as a

    "batch" by the consumer.  But the larger the batch, the longer it

    will take for the consumer to process it.  This brings a couple of

    issues:

    If the consumer is a Rattail DB, and data versioning is enabled,

    this may cause rather massive resource usage if too many data

    writes happen.

    Additionally, there is no true "progress indicator" for datasync

    at this time.  A semi-practical way to determine its progress is

    simply to view the queue table and see what if anything it

    contains (when empty, processing is complete).  The set of changes

    being processed here, will be removed from the queue only after

    being processed.  Hence, the larger the batch, the "less granular"

    the "progress indicator" will be.

    To address these issues then, this function may "prune" the set of

    changes such that only so many are processed at a time.

    And finally, we have a (possibly smaller) set of changes to be

    processed.  This function will then ask the consumer to begin a

    new transaction, then process the changes, and ultimately commit

    the transaction.

    Once processing is complete (i.e. assuming no error) then those

    changes are removed from the queue.

    :param session: Current session for Rattail DB.

    :param consumer: Reference to datasync consumer object.

              app.localtime(obtained, from_utc=True))

            try:

                consumer.rollback_transaction()

            except:

                log.exception("consumer failed to rollback transaction")

                return False

            # commit consumer transaction

            try:

                consumer.commit_transaction()

            except:

                log.exception("consumer failed to commit transaction")

                return False

            try:

                for i, change in enumerate(changes):

                    session.delete(change)

                    if i % 200 == 0:

                        session.flush()

                session.commit()

            except:

                log.exception("failed to delete changes from queue")

                return False

        try:

            session = self.app.make_session(bind=self.engine)

        except:

            log.exception("failed to make session for pruning changes")

            raise

        try:

            for key in keys:

                if key: # note that key can sometimes be None

                    change = session.query(model.Change).get(key)

                    if change:

                        session.delete(change)

                        session.flush()

                        deleted += 1

        except:

            log.exception("failed to prune changes")

            session.rollback()

            raise

        else:

            session.commit()

        finally:

            session.close()