Major refactor of data API

* update methods are defined on resource types
* resource types know how to send messages
  - routing keys generated automatically from the message contents
* change sources use the master.data.updates.addChange
  - with some differences from master.addChange
* more consistent and automatic testing of data and message formats
* documentation on writing endpoints, resource types, etc.

NOTE: data.updates.addChange does not yet enforce unicode-ness on its
arguments, but it will.

master/buildbot/changes/bonsaipoller.py

@@ -21,7 +21,6 @@
 from twisted.web import client
 
 from buildbot.changes import base
-from buildbot.util import epoch2datetime
 
 class InvalidResultError(Exception):
     def __init__(self, value="InvalidResultError"):
@@ -268,8 +267,8 @@ def _process_changes(self, query):
             files = [file.filename + ' (revision '+file.revision+')'
                      for file in cinode.files]
             self.lastChange = self.lastPoll
-            yield self.master.addChange(author = cinode.who,
+            yield self.master.data.updates.addChange(author = cinode.who,
                                files = files,
                                comments = cinode.log,
-                               when_timestamp = epoch2datetime(cinode.date),
+                               when_timestamp = cinode.date,
                                branch = self.branch)

master/buildbot/changes/gerritchangesource.py

@@ -113,7 +113,7 @@ def flatten(event, base, d):
         flatten(properties, "event", event)
         return func(properties,event)
     def addChange(self, chdict):
-        d = self.master.addChange(**chdict)
+        d = self.master.data.updates.addChange(**chdict)
         # eat failures..
         d.addErrback(log.err, 'error adding change from GerritChangeSource')
         return d

master/buildbot/changes/gitpoller.py

@@ -21,7 +21,6 @@
 
 from buildbot.util import deferredLocked
 from buildbot.changes import base
-from buildbot.util import epoch2datetime
 
 class GitPoller(base.PollingChangeSource):
     """This source will poll a remote git repo for changes and submit
@@ -184,7 +183,7 @@ def process(git_output):
             stripped_output = git_output.strip()
             if self.usetimestamps:
                 try:
-                    stamp = float(stripped_output)
+                    stamp = int(stripped_output)
                 except Exception, e:
                         log.msg('gitpoller: caught exception converting output \'%s\' to timestamp' % stripped_output)
                         raise e
@@ -269,12 +268,12 @@ def _process_changes(self, unused_output):
                 raise failures[0]
 
             timestamp, author, files, comments = [ r[1] for r in results ]
-            yield self.master.addChange(
+            yield self.master.data.updates.addChange(
                    author=author,
                    revision=rev,
                    files=files,
                    comments=comments,
-                   when_timestamp=epoch2datetime(timestamp),
+                   when_timestamp=timestamp,
                    branch=self.branch,
                    category=self.category,
                    project=self.project,

master/buildbot/changes/mail.py

@@ -60,7 +60,7 @@ def add_change(chtuple):
             if chtuple:
                 src, chdict = chtuple
             if chdict:
-                return self.master.addChange(src=src, **chdict)
+                return self.master.data.updates.addChange(src=src, **chdict)
             else:
                 log.msg("no change found in maildir file '%s'" % filename)
         d.addCallback(add_change)

master/buildbot/changes/p4poller.py

@@ -155,7 +155,8 @@ def _poll(self):
             if not m:
                 raise P4PollerError("Unexpected 'p4 describe -s' result: %r" % result)
             who = m.group('who')
-            when = time.mktime(time.strptime(m.group('when'), self.datefmt))
+            when = int(time.mktime(time.strptime(m.group('when'),
+                                                 self.datefmt)))
             comments = ''
             while not lines[0].startswith('Affected files'):
                 comments += lines.pop(0) + '\n'
@@ -178,12 +179,12 @@ def _poll(self):
                         branch_files[branch] = [file]
 
             for branch in branch_files:
-                yield self.master.addChange(
+                yield self.master.data.updates.addChange(
                        author=who,
                        files=branch_files[branch],
                        comments=comments,
                        revision=str(num),
-                       when_timestamp=util.epoch2datetime(when),
+                       when_timestamp=when,
                        branch=branch,
                        project=self.project)
 

master/buildbot/changes/pb.py

@@ -19,7 +19,6 @@
 
 from buildbot.pbutil import NewCredPerspective
 from buildbot.changes import base
-from buildbot.util import epoch2datetime
 from buildbot import config
 
 class ChangePerspective(NewCredPerspective):
@@ -49,18 +48,17 @@ def perspective_addChange(self, changedict):
         # "old" names (who, when, and isdir), as they are not deprecated yet,
         # although the master will accept the new names (author,
         # when_timestamp, and is_dir).  After a few revisions have passed, we
-        # can switch the client to use the new names.
+        # can switch the client to use the new names.  isdir/is_dir are no
+        # longer used and thus deleted here
         if 'isdir' in changedict:
-            changedict['is_dir'] = changedict['isdir']
             del changedict['isdir']
+        if 'is_dir' in changedict:
+            del changedict['is_dir']
         if 'who' in changedict:
             changedict['author'] = changedict['who']
             del changedict['who']
         if 'when' in changedict:
-            when = None
-            if changedict['when'] is not None:
-                when = epoch2datetime(changedict['when'])
-            changedict['when_timestamp'] = when
+            changedict['when_timestamp'] = changedict['when']
             del changedict['when']
 
         # turn any bytestring keys into unicode, assuming utf8 but just
@@ -87,9 +85,9 @@ def perspective_addChange(self, changedict):
 
         if not files:
             log.msg("No files listed in change... bit strange, but not fatal.")
-        d = self.master.addChange(**changedict)
-        # since this is a remote method, we can't return a Change instance, so
-        # this just sets the return value to None:
+        d = self.master.data.updates.addChange(**changedict)
+        # set the return value to None, so we don't get users depending on
+        # getting a changeid
         d.addCallback(lambda _ : None)
         return d
 

master/buildbot/changes/svnpoller.py

@@ -310,10 +310,6 @@ def create_changes(self, new_logentries):
             for p in pathlist.getElementsByTagName("path"):
                 action = p.getAttribute("action")
                 path = "".join([t.data for t in p.childNodes])
-                # the rest of buildbot is certaily not yet ready to handle
-                # unicode filenames, because they get put in RemoteCommands
-                # which get sent via PB to the buildslave, and PB doesn't
-                # handle unicode.
                 path = path.encode("ascii")
                 if path.startswith("/"):
                     path = path[1:]
@@ -355,7 +351,7 @@ def create_changes(self, new_logentries):
     @defer.inlineCallbacks
     def submit_changes(self, changes):
         for chdict in changes:
-            yield self.master.addChange(src='svn', **chdict)
+            yield self.master.data.updates.addChange(src='svn', **chdict)
 
     def finished_ok(self, res):
         if self.cachepath:

master/buildbot/data/base.py

@@ -13,18 +13,32 @@
 #
 # Copyright Buildbot Team Members
 
-import re
+class ResourceType(object):
+    type = None
+    endpoints = []
+    keyFields = []
 
-class Endpoint(object):
+    def __init__(self, master):
+        self.master = master
 
-    # set the pathPattern to the pattern that should trigger this endpoint
-    pathPattern = None
+    def getEndpoints(self):
+        endpoints = self.endpoints[:]
+        for i in xrange(len(endpoints)):
+            ep = endpoints[i]
+            if not issubclass(ep, Endpoint):
+                raise TypeError("Not an Endpoint subclass")
+            endpoints[i] = ep(self.master)
+        return endpoints
 
-    # the mq topic corresponding to this path, with %(..)s safely substituted
-    # from the path kwargs; for more complex cases, override
-    # getSubscriptionTopic.
+    def produceEvent(self, msg, event):
+        routingKey = (self.type,) \
+             + tuple(str(msg[k]) for k in self.keyFields) \
+             + (event,)
+        self.master.mq.produce(routingKey, msg)
 
-    pathTopicTemplate = None
+
+class Endpoint(object):
+    pathPattern = None
 
     def __init__(self, master):
         self.master = master
@@ -35,33 +49,24 @@ def get(self, options, kwargs):
     def control(self, action, args, kwargs):
         raise NotImplementedError
 
-    def getSubscriptionTopic(self, options, kwargs):
-        if self.pathTopicTemplate:
-            if '%'  not in self.pathTopicTemplate:
-                return self.pathTopicTemplate
-            safekwargs = SafeDict(kwargs)
-            return self.pathTopicTemplate % safekwargs
-
-
-class SafeDict(object):
-    # utility class to allow %-substitution with the results not containing
-    # topic metacharacters (.*#)
-
-    def __init__(self, dict):
-        self.dict = dict
-
-    metacharacters_re = re.compile('[.*#]')
-    def __getitem__(self, k):
-        return self.metacharacters_re.sub('_', self.dict[k])
+    def startConsuming(self, callback, options, kwargs):
+        raise NotImplementedError
 
 
 class Link(object):
+    "A Link points to another resource, specified by path"
 
     __slots__ = [ 'path' ]
 
-    # a link to another resource, specified as a path
     def __init__(self, path):
         self.path = path
 
     def __repr__(self):
         return "Link(%r)" % (self.path,)
+
+
+def updateMethod(func):
+    """Decorate this resourceType instance as an update method, made available
+    at master.data.updates.$funcname"""
+    func.isUpdateMethod = True
+    return func

master/buildbot/data/changes.py

@@ -14,20 +14,13 @@
 # Copyright Buildbot Team Members
 
 from twisted.internet import defer
+from twisted.python import log
+from buildbot import util
 from buildbot.data import base, exceptions
-from buildbot.util import datetime2epoch
+from buildbot.process import metrics
+from buildbot.util import datetime2epoch, epoch2datetime
 
-def _fixChange(change):
-    # TODO: make these mods in the DB API
-    if change:
-        change = change.copy()
-        del change['is_dir']
-        change['when_timestamp'] = datetime2epoch(change['when_timestamp'])
-        change['link'] = base.Link(('change', str(change['changeid'])))
-    return change
-
-
-class Change(base.Endpoint):
+class ChangeEndpoint(base.Endpoint):
 
     pathPattern = ( 'change', 'i:changeid' )
 
@@ -37,10 +30,9 @@ def get(self, options, kwargs):
         return d
 
 
-class Changes(base.Endpoint):
+class ChangesEndpoint(base.Endpoint):
 
     pathPattern = ( 'change', )
-    pathTopicTemplate = 'change.#' # TODO: test
 
     def get(self, options, kwargs):
         try:
@@ -54,3 +46,77 @@ def sort(changes):
             changes.sort(key=lambda chdict : chdict['changeid'])
             return map(_fixChange, changes)
         return d
+
+
+class ChangeResourceType(base.ResourceType):
+
+    type = "change"
+    endpoints = [ ChangeEndpoint, ChangesEndpoint ]
+    keyFields = [ 'changeid' ]
+
+    @base.updateMethod
+    @defer.inlineCallbacks
+    def addChange(self, files=None, comments=None, author=None, revision=None,
+            when_timestamp=None, branch=None, category=None, revlink='',
+            properties={}, repository='', codebase=None, project='', src=None):
+        metrics.MetricCountEvent.log("added_changes", 1)
+
+        # TODO: temporary
+        properties = dict( (util.ascii2unicode(k), (v, u'Change'))
+                           for k, v in properties.iteritems() )
+        if src:
+            # create user object, returning a corresponding uid
+            uid = yield self.master.users.createUserObject(self.master,
+                    author, src)
+        else:
+            uid = None
+
+        change = {
+            'changeid': None, # not known yet
+            'author': unicode(author),
+            'files': map(unicode, files),
+            'comments': unicode(comments),
+            'revision': unicode(revision) if revision is not None else None,
+            'when_timestamp': datetime2epoch(when_timestamp),
+            'branch': unicode(branch) if branch is not None else None,
+            'category': unicode(category) if category is not None else None,
+            'revlink': unicode(revlink) if revlink is not None else None,
+            'properties': properties,
+            'repository': unicode(repository),
+            'project': unicode(project),
+            'codebase': None, # not set yet
+            # 'uid': uid, -- not in data API yet?
+        }
+
+        # if the codebase is default, and 
+        if codebase is None \
+                and self.master.config.codebaseGenerator is not None:
+            codebase = self.master.config.codebaseGenerator(change)
+            change['codebase'] = unicode(codebase)
+        else:
+            change['codebase'] = codebase or u''
+
+        # add the Change to the database and notify, converting briefly
+        # to database format, then back, and adding the changeid
+        del change['changeid']
+        change['when_timestamp'] = epoch2datetime(change['when_timestamp'])
+        changeid = yield self.master.db.changes.addChange(uid=uid, **change)
+        change['when_timestamp'] = datetime2epoch(change['when_timestamp'])
+        change['changeid'] = changeid
+        self.produceEvent(change, 'new')
+
+        # log, being careful to handle funny characters
+        msg = u"added change with revision %s to database" % (revision,)
+        log.msg(msg.encode('utf-8', 'replace'))
+
+        defer.returnValue(changeid)
+
+
+def _fixChange(change):
+    # TODO: make these mods in the DB API
+    if change:
+        change = change.copy()
+        del change['is_dir']
+        change['when_timestamp'] = datetime2epoch(change['when_timestamp'])
+        change['link'] = base.Link(('change', str(change['changeid'])))
+    return change