multi_cell_list.py

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import abc
import copy
import heapq

import eventlet
import six

from oslo_log import log as logging

import nova.conf
from nova import context
from nova import exception
from nova.i18n import _

LOG = logging.getLogger(__name__)

CONF = nova.conf.CONF


class RecordSortContext(object):
    def __init__(self, sort_keys, sort_dirs):
        self.sort_keys = sort_keys
        self.sort_dirs = sort_dirs

    def compare_records(self, rec1, rec2):
        """Implements cmp(rec1, rec2) for the first key that is different.

        Adjusts for the requested sort direction by inverting the result
        as needed.
        """
        for skey, sdir in zip(self.sort_keys, self.sort_dirs):
            resultflag = 1 if sdir == 'desc' else -1
            if rec1[skey] < rec2[skey]:
                return resultflag
            elif rec1[skey] > rec2[skey]:
                return resultflag * -1
        return 0


class RecordWrapper(object):
    """Wrap a DB object from the database so it is sortable.

    We use heapq.merge() below to do the merge sort of things from the
    cell databases. That routine assumes it can use regular python
    operators (> and <) on the contents. Since that won't work with
    instances from the database (and depends on the sort keys/dirs),
    we need this wrapper class to provide that.

    Implementing __lt__ is enough for heapq.merge() to do its work.
    """
    def __init__(self, ctx, sort_ctx, db_record):
        self.cell_uuid = ctx.cell_uuid
        self._sort_ctx = sort_ctx
        self._db_record = db_record

    def __lt__(self, other):
        # NOTE(danms): This makes us always sort failure sentinels
        # higher than actual results. We do this so that they bubble
        # up in the get_records_sorted() feeder loop ahead of anything
        # else, and so that the implementation of RecordSortContext
        # never sees or has to handle the sentinels. If we did not
        # sort these to the top then we could potentially return
        # $limit results from good cells before we noticed the failed
        # cells, and would not properly report them as failed for
        # fix-up in the higher layers.
        if context.is_cell_failure_sentinel(self._db_record):
            return True
        elif context.is_cell_failure_sentinel(other._db_record):
            return False

        r = self._sort_ctx.compare_records(self._db_record,
                                           other._db_record)
        # cmp(x, y) returns -1 if x < y
        return r == -1


def query_wrapper(ctx, fn, *args, **kwargs):
    """This is a helper to run a query with predictable fail semantics.

    This is a generator which will mimic the scatter_gather_cells() behavior
    by honoring a timeout and catching exceptions, yielding the usual
    sentinel objects instead of raising. It wraps these in RecordWrapper
    objects, which will prioritize them to the merge sort, causing them to
    be handled by the main get_objects_sorted() feeder loop quickly and
    gracefully.
    """
    with eventlet.timeout.Timeout(context.CELL_TIMEOUT, exception.CellTimeout):
        try:
            for record in fn(ctx, *args, **kwargs):
                yield record
        except exception.CellTimeout:
            # Here, we yield a RecordWrapper (no sort_ctx needed since
            # we won't call into the implementation's comparison routines)
            # wrapping the sentinel indicating timeout.
            yield RecordWrapper(ctx, None, context.did_not_respond_sentinel)
            return
        except Exception as e:
            # Here, we yield a RecordWrapper (no sort_ctx needed since
            # we won't call into the implementation's comparison routines)
            # wrapping the exception object indicating failure.
            yield RecordWrapper(ctx, None, e.__class__(e.args))
            return


