hgpoller.py

# ***** BEGIN LICENSE BLOCK *****
# Version: MPL 1.1/GPL 2.0/LGPL 2.1
#
# The contents of this file are subject to the Mozilla Public License Version
# 1.1 (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.mozilla.org/MPL/
#
# Software distributed under the License is distributed on an "AS IS" basis,
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
# for the specific language governing rights and limitations under the
# License.
#
# The Original Code is Mozilla-specific Buildbot steps.
#
# The Initial Developer of the Original Code is
# Mozilla Foundation.
# Portions created by the Initial Developer are Copyright (C) 2009
# the Initial Developer. All Rights Reserved.
#
# Contributor(s):
#   Axel Hecht <l10n@mozilla.com>
#   Ben Hearsum <bhearsum@mozilla.com>
#   Benjamin Smedberg <benjamin@smedbergs.us>
#   Chris AtLee <catlee@mozilla.com>
#   Chris Cooper <ccooper@deadsquid.com>
#
# Alternatively, the contents of this file may be used under the terms of
# either the GNU General Public License Version 2 or later (the "GPL"), or
# the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
# in which case the provisions of the GPL or the LGPL are applicable instead
# of those above. If you wish to allow use of your version of this file only
# under the terms of either the GPL or the LGPL, and not to allow others to
# use your version of this file under the terms of the MPL, indicate your
# decision by deleting the provisions above and replace them with the notice
# and other provisions required by the GPL or the LGPL. If you do not delete
# the provisions above, a recipient may use your version of this file under
# the terms of any one of the MPL, the GPL or the LGPL.
#
# ***** END LICENSE BLOCK *****

"""hgpoller provides Pollers to work on single hg repositories as well
as on a group of hg repositories. It's polling the json feed of pushlog,
which of the form

{
    "lastpushid": 30492,
    "pushes": {
        "15092": {
            "changesets": [
                {
                    "author": "Julian Seward <jseward@acm.org>",
                    "branch": "default",
                    "desc": "Bug 582668 - gfxAlphaBoxBlur::Paint appears to pass garbage down through Cairo. r=roc",
                    "files": [
                        "gfx/thebes/gfxBlur.cpp"
                    ],
                    "node": "ace72819f4a94b9175519a8fa5a1db654edae098",
                    "parents": [
                        "f2af48b0cd7c22c0af016d33a34ae5dc6e3141ab"
                    ],
                    "tags": []
                },
                {
                    "author": "Ms2ger <ms2ger@gmail.com>",
                    "branch": "default",
                    "desc": "Bug 585877 - remove document.height / document.width. r=sicking, sr=jst",
                    "files": [
                        "content/html/document/src/nsHTMLDocument.cpp",
                        "dom/interfaces/html/nsIDOMNSHTMLDocument.idl",
                        "js/src/xpconnect/src/dom_quickstubs.qsconf"
                    ],
                    "node": "43b490ef9dab30db2c4e2706110ad5d524a21597",
                    "parents": [
                        "ace72819f4a94b9175519a8fa5a1db654edae098"
                    ],
                    "tags": []
                },
                {
                    "author": "Martin Str\u00e1nsk\u00fd <stransky@redhat.com>",
                    "branch": "default",
                    "desc": "Bug 574354 - Disable OOP for plugins wrapped by nspluginwrapper. r=josh",
                    "files": [
                        "modules/plugin/base/src/nsNPAPIPlugin.cpp"
                    ],
                    "node": "75caf7ab03760f6bc39775cd8c4e097f33161c58",
                    "parents": [
                        "43b490ef9dab30db2c4e2706110ad5d524a21597"
                    ],
                    "tags": []
                }
            ],
            "date": 1281863455,
            "user": "dgottwald@mozilla.com"
        }
    }
}
"""

import time

from twisted.python import log
from twisted.internet import defer, reactor
from twisted.internet.task import LoopingCall
from twisted.web.client import getPage

from buildbot.changes import base, changes
from buildbot.util import json


def parse_pushlog_json(data):
    """Parse pushlog data version 2.

    Returns a normalized version of the data. The "pushes" key
    is a list of pushlog entries in order instead of a dict.
    """
    data = json.loads(data)

    pushes = []
    for pushid, push in data['pushes'].items():
        # Keys are strings in the JSON. We convert to int so the sort
        # below works appropriately. (A string sort may not always
        # have appropriate ordering.)
        push['pushid'] = int(pushid)
        pushes.append(push)

    # Sort by the push id because that is the monotonically incrementing
    # key.
    pushes.sort(key=lambda p: p['pushid'])

    data['pushes'] = pushes

    return data


