~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

# pylint: disable-msg=W0702

__metaclass__ = type

__all__ = [

    'BadMessage',

    'JobScheduler',

    'LockError',

    'PullerMaster',

    'PullerMonitorProtocol',

    ]

import os

import socket

from StringIO import StringIO

from contrib.glock import (

    GlobalLock,

    LockAlreadyAcquired,

    )

from twisted.internet import (

    defer,

    error,

    reactor,

    )

from twisted.protocols.basic import (

    NetstringParseError,

    NetstringReceiver,

    )

from twisted.python import (

    failure,

    log,

    )

from canonical.config import config

from canonical.launchpad.webapp import errorlog

from lp.code.interfaces.codehosting import LAUNCHPAD_SERVICES

from lp.codehosting.puller import get_lock_id_for_branch_id

from lp.codehosting.puller.worker import get_canonical_url_for_branch_name

from lp.services.twistedsupport.processmonitor import (

    ProcessMonitorProtocolWithTimeout,

    )

from lp.services.twistedsupport.task import (

    ParallelLimitedTaskConsumer,

    PollingTaskSource,

    )

class BadMessage(Exception):

    """Raised when the protocol receives a message that we don't recognize."""

    def __init__(self, bad_netstring):

        Exception.__init__(

            self, 'Received unrecognized message: %r' % bad_netstring)

class UnexpectedStderr(Exception):

    """Raised when the worker prints to stderr."""

    def __init__(self, stderr):

        if stderr:

            last_line = stderr.splitlines()[-1]

        else:

            last_line = stderr

        Exception.__init__(

            self, "Unexpected standard error from subprocess: %s" % last_line)

        self.error = stderr

class PullerWireProtocol(NetstringReceiver):

    """The wire protocol for receiving events from the puller worker.

    The wire-level protocol is a series of netstrings.

    At the next level up, the protocol consists of messages which each look

    like this::

            [method-name] [number-of-arguments] [arguments]+

    Thus the instance is always in one of three states::

        [0] Waiting for command name.

        [1] Waiting for argument count.

        [2] Waiting for an argument.

    In state [0], we are waiting for a command name.  When we get one, we

    store it in self._current_command and move into state [1].

    In state [1], we are waiting for an argument count.  When we receive a

    message, we try to convert it to an integer.  If we fail in this, we call

    unexpectedError().  Otherwise, if it's greater than zero, we store the

    number in self._expected_args and go into state [2] or if it's zero

    execute the command (see below).

    In state [2], we are waiting for an argument.  When we receive one, we

    append it to self._current_args.  If len(self._current_args) ==

    self._expected_args, execute the command.

    "Executing the command" means looking for a method called

    do_<command name> on self.puller_protocol and calling it with

    *self._current_args.  If this raises, call

    self.puller_protocol.unexpectedError().

    The method _resetState() forces us back into state [0].

    """

    def __init__(self, puller_protocol):

        self.puller_protocol = puller_protocol

        self._resetState()

    def dataReceived(self, data):

        """See `NetstringReceiver.dataReceived`."""

        NetstringReceiver.dataReceived(self, data)

        # XXX: JonathanLange 2007-10-16

        # bug=http://twistedmatrix.com/trac/ticket/2851: There are no hooks in

        # NetstringReceiver to catch a NetstringParseError. The best we can do

        # is check the value of brokenPeer.

        if self.brokenPeer:

            self.puller_protocol.unexpectedError(

                failure.Failure(NetstringParseError(data)))

    def stringReceived(self, line):

        """See `NetstringReceiver.stringReceived`."""

        if (self._current_command is not None

            and self._expected_args is not None):

            # state [2]

            self._current_args.append(line)

        elif self._current_command is not None:

            # state [1]

            try:

                self._expected_args = int(line)

            except ValueError:

                self.puller_protocol.unexpectedError(failure.Failure())

        else:

            # state [0]

            if getattr(self.puller_protocol, 'do_%s' % line, None) is None:

                self.puller_protocol.unexpectedError(

                    failure.Failure(BadMessage(line)))

            else:

                self._current_command = line

        if len(self._current_args) == self._expected_args:

            # Execute the command.

            method = getattr(

                self.puller_protocol, 'do_%s' % self._current_command)

            try:

                try:

                    method(*self._current_args)

                except:

                    self.puller_protocol.unexpectedError(failure.Failure())

            finally:

                self._resetState()

    def _resetState(self):

        """Force into the 'waiting for command' state."""

        self._current_command = None

        self._expected_args = None

        self._current_args = []