@six.add_metaclass(abc.ABCMeta)
class CrossCellLister(object):
    """An implementation of a cross-cell efficient lister.

    This primarily provides a listing implementation for fetching
    records from across multiple cells, paginated and sorted
    appropriately.  The external interface is the get_records_sorted()
    method. You should implement this if you need to efficiently list
    your data type from cell databases.

    """
    def __init__(self, sort_ctx, cells=None, batch_size=None):
        self.sort_ctx = sort_ctx
        self.cells = cells
        self.batch_size = batch_size
        self._cells_responded = set()
        self._cells_failed = set()
        self._cells_timed_out = set()

    @property
    def cells_responded(self):
        """A list of uuids representing those cells that returned a successful
        result.
        """
        return list(self._cells_responded)

    @property
    def cells_failed(self):
        """A list of uuids representing those cells that failed to return a
        successful result.
        """
        return list(self._cells_failed)

    @property
    def cells_timed_out(self):
        """A list of uuids representing those cells that timed out while being
        contacted.
        """
        return list(self._cells_timed_out)

    @property
    @abc.abstractmethod
    def marker_identifier(self):
        """Return the name of the property used as the marker identifier.

        For instances (and many other types) this is 'uuid', but could also
        be things like 'id' or anything else used as the marker identifier
        when fetching a page of results.
        """
        pass

    @abc.abstractmethod
    def get_marker_record(self, ctx, marker_id):
        """Get the cell UUID and instance of the marker record by id.

        This needs to look up the marker record in whatever cell it is in
        and return it. It should be populated with values corresponding to
        what is in self.sort_ctx.sort_keys.

        :param ctx: A RequestContext
        :param marker_id: The identifier of the marker to find
        :returns: A tuple of cell_uuid where the marker was found and an
                  instance of the marker from the database
        :raises: MarkerNotFound if the marker does not exist
        """
        pass

    @abc.abstractmethod
    def get_marker_by_values(self, ctx, values):
        """Get the identifier of the marker record by value.

        When we need to paginate across cells, the marker record exists
        in only one of those cells. The rest of the cells must decide on
        a record to be their equivalent marker with which to return the
        next page of results. This must be done by value, based on the
        values of the sort_keys properties on the actual marker, as if
        the results were sorted appropriately and the actual marker existed
        in each cell.

        :param ctx: A RequestContext
        :param values: The values of the sort_keys properties of fhe actual
                       marker instance
        :returns: The identifier of the equivalent marker in the local database
        """
        pass

    @abc.abstractmethod
    def get_by_filters(self, ctx, filters, limit, marker, **kwargs):
        """List records by filters, sorted and paginated.

        This is the standard filtered/sorted list method for the data type
        we are trying to list out of the database. Additional kwargs are
        passsed through.

        :param ctx: A RequestContext
        :param filters: A dict of column=filter items
        :param limit: A numeric limit on the number of results, or None
        :param marker: The marker identifier, or None
        :returns: A list of records
        """
        pass

    def get_records_sorted(self, ctx, filters, limit, marker, **kwargs):
        """Get a cross-cell list of records matching filters.

        This iterates cells in parallel generating a unified and sorted
        list of records as efficiently as possible. It takes care to
        iterate the list as infrequently as possible. We wrap the results
        in RecordWrapper objects so that they are sortable by
        heapq.merge(), which requires that the '<' operator just works.

        Our sorting requirements are encapsulated into the
        RecordSortContext provided to the constructor for this object.

        This function is a generator of records from the database like what you
        would get from instance_get_all_by_filters_sort() in the DB API.

        NOTE: Since we do these in parallel, a nonzero limit will be passed
        to each database query, although the limit will be enforced in the
        output of this function. Meaning, we will still query $limit from each
        database, but only return $limit total results.

        :param cell_down_support: True if the API (and caller) support
                                  returning a minimal instance
                                  construct if the relevant cell is
                                  down. If its True, then the value of
                                  CONF.api.list_records_by_skipping_down_cells
                                  is ignored and if its False, results are
                                  either skipped or erred based on the value of
                                  CONF.api.list_records_by_skipping_down_cells.
        """

        cell_down_support = kwargs.pop('cell_down_support', False)

        if marker:
            # A marker identifier was provided from the API. Call this
            # the 'global' marker as it determines where we start the
            # process across all cells. Look up the record in
            # whatever cell it is in and record the values for the
            # sort keys so we can find the marker instance in each
            # cell (called the 'local' marker).
            global_marker_cell, global_marker_record = self.get_marker_record(
                ctx, marker)
            global_marker_values = [global_marker_record[key]
                                    for key in self.sort_ctx.sort_keys]

        def do_query(cctx):
            """Generate RecordWrapper(record) objects from a cell.

            We do this inside the thread (created by
            scatter_gather_all_cells()) so that we return wrappers and
            avoid having to iterate the combined result list in the
            caller again. This is run against each cell by the
            scatter_gather routine.
            """

            # The local marker is an identifier of a record in a cell
            # that is found by the special method
            # get_marker_by_values(). It should be the next record
            # in order according to the sort provided, but after the
            # marker instance which may have been in another cell.
            local_marker = None

            # Since the regular DB query routines take a marker and assume that
            # the marked record was the last entry of the previous page, we
            # may need to prefix it to our result query if we're not the cell
            # that had the actual marker record.
            local_marker_prefix = []

            marker_id = self.marker_identifier

            if marker:
                if cctx.cell_uuid == global_marker_cell:
                    local_marker = marker
                else:
                    local_marker = self.get_marker_by_values(
                        cctx, global_marker_values)
                if local_marker:
                    if local_marker != marker:
                        # We did find a marker in our cell, but it wasn't
                        # the global marker. Thus, we will use it as our
                        # marker in the main query below, but we also need
                        # to prefix that result with this marker instance
                        # since the result below will not return it and it
                        # has not been returned to the user yet. Note that
                        # we do _not_ prefix the marker instance if our
                        # marker was the global one since that has already
                        # been sent to the user.
                        local_marker_filters = copy.copy(filters)
                        if marker_id not in local_marker_filters:
                            # If an $id filter was provided, it will
                            # have included our marker already if this
                            # instance is desired in the output
                            # set. If it wasn't, we specifically query
                            # for it. If the other filters would have
                            # excluded it, then we'll get an empty set
                            # here and not include it in the output as
                            # expected.
                            local_marker_filters[marker_id] = [local_marker]
                        local_marker_prefix = self.get_by_filters(
                            cctx, local_marker_filters, limit=1, marker=None,
                            **kwargs)
                else:
                    # There was a global marker but everything in our
                    # cell is _before_ that marker, so we return
                    # nothing. If we didn't have this clause, we'd
                    # pass marker=None to the query below and return a
                    # full unpaginated set for our cell.
                    return

            if local_marker_prefix:
                # Per above, if we had a matching marker object, that is
                # the first result we should generate.
                yield RecordWrapper(cctx, self.sort_ctx,
                                    local_marker_prefix[0])

            # If a batch size was provided, use that as the limit per
            # batch. If not, then ask for the entire $limit in a single
            # batch.
            batch_size = self.batch_size or limit

            # Keep track of how many we have returned in all batches
            return_count = 0

            # If limit was unlimited then keep querying batches until
            # we run out of results. Otherwise, query until the total count
            # we have returned exceeds the limit.
            while limit is None or return_count < limit:
                batch_count = 0

                # Do not query a full batch if it would cause our total
                # to exceed the limit
                if limit:
                    query_size = min(batch_size, limit - return_count)
                else:
                    query_size = batch_size

                # Get one batch
                query_result = self.get_by_filters(
                    cctx, filters,
                    limit=query_size or None, marker=local_marker,
                    **kwargs)

                # Yield wrapped results from the batch, counting as we go
                # (to avoid traversing the list to count). Also, update our
                # local_marker each time so that local_marker is the end of
                # this batch in order to find the next batch.
                for item in query_result:
                    local_marker = item[self.marker_identifier]
                    yield RecordWrapper(cctx, self.sort_ctx, item)
                    batch_count += 1

                # No results means we are done for this cell
                if not batch_count:
                    break

                return_count += batch_count
                LOG.debug(('Listed batch of %(batch)i results from cell '
                           'out of %(limit)s limit. Returned %(total)i '
                           'total so far.'),
                          {'batch': batch_count,
                           'total': return_count,
                           'limit': limit or 'no'})

        # NOTE(danms): The calls to do_query() will return immediately
        # with a generator. There is no point in us checking the
        # results for failure or timeout since we have not actually
        # run any code in do_query() until the first iteration
        # below. The query_wrapper() utility handles inline
        # translation of failures and timeouts to sentinels which will
        # be generated and consumed just like any normal result below.
        if self.cells:
            results = context.scatter_gather_cells(ctx, self.cells,
                                                   context.CELL_TIMEOUT,
                                                   query_wrapper, do_query)
        else:
            results = context.scatter_gather_all_cells(ctx,
                                                       query_wrapper, do_query)

        # If a limit was provided, it was passed to the per-cell query
        # routines.  That means we have NUM_CELLS * limit items across
        # results. So, we need to consume from that limit below and
        # stop returning results. Call that total_limit since we will
        # modify it in the loop below, but do_query() above also looks
        # at the original provided limit.
        total_limit = limit or 0

        # Generate results from heapq so we can return the inner
        # instance instead of the wrapper. This is basically free
        # as it works as our caller iterates the results.
        feeder = heapq.merge(*results.values())
        while True:
            try:
                item = next(feeder)
            except StopIteration:
                return

            if context.is_cell_failure_sentinel(item._db_record):
                if (not CONF.api.list_records_by_skipping_down_cells and
                        not cell_down_support):
                    # Value the config
                    # ``CONF.api.list_records_by_skipping_down_cells`` only if
                    # cell_down_support is False and generate the exception
                    # if CONF.api.list_records_by_skipping_down_cells is False.
                    # In all other cases the results from the down cell should
                    # be skipped now to either construct minimal constructs
                    # later if cell_down_support is True or to simply return
                    # the skipped results if cell_down_support is False.
                    raise exception.NovaException(
                        _('Cell %s is not responding but configuration '
                          'indicates that we should fail.')
                          % item.cell_uuid)
                LOG.warning('Cell %s is not responding and hence is '
                            'being omitted from the results',
                            item.cell_uuid)
                if item._db_record == context.did_not_respond_sentinel:
                    self._cells_timed_out.add(item.cell_uuid)
                elif isinstance(item._db_record, Exception):
                    self._cells_failed.add(item.cell_uuid)
                # We might have received one batch but timed out or failed
                # on a later one, so be sure we fix the accounting.
                if item.cell_uuid in self._cells_responded:
                    self._cells_responded.remove(item.cell_uuid)
                continue

            yield item._db_record
            self._cells_responded.add(item.cell_uuid)
            total_limit -= 1
            if total_limit == 0:
                # We'll only hit this if limit was nonzero and we just
                # generated our last one
                return