class Pluggable(object):
    '''The Pluggable class implements a forward for Deferred's that
    can be thrown away.

    This is in particular useful when a network request doesn't really
    error in a reasonable time, and you want to make sure that if it
    answers after you tried to give up on it, it's not confusing the
    rest of your app by calling back with data twice or something.
    '''
    def __init__(self, d):
        self.d = defer.Deferred()
        self.dead = False
        d.addCallbacks(self.succeeded, self.failed)

    def succeeded(self, result):
        if self.dead:
            log.msg("Dead pluggable got called")
        else:
            self.d.callback(result)

    def failed(self, fail=None):
        if self.dead:
            log.msg("Dead pluggable got errbacked")
        else:
            self.d.errback(fail)


class BasePoller(object):
    attemptLimit = 3

    def __init__(self):
        self.attempts = 0
        self.startLoad = 0
        self.loadTime = None

    def poll(self):
        if self.attempts:
            if self.attempts > self.attemptLimit:
                self.plug.dead = True
                self.attempts = 0
                log.msg("dropping the ball on %s, starting new" % self)
            else:
                self.attempts += 1
                log.msg("Not polling %s because last poll is still working" %
                        self)
                reactor.callLater(0, self.pollDone, None)
                return
        self.attempts = 1
        self.startLoad = time.time()
        self.loadTime = None
        self.plug = Pluggable(self.getData())
        d = self.plug.d
        d.addCallback(self.stopLoad)
        d.addCallback(self.processData)
        d.addCallbacks(self.dataFinished, self.dataFailed)
        d.addCallback(self.pollDone)
        return d

    def stopLoad(self, res):
        self.loadTime = time.time() - self.startLoad
        return res

    def dataFinished(self, res):
        assert self.attempts
        self.attempts = 0

    def dataFailed(self, res):
        assert self.attempts
        self.attempts = 0
        log.msg("%s: polling failed, result %s" % (self, res.value.message))
        res.printTraceback()

    def pollDone(self, res):
        pass


