~bac/charmworld/tag-constraints

# Copyright 2012-2014 Marco Ceppi, Canonical Ltd.  This software is
# licensed under the GNU Affero General Public License version 3 (see
# the file LICENSE).

import argparse
from bzrlib.branch import Branch
from datetime import (
    datetime,
)
from email.utils import parseaddr
import os
import logging
from requests.exceptions import HTTPError
import sys
import yaml

from charmworld.charmstore import (
    CharmStore,
    get_store_charm_info,
    LocalCharmStore,
)
from charmworld.jobs.ingest import update_from_store
from charmworld.lp import (
    BASKET_SERIES,
    get_branch_tips,
)
from charmworld.models import (
    getconnection,
    getdb,
)
from charmworld.utils import (
    configure_logging,
    get_ini,
)
from config import (
    BASKET_QUEUE,
    CHARM_IMPORT_FILTER,
    CHARM_QUEUE,
)
from utils import (
    get_queue,
    lock,
    LockHeld,
    parse_branch,
)


unspecified = object()


def getLogger():
    return logging.getLogger('charm.launchpad')


def local_branch_data(local_repo, branch_data=None):
    """Find branch data for local charms.

    The tree rooted at local_repo is searched for Bazaar branches, which must
    be charms.  Any directory with a '.bzr' subdirectory is assumed to be the
    Bazaar branch for a charm and it will be included.
    :param local_repo: Path to root directory holding charms.
    :param branch_data: Existing branch data to be extended.
    :returns: List of charms and bundles, later is guaranteed to be empty.
    """
    log = getLogger()
    local_repo = os.path.abspath(os.path.expanduser(local_repo))
    charms = []
    baskets = []
    if branch_data is None:
        branch_data = []
        log = getLogger()
        for root, dirs, files in os.walk(local_repo):
            if '.bzr' in dirs:
                # Remove remaining dirs from examination.
                while len(dirs):
                    dirs.pop()
                try:
                    fullpath = os.path.join(local_repo, root)
                    series = root.split(os.sep)[-2]
                    branch = Branch.open(fullpath)
                    branch_data.append(
                        (fullpath, branch, series))
                except Exception as e:
                    # Log the exception but keep going.
                    log.error(e)
    for repo, branch, series in branch_data:
        metadata_file = os.path.join(repo, 'metadata.yaml')

        # If no owner can by gleaned, just use 'local'.
        owner = 'local'
        try:
            with open(metadata_file) as f:
                data = yaml.load(f)
                maintainers = data.get('maintainer') or data.get('maintainers')
                if maintainers:
                    if isinstance(maintainers, basestring):
                        maintainer = maintainers
                    else:
                        maintainer = maintainers[0]
                    if maintainer:
                        owner = parseaddr(maintainer)[1].split('@')[0]
        except IOError as e:
            msg = 'Error parsing maintainers from {} for {}: {}'.format(
                metadata_file, repo, e)
            log.warning(msg)

        name = repo.split(os.sep)[-1]
        log.info('{} owned by {}'.format(repo, owner))
        data = {
            'bname': u'trunk',
            'branch_deleted': False,
            'branch_spec': 'local:' + repo,
            'commit': branch.last_revision(),
            'distro_series': [],
            'name': name,
            'owner': owner,
            'promulgated': True,
            'series': series,
        }
        charms.append(data)
    return charms, baskets


def all_branch_data(branch_data=None):
    # Return data needed for the ingest job for charm branches on Launchpad.
    charms = []
    baskets = []
    log = getLogger()
    if branch_data is None:
        # Get all charm and bundle tip branches from Launchpad.
        try:
            branch_data = get_branch_tips(cache=False)
        except HTTPError as e:
            msg = 'Error retrieving branch tips:  {}'.format(e.message)
            log.error(msg)
            branch_data = []

    for repo, commit, series in branch_data:
        _, branch_name = repo.rsplit('/', 1)

        try:
            data = parse_branch(repo, series, commit)
        except ValueError as e:
            log.error('Parse error: {}'.format(e.message))
            continue

        if data['series'] == BASKET_SERIES:
            # The branch is a basket of bundles.
            data['promulgated'] = data['owner'] == u'charmers'
            if data['bname'] == 'bundle':
                baskets.append(data)
        else:
            # The branch is a charm.
            data['promulgated'] = data['series'] in data['distro_series']

            if data['bname'] != 'trunk':
                continue
            # The branch is a charm trunk branch.  Use it.
            data['branch_deleted'] = False
            charms.append(data)

    return charms, baskets


def db_charms(db):
    # Return data needed for the ingest job for charms stored in
    # charmworld's database.
    for charm in db.charms.find(
            fields=('owner', 'series', 'name', 'bname', 'branch_spec')):
        # Override the promulgation related settings: Data from this iterator
        # is used by all_charms() only when the charm no longer has a related
        # Launchpad branch. A charm is promulgated if its LP branch is linked
        # to a sourcepackage, hence a charm without a branch cannot be
        # promulgated.
        charm['promulgated'] = False
        charm['distro_series'] = []
        charm['branch_deleted'] = True
        yield charm


def _filtered_payloads(payloads, import_filter):
    if not import_filter:
        return iter(payloads)
    return (p for p in payloads if p['branch_spec'].startswith(import_filter))