class PullerMonitorProtocol(ProcessMonitorProtocolWithTimeout,

                            NetstringReceiver):

    """The protocol for receiving events from the puller worker."""

    def __init__(self, deferred, listener, clock=None):

        """Construct an instance of the protocol, for listening to a worker.

        :param deferred: A Deferred that will be fired when the worker has

            finished (either successfully or unsuccesfully).

        :param listener: A PullerMaster object that is notified when the

            protocol receives events from the worker.

        :param clock: A provider of Twisted's IReactorTime.  This parameter

            exists to allow testing that does not depend on an external clock.

            If a clock is not passed in explicitly the reactor is used.

        """

        ProcessMonitorProtocolWithTimeout.__init__(

            self, deferred, config.supermirror.worker_timeout, clock)

        self.reported_mirror_finished = False

        self.listener = listener

        self.wire_protocol = PullerWireProtocol(self)

        self._stderr = StringIO()

        self._deferred.addCallbacks(

            self.checkReportingFinishedAndNoStderr,

            self.ensureReportingFinished)

    def reportMirrorFinished(self, ignored):

        self.reported_mirror_finished = True

    def checkReportingFinishedAndNoStderr(self, result):

        """Check that the worker process behaved properly on clean exit.

        When the process exits cleanly, we expect it to have not printed

        anything to stderr and to have reported success or failure.  If it has

        failed to do either of these things, we should fail noisily."""

        stderr = self._stderr.getvalue()

        if stderr:

            fail = failure.Failure(UnexpectedStderr(stderr))

            fail.error = stderr

            return fail

        if not self.reported_mirror_finished:

            raise AssertionError('Process exited successfully without '

                                 'reporting success or failure?')

        return result

    def ensureReportingFinished(self, reason):

        """Clean up after the worker process exits uncleanly.

        If the worker process exited uncleanly, it probably didn't report

        success or failure, so we should report failure.  If there was output

        on stderr, it's probably a traceback, so we use the last line of that

        as a failure reason.

        """

        if not self.reported_mirror_finished:

            stderr = self._stderr.getvalue()

            reason.error = stderr

            if stderr:

                errorline = stderr.splitlines()[-1]

            else:

                errorline = str(reason.value)

            # The general policy when multiple errors occur is to report the

            # one that happens first and as an error has already happened here

            # (the process exiting uncleanly) we can only log.err() any

            # failure that comes from mirrorFailed failing.  In any case, we

            # just pass along the failure.

            report_failed_deferred = defer.maybeDeferred(

                self.listener.mirrorFailed, errorline, None)

            report_failed_deferred.addErrback(log.err)

            return report_failed_deferred.addCallback(

                lambda result: reason)

        else:

            return reason

    def makeConnection(self, process):

        """Called when the process has been created."""

        ProcessMonitorProtocolWithTimeout.makeConnection(self, process)

        NetstringReceiver.makeConnection(self, process)

        self.wire_protocol.makeConnection(process)

    def outReceived(self, data):

        self.wire_protocol.dataReceived(data)

    def errReceived(self, data):

        self._stderr.write(data)

    def do_startMirroring(self):

        self.resetTimeout()

        self.runNotification(self.listener.startMirroring)

    def do_branchChanged(self, stacked_on_url, revid_before, revid_after,

                         control_string, branch_string, repository_string):

        def branchChanged():

            d = defer.maybeDeferred(

                self.listener.branchChanged, stacked_on_url, revid_before,

                revid_after, control_string, branch_string, repository_string)

            d.addCallback(self.reportMirrorFinished)

            return d

        self.runNotification(branchChanged)

    def do_mirrorFailed(self, reason, oops):

        def mirrorFailed():

            d = defer.maybeDeferred(

                self.listener.mirrorFailed, reason, oops)

            d.addCallback(self.reportMirrorFinished)

            return d

        self.runNotification(mirrorFailed)

    def do_progressMade(self):

        """Any progress resets the timout counter."""

        self.resetTimeout()

    def do_log(self, message):

        self.listener.log(message)