class BaseHgPoller(BasePoller):
    """Common base of HgPoller, HgLocalePoller, and HgAllLocalesPoller.

    Subclasses should implement getData, processData, and __str__"""
    verbose = True
    timeout = 30

    def __init__(self, hgURL, branch, pushlogUrlOverride=None,
                 tipsOnly=False, tree=None, repo_branch=None, maxChanges=100,
                 mergePushChanges=True):
        self.super_class = BasePoller
        self.super_class.__init__(self)
        self.hgURL = hgURL
        self.branch = branch
        self.tree = tree
        if hgURL.endswith("/"):
            hgURL = hgURL[:-1]
        fragments = [hgURL, branch]
        if tree is not None:
            fragments.append(tree)
        self.baseURL = "/".join(fragments)
        self.pushlogUrlOverride = pushlogUrlOverride
        self.tipsOnly = tipsOnly
        self.lastChangeset = None
        self.lastPushID = None
        self.startLoad = 0
        self.loadTime = None
        self.repo_branch = repo_branch
        self.maxChanges = maxChanges
        # With mergePushChanges=True we get one buildbot change per push to hg.
        # The files from all changes in the push will be accumulated in the buildbot change
        # and the comments of tipmost change of the push will be used
        self.mergePushChanges = mergePushChanges

        self.emptyRepo = False

    def getData(self):
        url = self._make_url()
        if self.verbose:
            log.msg("Polling Hg server at %s" % url)
        return getPage(url, timeout=self.timeout)

    def _make_url(self):
        url = None
        if self.pushlogUrlOverride:
            url = self.pushlogUrlOverride
        else:
            url = "/".join((self.baseURL, 'json-pushes?version=2&full=1'))

        args = []
        if self.lastChangeset is not None:
            args.append('fromchange=' + self.lastChangeset)
        if self.tipsOnly:
            args.append('tipsonly=1')
        if args:
            if '?' not in url:
                url += '?'
            else:
                url += '&'
            url += '&'.join(args)

        return str(url)

    def processData(self, query):
        push_data = parse_pushlog_json(query)

        # The payload tells us the most recent push ID. If it is the empty
        # string, the pushlog is empty and there is no data to consume.
        if not push_data['lastpushid']:
            self.emptyRepo = True
            self.lastPushID = None
            if self.verbose:
                log.msg('%s is empty' % self.baseURL)
            return

        # If nothing has changed and we're fully caught up, the remote
        # lastpushid will be the same as self.lastPushID.
        #
        # If the remote lastpushid is less than a previously observed value,
        # this could mean one of the following:
        #
        #    a) Data from the pushlog was removed (perhaps the repo was
        #       stripped)
        #    b) The repo/pushlog was reset.
        #
        # These scenarios should be rare. In both of them, our assumption
        # about the behavior of the pushlog always being monotonically
        # increasing have been invalidated. So we reset state and start
        # again.
        #
        # It's worth noting that a reset repo's pushlog could have *more*
        # entries than the former repo. In this case, this code will fail
        # to detect a reset repo from the pushlog alone.
        if self.lastPushID and push_data['lastpushid'] < self.lastPushID:
            self.emptyRepo = False
            self.lastPushID = None
            self.lastChangeset = None
            log.msg('%s appears to have been reset; clearing state' %
                    self.baseURL)
            return

        # We want to add at most self.maxChanges changes per push. If
        # mergePushChanges is True, then we'll get up to maxChanges pushes,
        # each with up to maxChanges changes.
        # Go through the list of pushes backwards, since we want to keep the
        # latest ones and possibly discard earlier ones.
        change_list = []
        too_many = False
        for push in reversed(push_data['pushes']):
            # Used for merging push changes
            c = dict(
                user=push['user'],
                date=push['date'],
                files=[],
                desc="",
                node=None,
                commit_titles=[],
                commit_titles_total_length=0,
            )

            i = 0
            for change in reversed(push['changesets']):
                if self.maxChanges is not None and (len(change_list) >= self.maxChanges or
                                                    i >= self.maxChanges):
                    too_many = True
                    log.msg("%s: got too many changes" % self.baseURL)
                    break

                # Ignore changes not on the specified in-repo branch.
                if self.repo_branch is not None and self.repo_branch != change['branch']:
                    continue

                i += 1

                if self.mergePushChanges:
                    # Collect all the files for this push
                    c['files'].extend(change['files'])
                    # Keep the comments and revision of the last change of this push.
                    # We're going through the changes in reverse order, so we
                    # should use the comments and revision of the first change
                    # in this loop
                    if c['node'] is None:
                        c['desc'] = change['desc']
                        c['node'] = change['node']

                    title = change['desc'].split('\n', 1)[0]
                    if len(title) > 100:
                        trim_pos = title.rfind(' ', 0, 100)
                        if trim_pos == -1:
                            trim_pos = 100
                        title = title[:trim_pos]
                    # The commit titles are stored in a Change property, which
                    # are limited to 1024 chars in the database (see
                    # change_properties in buildbot/db/scheme/tables.sql). In
                    # order to avoid insert/update failures, we enforce a cap
                    # on the total length with enough room for JSON overhead.
                    if c['commit_titles_total_length'] + len(title) + 5 <= 800:
                        c['commit_titles_total_length'] += len(title) + 5  # for json encoding like , " etc.
                        c['commit_titles'].append(title)
                else:
                    c = dict(
                        user=push['user'],
                        date=push['date'],
                        files=change['files'],
                        desc=change['desc'],
                        node=change['node'],
                        branch=change['branch'],
                    )
                    change_list.append(c)

            if too_many and self.mergePushChanges:
                # Add a dummy change to indicate we had too many changes
                c['files'].extend(['overflow'])

            if self.mergePushChanges and c['node'] is not None:
                change_list.append(c)

        if too_many and not self.mergePushChanges:
            # We add this at the end, and the list gets reversed below. That
            # means this dummy change ends up being the 'first' change of the
            # set, and buildbot chooses the last change as the one to
            # build, so this dummy change doesn't impact which revision
            # gets built.
            c = dict(
                user='buildbot',
                files=['overflow'],
                node=None,
                desc='more than maxChanges(%i) received; ignoring the rest' % self.maxChanges,
                date=time.time(),
            )
            change_list.append(c)

        # Un-reverse the list of changes so they get added in the right order
        change_list.reverse()

        # If we have a lastChangeset we're comparing against, we've been
        # running for a while and so any changes returned here are new.

        # If the repository was previously empty (indicated by emptyRepo=True),
        # we also want to pay attention to all these pushes.

        # If we don't have a lastChangeset and the repository isn't empty, then
        # don't trigger any new builds, and start monitoring for changes since
        # the latest changeset in the repository
        if self.lastChangeset is not None or self.emptyRepo:
            for change in change_list:
                link = "%s/rev/%s" % (self.baseURL, change["node"])
                c = changes.Change(who=change["user"],
                                   files=change["files"],
                                   revision=change["node"],
                                   comments=change["desc"],
                                   revlink=link,
                                   when=change["date"],
                                   branch=self.branch)
                if 'commit_titles' in change:
                    c.properties.setProperty('commit_titles',
                                             change['commit_titles'],
                                             'BaseHgPoller')
                self.changeHook(c)
                self.parent.addChange(c)

        # The repository isn't empty any more!
        self.emptyRepo = False
        # Use the last change found by the poller, regardless of if it's on our
        # branch or not. This is so we don't have to constantly ignore it in
        # future polls.
        self.lastChangeset = push_data['pushes'][-1]['changesets'][-1]['node']
        self.lastPushID = push_data['pushes'][-1]['pushid']
        if self.verbose:
            log.msg("last changeset %s on %s" %
                    (self.lastChangeset, self.baseURL))

    def changeHook(self, change):
        pass