def all_charms(db, available, limit=None, import_filter=None):
    # Return a sequence of charms that have a Launchpad branch or that are
    # already known in charmworld's database.
    log = getLogger()
    all_ = dict((charm['branch_spec'], charm) for charm in db_charms(db))
    all_.update((charm['branch_spec'], charm) for charm in available)
    all_ = sorted(all_.values(), key=lambda charm: charm['branch_spec'])

    count = 0
    for charm in _filtered_payloads(all_, import_filter):
        yield charm
        count += 1
        if limit and count >= limit:
            log.info("Import limit reached (%d)... stopping", limit)
            break


def queue_from_branches(db, store, charm_queue, basket_queue, limit=None,
                        import_filter=unspecified, check_time=None,
                        local_repo=None):
    log = getLogger()
    # Get charm and basket data from Launchpad.
    if local_repo is None:
        charm_data, baskets = all_branch_data()
    else:
        charm_data, baskets = local_branch_data(local_repo)
        store.set_charm_data(charm_data)

    if import_filter is unspecified:
        import_filter = CHARM_IMPORT_FILTER

    # Now get all charms, those just found on Launchpad and those already in
    # the database.
    if local_repo is None:
        charms = list(all_charms(db, charm_data, limit=limit,
                                 import_filter=import_filter))
    else:
        charms = charm_data
    charms_info = get_store_charm_info(store, charms, num_revisions=1)
    if check_time is None:
        check_time = datetime.utcnow().ctime()
    for charm, charm_info in zip(charms, charms_info):
        # Process the charms in reverse order to ensure the newest is last and
        # therefore the most current in the databases.
        newest_revision = charm_info[0][1]['revision']
        for address, store_data in reversed(charm_info):
            charm_copy = dict(charm)
            charm_copy['newest_revision'] = newest_revision
            update_from_store(charm_copy, address, store_data, check_time,
                              log)
            added = charm_queue.put(charm_copy)
            if added:
                log.info('Queueing charm %s (%d)',
                         charm_copy['branch_spec'],
                         charm_copy['store_data']['revision'])
    filtered_baskets = _filtered_payloads(baskets, import_filter)
    for basket in filtered_baskets:
        basket['branch_deleted'] = False
        added = basket_queue.put(basket)
        if added:
            log.info('Queueing bundle %s', basket['branch_spec'])


def dequeue():
    """Dequeue all the charm data in the CHARM_QUEUE.

    This is a script entry point to dequeue all charms. This might be done when
    when a running instance will get a code update that is incompatible with
    charm data in the queue.
    """
    configure_logging()
    log = getLogger()
    settings = get_ini()
    connection = getconnection(settings)
    db = getdb(connection, settings.get('mongo.database'))
    lease_time = int(settings['script_lease_time']) * 60
    try:
        with lock('ingest-queue', lease_time, db, log):
            queue = get_queue(db, CHARM_QUEUE)
            queue_size = queue.size()
            queue.clear()
            log.info("Dequeued %s charms", queue_size)
    except LockHeld, error:
        log.warn(str(error))


def configure_parser(parser):
    parser.add_argument(
        '--limit',
        help='Maximum number of charms to queue.',
        type=int)
    parser.add_argument(
        '--prefix',
        help='Only branches matching this prefix are queued.',
        type=str, default=unspecified)
    parser.add_argument(
        '--clear',
        help='Clear the queues before beginning.',
        action='store_true', dest='clear', default=False)
    parser.add_argument(
        '--debug',
        help='Turn on debugging',
        action='store_true', dest='debug',
        default=False)
    parser.add_argument(
        '--local-repo',
        help='Path to local repo.',
        default=None)


def clear_queues(queues):
    """Clear the queues."""
    for queue in queues:
        queue.clear()


def run(args):
    log = getLogger()
    if args.debug:
        log.setLevel(logging.DEBUG)
    settings = get_ini()
    if args.limit is None:
        charm_import_limit = settings.get('charm_import_limit')
        charm_import_limit = (
            int(charm_import_limit) if charm_import_limit else None)
    else:
        charm_import_limit = args.limit
    log.debug('Using import limit {}'.format(charm_import_limit))
    connection = getconnection(settings)
    db = getdb(connection, settings.get('mongo.database'))
    try:
        with lock('ingest-queue', int(settings['script_lease_time']) * 60,
                  db, log):
            charm_queue = get_queue(db, CHARM_QUEUE)
            basket_queue = get_queue(db, BASKET_QUEUE)
            # Release any entries that already exist.
            if args.clear:
                clear_queues((charm_queue, basket_queue))
            if args.local_repo is None:
                store = CharmStore()
            else:
                store = LocalCharmStore()
            queue_from_branches(db, store, charm_queue, basket_queue,
                                charm_import_limit, args.prefix,
                                local_repo=args.local_repo)
    except LockHeld, error:
        log.warn(str(error))


def main(arglist=None):
    if arglist is None:
        arglist = sys.argv[1:]
    configure_logging()
    parser = argparse.ArgumentParser()
    configure_parser(parser)
    args = parser.parse_args(arglist)
    run(args)


if __name__ == '__main__':
    main(sys.argv[1:])