class PullerMaster:

    """Controller for a single puller worker.

    The `PullerMaster` kicks off a child worker process and handles the events

    generated by that process.

    """

    path_to_script = os.path.join(config.root, 'scripts/mirror-branch.py')

    protocol_class = PullerMonitorProtocol

    def __init__(self, branch_id, source_url, unique_name, branch_type_name,

                 default_stacked_on_url, logger, client):

        """Construct a PullerMaster object.

        :param branch_id: The database ID of the branch to be mirrored.

        :param source_url: The location from which the branch is to be

            mirrored.

        :param unique_name: The unique name of the branch to be mirrored.

        :param branch_type_name: The name of the BranchType of the branch to

            be mirrored (e.g. 'MIRRORED').

        :param default_stacked_on_url: The default stacked-on URL for the

            product that the branch is in. '' implies that there is no such

            default.

        :param logger: A Python logging object.

        :param client: An asynchronous client for the branch status XML-RPC

            service.

        """

        self.branch_id = branch_id

        self.source_url = source_url.strip()

        self.destination_url = 'lp-internal:///%s' % (unique_name,)

        self.unique_name = unique_name

        self.branch_type_name = branch_type_name

        self.default_stacked_on_url = default_stacked_on_url

        self.logger = logger

        self.codehosting_endpoint = client

    def mirror(self):

        """Spawn a worker process to mirror a branch."""

        deferred = defer.Deferred()

        protocol = self.protocol_class(deferred, self)

        interpreter = '%s/bin/py' % config.root

        command = [

            interpreter, self.path_to_script, self.source_url,

            self.destination_url, str(self.branch_id), str(self.unique_name),

            self.branch_type_name,

            self.default_stacked_on_url]

        self.logger.debug("executing %s", command)

        env = os.environ.copy()

        env['BZR_EMAIL'] = get_lock_id_for_branch_id(self.branch_id)

        reactor.spawnProcess(protocol, interpreter, command, env=env)

        return deferred

    def run(self):

        """Launch a child worker and mirror a branch, handling errors.

        This is the main method to call to mirror a branch.

        :return: A Deferred that fires when the mirroring job is completed,

            one way or the other. It will never fire with a failure. The value

            of the Deferred itself is uninteresting (probably None).

        """

        deferred = self.mirror()

        deferred.addErrback(self.unexpectedError)

        return deferred

    def startMirroring(self):

        self.logger.info(

            'Worker started on branch %d: %s to %s', self.branch_id,

            self.source_url, self.destination_url)

    def mirrorFailed(self, reason, oops):

        self.logger.info('Recorded %s', oops)

        self.logger.info('Recorded failure: %s', str(reason))

        return self.codehosting_endpoint.callRemote(

            'mirrorFailed', self.branch_id, reason)

    def branchChanged(self, stacked_on_url, revid_before, revid_after,

                      control_string, branch_string, repository_string):

        if revid_before == revid_after:

            was_noop = 'noop'

        else:

            was_noop = 'non-trivial'

        self.logger.info(

            'Successfully mirrored %s branch %d %s to %s to from rev %s to %s'

            ' (%s)', self.branch_type_name, self.branch_id, self.source_url,

            self.destination_url, revid_before, revid_after, was_noop)

        return self.codehosting_endpoint.callRemote(

            'branchChanged', LAUNCHPAD_SERVICES, self.branch_id,

            stacked_on_url, revid_after, control_string, branch_string,

            repository_string)

    def log(self, message):

        self.logger.info('From worker: %s', message)

    def unexpectedError(self, failure):

        request = errorlog.ScriptRequest([

            ('branch_id', self.branch_id),

            ('source', self.source_url),

            ('dest', self.destination_url),

            ('error-explanation', failure.getErrorMessage())])

        request.URL = get_canonical_url_for_branch_name(self.unique_name)

        # If the sub-process exited abnormally, the stderr it produced is

        # probably a much more interesting traceback than the one attached to

        # the Failure we've been passed.

        tb = None

        if failure.check(error.ProcessTerminated, UnexpectedStderr):

            tb = getattr(failure, 'error', None)

        if tb is None:

            tb = failure.getTraceback()

        errorlog.globalErrorUtility.raising(

            (failure.type, failure.value, tb), request)

        self.logger.info('Recorded %s', request.oopsid)