class HgPoller(base.ChangeSource, BaseHgPoller):
    """This source will poll a Mercurial server over HTTP using
    the built-in RSS feed for changes and submit them to the
    change master."""

    compare_attrs = ['hgURL', 'branch', 'pollInterval',
                     'pushlogUrlOverride', 'tipsOnly', 'storeRev',
                     'repo_branch', 'maxChanges']
    parent = None
    loop = None
    volatile = ['loop']

    def __init__(self, hgURL, branch, pushlogUrlOverride=None,
                 tipsOnly=False, pollInterval=30, storeRev=None,
                 repo_branch="default", maxChanges=100):
        """
        @type   hgURL:          string
        @param  hgURL:          The base URL of the Hg repo
                                (e.g. https://hg.mozilla.org/)
        @type   branch:         string
        @param  branch:         The branch to check (e.g. mozilla-central)
        @type   pollInterval:   int
        @param  pollInterval:   The time (in seconds) between queries for
                                changes
        @type   tipsOnly:       bool
        @param  tipsOnly:       Make the pushlog only show the tips of pushes.
                                With this enabled every push will only show up
                                as *one* changeset
        @type   storeRev:       string
        @param  storeRev:       A name of a property to set on the resulting
                                Change to help identify the specific repository
                                if multiple HgPollers are used in one branch.
        @type   repo_branch:    string or None
        @param  repo_branch:    Name of the in-repo branch to pay attention to.
                                If None, then pay attention to all branches.
        """

        BaseHgPoller.__init__(self, hgURL, branch, pushlogUrlOverride,
                              tipsOnly, repo_branch=repo_branch, maxChanges=maxChanges)
        self.pollInterval = pollInterval
        self.storeRev = storeRev

    def startService(self):
        self.loop = LoopingCall(self.poll)
        base.ChangeSource.startService(self)
        reactor.callLater(0, self.loop.start, self.pollInterval)

    def stopService(self):
        if self.running:
            self.loop.stop()
        return base.ChangeSource.stopService(self)

    def describe(self):
        return "Getting changes from: %s" % self._make_url()

    def __str__(self):
        return "<HgPoller for %s%s>" % (self.hgURL, self.branch)

    def changeHook(self, change):
        if self.storeRev:
            change.properties.setProperty(
                self.storeRev, change.revision, 'HgPoller')


class HgLocalePoller(BaseHgPoller):
    """This helper class for HgAllLocalesPoller polls a single locale and
    submits changes if necessary."""

    timeout = 30
    verbose = False

    def __init__(self, locale, parent, branch, hgURL):
        BaseHgPoller.__init__(self, hgURL, branch, tree=locale)
        self.locale = locale
        self.parent = parent
        self.branch = branch

    def changeHook(self, change):
        change.properties.setProperty('locale', self.locale, 'HgLocalePoller')
        change.properties.setProperty(
            'l10n_revision', change.revision, 'HgLocalePoller')

    def pollDone(self, res):
        self.parent.localeDone(self.locale)

    def __str__(self):
        return "<HgLocalePoller for %s>" % self.baseURL


class HgAllLocalesPoller(base.ChangeSource, BasePoller):
    """Poll all localization repositories from an index page.

    For a index page like https://hg.mozilla.org/releases/l10n-mozilla-1.9.1/,
    all links look like /releases/l10n-mozilla-1.9.1/af/, where the last
    path step will be the locale code, and the others will be passed
    as branch for the changes, i.e. 'releases/l10n-mozilla-1.9.1'.
    """

    compare_attrs = ['repositoryIndex', 'pollInterval']
    parent = None
    loop = None
    volatile = ['loop']

    timeout = 10
    parallelRequests = 2
    verboseChilds = False

    def __init__(self, hgURL, repositoryIndex, pollInterval=120, branch=None):
        """
        @type  repositoryIndex:      string
        @param repositoryIndex:      The URL listing all locale repos
        @type  pollInterval        int
        @param pollInterval        The time (in seconds) between queries for
                                   changes
        @type  branch:      string
        @param branch:      Used by caller to uniquely identify this object
        """

        BasePoller.__init__(self)
        self.hgURL = hgURL
        if hgURL.endswith("/"):
            hgURL = hgURL[:-1]
        self.repositoryIndex = repositoryIndex
        self.pollInterval = pollInterval
        self.localePollers = {}
        self.locales = []
        self.pendingLocales = []
        self.activeRequests = 0
        self.branch = branch

    def startService(self):
        self.loop = LoopingCall(self.poll)
        base.ChangeSource.startService(self)
        reactor.callLater(0, self.loop.start, self.pollInterval)

    def stopService(self):
        if self.running:
            self.loop.stop()
        return base.ChangeSource.stopService(self)

    def addChange(self, change):
        self.parent.addChange(change)

    def describe(self):
        return "Getting changes from all locales at %s" % self.repositoryIndex

    def getData(self):
        log.msg("Polling all locales at %s/%s/" % (self.hgURL,
                                                   self.repositoryIndex))
        return getPage(self.hgURL + '/' + self.repositoryIndex + '/?style=raw',
                       timeout=self.timeout)

    def getLocalePoller(self, locale, branch):
        if (locale, branch) not in self.localePollers:
            lp = HgLocalePoller(locale, self, branch,
                                self.hgURL)
            lp.verbose = self.verboseChilds
            self.localePollers[(locale, branch)] = lp
        return self.localePollers[(locale, branch)]

    def processData(self, data):
        locales = filter(None, data.split())
        # get locales and branches

        def brancher(link):
            steps = filter(None, link.split('/'))
            loc = steps.pop()
            branch = '/'.join(steps)
            return (loc, branch)
        # locales is now locale code / branch tuple
        locales = map(brancher, locales)
        if locales != self.locales:
            log.msg("new locale list: " + " ".join(map(str, locales)))
        self.locales = locales
        self.pendingLocales = locales[:]
        # prune removed locales from pollers
        for oldLoc in self.localePollers.keys():
            if oldLoc not in locales:
                self.localePollers.pop(oldLoc)
                log.msg("not polling %s on %s anymore, dropped from repositories" %
                        oldLoc)
        for i in xrange(self.parallelRequests):
            self.activeRequests += 1
            reactor.callLater(0, self.pollNextLocale)

    def pollNextLocale(self):
        if not self.pendingLocales:
            self.activeRequests -= 1
            if not self.activeRequests:
                msg = "%s done with all locales" % str(self)
                loadTimes = map(
                    lambda p: p.loadTime, self.localePollers.values())
                goodTimes = filter(lambda t: t is not None, loadTimes)
                if not goodTimes:
                    msg += ". All %d locale pollers failed" % len(loadTimes)
                else:
                    msg += ", min: %.1f, max: %.1f, mean: %.1f" % \
                        (min(goodTimes), max(goodTimes),
                         sum(goodTimes) / len(goodTimes))
                    if len(loadTimes) > len(goodTimes):
                        msg += ", %d failed" % (
                            len(loadTimes) - len(goodTimes))
                log.msg(msg)
                log.msg("Total time: %.1f" % (time.time() - self.startLoad))
            return
        loc, branch = self.pendingLocales.pop(0)
        poller = self.getLocalePoller(loc, branch)
        poller.poll()

    def localeDone(self, loc):
        if self.verboseChilds:
            log.msg("done with " + loc)
        reactor.callLater(0, self.pollNextLocale)

    def __str__(self):
        return "<HgAllLocalesPoller for %s/%s/>" % (self.hgURL,
                                                    self.repositoryIndex)