class JobScheduler:

    """Schedule and manage the mirroring of branches.

    The jobmanager is responsible for organizing the mirroring of all

    branches.

    """

    def __init__(self, codehosting_endpoint, logger, branch_type_names):

        self.codehosting_endpoint = codehosting_endpoint

        self.logger = logger

        self.branch_type_names = branch_type_names

        self.actualLock = None

        self.name = 'branch-puller'

        self.lockfilename = '/var/lock/launchpad-%s.lock' % self.name

    def _turnJobTupleIntoTask(self, job_tuple):

        """Turn the return value of `acquireBranchToPull` into a job.

        `IBranchPuller.acquireBranchToPull` returns either an empty tuple

        (indicating there are no branches to pull currently) or a tuple of 6

        arguments, which are more or less those needed to construct a

        `PullerMaster` object.

        """

        if len(job_tuple) == 0:

            return None

        (branch_id, pull_url, unique_name,

         default_stacked_on_url, branch_type_name) = job_tuple

        master = PullerMaster(

            branch_id, pull_url, unique_name, branch_type_name,

            default_stacked_on_url, self.logger,

            self.codehosting_endpoint)

        return master.run

    def _poll(self):

        deferred = self.codehosting_endpoint.callRemote(

            'acquireBranchToPull', self.branch_type_names)

        deferred.addCallback(self._turnJobTupleIntoTask)

        return deferred

    def run(self):

        consumer = ParallelLimitedTaskConsumer(

            config.supermirror.maximum_workers, logger=self.logger)

        self.consumer = consumer

        source = PollingTaskSource(

            config.supermirror.polling_interval, self._poll,

            logger=self.logger)

        deferred = consumer.consume(source)

        deferred.addCallback(self._finishedRunning)

        return deferred

    def _finishedRunning(self, ignored):

        self.logger.info('Mirroring complete')

        return ignored

    def lock(self):

        self.actualLock = GlobalLock(self.lockfilename)

        try:

            self.actualLock.acquire()

        except LockAlreadyAcquired:

            raise LockError(self.lockfilename)

    def unlock(self):

        self.actualLock.release()

    def recordActivity(self, date_started, date_completed):

        """Record successful completion of the script."""

        started_tuple = tuple(date_started.utctimetuple())

        completed_tuple = tuple(date_completed.utctimetuple())

        return self.codehosting_endpoint.callRemote(

            'recordSuccess', self.name, socket.gethostname(), started_tuple,

            completed_tuple)

class LockError(StandardError):

    def __init__(self, lockfilename):

        StandardError.__init__(self)

        self.lockfilename = lockfilename

    def __str__(self):

        return 'Jobmanager unable to get master lock: %s' % self.lockfilename