Add support for 'connection' concept

This is a large refactor and as small as I could feasibly make it
while keeping the tests working. I'll do the documentation and
touch ups in the next commit to make digesting easier.

Change-Id: Iac5083996a183d1d8a9b6cb8f70836f7c39ee910

tests/base.py

@@ -42,6 +42,8 @@ import statsd
 import testtools
 from git import GitCommandError
 
+import zuul.connection.gerrit
+import zuul.connection.smtp
 import zuul.scheduler
 import zuul.webapp
 import zuul.rpclistener
@@ -379,20 +381,20 @@ class FakeChange(object):
         self.reported += 1
 
 
-class FakeGerrit(object):
-    log = logging.getLogger("zuul.test.FakeGerrit")
+class FakeGerritConnection(zuul.connection.gerrit.GerritConnection):
+    log = logging.getLogger("zuul.test.FakeGerritConnection")
 
-    def __init__(self, hostname, username, port=29418, keyfile=None,
-                 changes_dbs={}, queues_dbs={}):
-        self.hostname = hostname
-        self.username = username
-        self.port = port
-        self.keyfile = keyfile
-        self.event_queue = queues_dbs.get(hostname, {})
+    def __init__(self, connection_name, connection_config,
+                 changes_db=None, queues_db=None):
+        super(FakeGerritConnection, self).__init__(connection_name,
+                                                   connection_config)
+
+        self.event_queue = queues_db
         self.fixture_dir = os.path.join(FIXTURE_DIR, 'gerrit')
         self.change_number = 0
-        self.changes = changes_dbs.get(hostname, {})
+        self.changes = changes_db
         self.queries = []
+        self.upstream_root = None
 
     def addFakeChange(self, project, branch, subject, status='NEW'):
         self.change_number += 1
@@ -402,15 +404,6 @@ class FakeGerrit(object):
         self.changes[self.change_number] = c
         return c
 
-    def addEvent(self, data):
-        return self.event_queue.put((time.time(), data))
-
-    def getEvent(self):
-        return self.event_queue.get()
-
-    def eventDone(self):
-        self.event_queue.task_done()
-
     def review(self, project, changeid, message, action):
         number, ps = changeid.split(',')
         change = self.changes[int(number)]
@@ -427,11 +420,11 @@ class FakeGerrit(object):
 
         for cat in ['CRVW', 'VRFY', 'APRV']:
             if cat in action:
-                change.addApproval(cat, action[cat], username=self.username)
+                change.addApproval(cat, action[cat], username=self.user)
 
         if 'label' in action:
             parts = action['label'].split('=')
-            change.addApproval(parts[0], parts[2], username=self.username)
+            change.addApproval(parts[0], parts[2], username=self.user)
 
         change.messages.append(message)
 
@@ -464,9 +457,12 @@ class FakeGerrit(object):
             l = [change.query() for change in self.changes.values()]
         return l
 
-    def startWatching(self, *args, **kw):
+    def _start_watcher_thread(self, *args, **kw):
         pass
 
+    def getGitUrl(self, project):
+        return os.path.join(self.upstream_root, project.name)
+
 
 class BuildHistory(object):
     def __init__(self, **kw):
@@ -500,19 +496,6 @@ class FakeURLOpener(object):
         return ret
 
 
-class FakeGerritSource(zuul.source.gerrit.GerritSource):
-    name = 'gerrit'
-
-    def __init__(self, upstream_root, *args):
-        super(FakeGerritSource, self).__init__(*args)
-        self.upstream_root = upstream_root
-        self.replication_timeout = 1.5
-        self.replication_retry_interval = 0.5
-
-    def getGitUrl(self, project):
-        return os.path.join(self.upstream_root, project.name)
-
-
 class FakeStatsd(threading.Thread):
     def __init__(self):
         threading.Thread.__init__(self)
@@ -898,7 +881,6 @@ class ZuulTestCase(BaseTestCase):
             shutil.rmtree(self.test_root)
         os.makedirs(self.test_root)
         os.makedirs(self.upstream_root)
-        os.makedirs(self.git_root)
 
         # Make per test copy of Configuration.
         self.setup_config()
@@ -942,15 +924,22 @@ class ZuulTestCase(BaseTestCase):
         self.worker.addServer('127.0.0.1', self.gearman_server.port)
         self.gearman_server.worker = self.worker
 
-        self.merge_server = zuul.merger.server.MergeServer(self.config)
-        self.merge_server.start()
+        zuul.source.gerrit.GerritSource.replication_timeout = 1.5
+        zuul.source.gerrit.GerritSource.replication_retry_interval = 0.5
+        zuul.connection.gerrit.GerritEventConnector.delay = 0.0
 
-        self.sched = zuul.scheduler.Scheduler()
+        self.sched = zuul.scheduler.Scheduler(self.config)
 
         self.useFixture(fixtures.MonkeyPatch('swiftclient.client.Connection',
                                              FakeSwiftClientConnection))
         self.swift = zuul.lib.swift.Swift(self.config)
 
+        # Set up connections and give out the default gerrit for testing
+        self.configure_connections()
+        self.sched.registerConnections(self.connections)
+        self.fake_gerrit = self.connections['gerrit']
+        self.fake_gerrit.upstream_root = self.upstream_root
+
         def URLOpenerFactory(*args, **kw):
             if isinstance(args[0], urllib2.Request):
                 return old_urlopen(*args, **kw)
@@ -960,30 +949,9 @@ class ZuulTestCase(BaseTestCase):
         old_urlopen = urllib2.urlopen
         urllib2.urlopen = URLOpenerFactory
 
-        self.smtp_messages = []
-
-        def FakeSMTPFactory(*args, **kw):
-            args = [self.smtp_messages] + list(args)
-            return FakeSMTP(*args, **kw)
-
-        # Set a changes database so multiple FakeGerrit's can report back to
-        # a virtual canonical database given by the configured hostname
-        self.gerrit_queues_dbs = {
-            self.config.get('gerrit', 'server'): Queue.Queue()
-        }
-        self.gerrit_changes_dbs = {
-            self.config.get('gerrit', 'server'): {}
-        }
-
-        def FakeGerritFactory(*args, **kw):
-            kw['changes_dbs'] = self.gerrit_changes_dbs
-            kw['queues_dbs'] = self.gerrit_queues_dbs
-            return FakeGerrit(*args, **kw)
-
-        self.useFixture(fixtures.MonkeyPatch('zuul.lib.gerrit.Gerrit',
-                                             FakeGerritFactory))
-
-        self.useFixture(fixtures.MonkeyPatch('smtplib.SMTP', FakeSMTPFactory))
+        self.merge_server = zuul.merger.server.MergeServer(self.config,
+                                                           self.connections)
+        self.merge_server.start()
 
         self.launcher = zuul.launcher.gearman.Gearman(self.config, self.sched,
                                                       self.swift)
@@ -993,13 +961,6 @@ class ZuulTestCase(BaseTestCase):
         self.sched.setLauncher(self.launcher)
         self.sched.setMerger(self.merge_client)
 
-        self.register_sources()
-        self.fake_gerrit = self.gerrit_source.gerrit
-        self.fake_gerrit.upstream_root = self.upstream_root
-
-        self.register_triggers()
-        self.register_reporters()
-
         self.webapp = zuul.webapp.WebApp(self.sched, port=0)
         self.rpc = zuul.rpclistener.RPCListener(self.config, self.sched)
 
@@ -1016,37 +977,67 @@ class ZuulTestCase(BaseTestCase):
         self.addCleanup(self.assertFinalState)
         self.addCleanup(self.shutdown)
 
-    def register_sources(self):
-        # Register the available sources
-        self.gerrit_source = FakeGerritSource(
-            self.upstream_root, self.config, self.sched)
-        self.gerrit_source.replication_timeout = 1.5
-        self.gerrit_source.replication_retry_interval = 0.5
+    def configure_connections(self):
+        # Register connections from the config
+        self.smtp_messages = []
 
-        self.sched.registerSource(self.gerrit_source)
+        def FakeSMTPFactory(*args, **kw):
+            args = [self.smtp_messages] + list(args)
+            return FakeSMTP(*args, **kw)
 
-    def register_triggers(self):
-        # Register the available triggers
-        self.gerrit_trigger = zuul.trigger.gerrit.GerritTrigger(
-            self.fake_gerrit, self.config, self.sched, self.gerrit_source)
-        self.gerrit_trigger.gerrit_connector.delay = 0.0
+        self.useFixture(fixtures.MonkeyPatch('smtplib.SMTP', FakeSMTPFactory))
 
-        self.sched.registerTrigger(self.gerrit_trigger)
-        self.timer = zuul.trigger.timer.TimerTrigger(self.config, self.sched)
-        self.sched.registerTrigger(self.timer)
-        self.zuultrigger = zuul.trigger.zuultrigger.ZuulTrigger(self.config,
-                                                                self.sched)
-        self.sched.registerTrigger(self.zuultrigger)
+        # Set a changes database so multiple FakeGerrit's can report back to
+        # a virtual canonical database given by the configured hostname
+        self.gerrit_changes_dbs = {}
+        self.gerrit_queues_dbs = {}
+        self.connections = {}
 
-    def register_reporters(self):
-        # Register the available reporters
-        self.sched.registerReporter(
-            zuul.reporter.gerrit.GerritReporter(self.fake_gerrit))
-        self.smtp_reporter = zuul.reporter.smtp.SMTPReporter(
-            self.config.get('smtp', 'default_from'),
-            self.config.get('smtp', 'default_to'),
-            self.config.get('smtp', 'server'))
-        self.sched.registerReporter(self.smtp_reporter)
+        for section_name in self.config.sections():
+            con_match = re.match(r'^connection ([\'\"]?)(.*)(\1)$',
+                                 section_name, re.I)
+            if not con_match:
+                continue
+            con_name = con_match.group(2)
+            con_config = dict(self.config.items(section_name))
+
+            if 'driver' not in con_config:
+                raise Exception("No driver specified for connection %s."
+                                % con_name)
+
+            con_driver = con_config['driver']
+
+            # TODO(jhesketh): load the required class automatically
+            if con_driver == 'gerrit':
+                self.gerrit_changes_dbs[con_name] = {}
+                self.gerrit_queues_dbs[con_name] = Queue.Queue()
+                self.connections[con_name] = FakeGerritConnection(
+                    con_name, con_config,
+                    changes_db=self.gerrit_changes_dbs[con_name],
+                    queues_db=self.gerrit_queues_dbs[con_name]
+                )
+            elif con_driver == 'smtp':
+                self.connections[con_name] = \
+                    zuul.connection.smtp.SMTPConnection(con_name, con_config)
+            else:
+                raise Exception("Unknown driver, %s, for connection %s"
+                                % (con_config['driver'], con_name))
+
+        # If the [gerrit] or [smtp] sections still exist, load them in as a
+        # connection named 'gerrit' or 'smtp' respectfully
+
+        if 'gerrit' in self.config.sections():
+            self.gerrit_changes_dbs['gerrit'] = {}
+            self.gerrit_queues_dbs['gerrit'] = Queue.Queue()
+            self.connections['gerrit'] = FakeGerritConnection(
+                '_legacy_gerrit', dict(self.config.items('gerrit')),
+                changes_db=self.gerrit_changes_dbs['gerrit'],
+                queues_db=self.gerrit_queues_dbs['gerrit'])
+
+        if 'smtp' in self.config.sections():
+            self.connections['smtp'] = \
+                zuul.connection.smtp.SMTPConnection(
+                    '_legacy_smtp', dict(self.config.items('smtp')))
 
     def setup_config(self):
         """Per test config object. Override to set different config."""
@@ -1074,8 +1065,6 @@ class ZuulTestCase(BaseTestCase):
         self.merge_server.join()
         self.merge_client.stop()
         self.worker.shutdown()
-        self.gerrit_trigger.stop()
-        self.timer.stop()
         self.sched.stop()
         self.sched.join()
         self.statsd.stop()

tests/fixtures/layouts/bad_gerrit_missing.yaml

@@ -0,0 +1,18 @@
+pipelines:
+  - name: check
+    manager: IndependentPipelineManager
+    trigger:
+      not_gerrit:
+        - event: patchset-created
+    success:
+      review_gerrit:
+        verified: 1
+    failure:
+      review_gerrit:
+        verified: -1
+
+projects:
+  - name: test-org/test
+    check:
+      - test-merge
+      - test-test

tests/fixtures/layouts/bad_merge_failure.yaml

@@ -2,13 +2,13 @@ pipelines:
   - name: check
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
     success:
-      gerrit:
+      review_gerrit:
         verified: 1
     failure:
-      gerrit:
+      review_gerrit:
         verified: -1
     # merge-failure-message needs a string.
     merge-failure-message:
@@ -17,20 +17,20 @@ pipelines:
     manager: DependentPipelineManager
     failure-message: Build failed.  For information on how to proceed, see http://wiki.example.org/Test_Failures
     trigger:
-      gerrit:
+      review_gerrit:
         - event: comment-added
           approval:
             - approved: 1
     success:
-      gerrit:
+      review_gerrit:
         verified: 2
         submit: true
     failure:
-      gerrit:
+      review_gerrit:
         verified: -2
     merge-failure:
     start:
-      gerrit:
+      review_gerrit:
         verified: 0
     precedence: high
 

tests/fixtures/layouts/bad_misplaced_ref.yaml

@@ -2,7 +2,7 @@ pipelines:
   - name: 'check'
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
           ref: /some/ref/path
 

tests/fixtures/layouts/bad_pipelines5.yaml

@@ -2,7 +2,7 @@ pipelines:
   - name: check
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         # event is a required item but it is missing.
         - approval:
             - approved: 1

tests/fixtures/layouts/bad_pipelines6.yaml

@@ -2,7 +2,7 @@ pipelines:
   - name: check
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: comment-added
           # approved is not a valid entry. Should be approval.
           approved: 1

tests/fixtures/layouts/bad_reject.yaml

@@ -17,5 +17,5 @@ pipelines:
         - code-review: [-1, -2]
           username: core-person
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created

tests/fixtures/layouts/bad_swift.yaml

@@ -2,13 +2,13 @@ pipelines:
   - name: check
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
     success:
-      gerrit:
+      review_gerrit:
         verified: 1
     failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
 jobs:

tests/fixtures/layouts/bad_template1.yaml

@@ -4,7 +4,7 @@ pipelines:
   - name: 'check'
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
 
 project-templates:

tests/fixtures/layouts/bad_template2.yaml

@@ -4,7 +4,7 @@ pipelines:
   - name: 'check'
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
 
 project-templates:

tests/fixtures/layouts/good_connections1.conf

@@ -0,0 +1,42 @@
+[gearman]
+server=127.0.0.1
+
+[zuul]
+layout_config=layout.yaml
+url_pattern=http://logs.example.com/{change.number}/{change.patchset}/{pipeline.name}/{job.name}/{build.number}
+job_name_in_report=true
+
+[merger]
+git_dir=/tmp/zuul-test/git
+git_user_email=zuul@example.com
+git_user_name=zuul
+zuul_url=http://zuul.example.com/p
+
+[swift]
+authurl=https://identity.api.example.org/v2.0/
+user=username
+key=password
+tenant_name=" "
+
+default_container=logs
+region_name=EXP
+logserver_prefix=http://logs.example.org/server.app/
+
+[connection review_gerrit]
+driver=gerrit
+server=review.example.com
+user=jenkins
+sshkey=none
+
+[connection other_gerrit]
+driver=gerrit
+server=review2.example.com
+user=jenkins2
+sshkey=none
+
+[connection my_smtp]
+driver=smtp
+server=localhost
+port=25
+default_from=zuul@example.com
+default_to=you@example.com

tests/fixtures/layouts/good_connections1.yaml

@@ -0,0 +1,18 @@
+pipelines:
+  - name: check
+    manager: IndependentPipelineManager
+    source: review_gerrit
+    trigger:
+      review_gerrit:
+        - event: patchset-created
+    success:
+      review_gerrit:
+        verified: 1
+    failure:
+      other_gerrit:
+        verified: -1
+
+projects:
+  - name: org/project
+    check:
+      - project-check

tests/fixtures/layouts/good_layout.yaml

@@ -8,7 +8,7 @@ pipelines:
       open: True
       current-patchset: True
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
         - event: comment-added
           require-approval:
@@ -17,16 +17,16 @@ pipelines:
           approval:
             - workflow: 1
     success:
-      gerrit:
+      review_gerrit:
         verified: 1
     failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
   - name: post
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: ref-updated
           ref: ^(?!refs/).*$
           ignore-deletes: True
@@ -46,32 +46,32 @@ pipelines:
       approval:
         - code-review: [-1, -2]
     trigger:
-      gerrit:
+      review_gerrit:
         - event: comment-added
           approval:
             - approved: 1
     start:
-      gerrit:
+      review_gerrit:
         verified: 0
     success:
-      gerrit:
+      review_gerrit:
         verified: 2
         code-review: 1
         submit: true
     failure:
-      gerrit:
+      review_gerrit:
         verified: -2
         workinprogress: true
 
   - name: merge-check
     manager: IndependentPipelineManager
-    source: gerrit
+    source: review_gerrit
     ignore-dependencies: true
     trigger:
       zuul:
         - event: project-change-merged
     merge-failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
 jobs:

tests/fixtures/layouts/good_merge_failure.yaml

@@ -3,47 +3,47 @@ pipelines:
     manager: IndependentPipelineManager
     merge-failure-message: "Could not merge the change. Please rebase..."
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
     success:
-      gerrit:
+      review_gerrit:
         verified: 1
     failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
   - name: post
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: ref-updated
           ref: ^(?!refs/).*$
     merge-failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
   - name: gate
     manager: DependentPipelineManager
     failure-message: Build failed.  For information on how to proceed, see http://wiki.example.org/Test_Failures
     trigger:
-      gerrit:
+      review_gerrit:
         - event: comment-added
           approval:
             - approved: 1
     success:
-      gerrit:
+      review_gerrit:
         verified: 2
         submit: true
     failure:
-      gerrit:
+      review_gerrit:
         verified: -2
     merge-failure:
-      gerrit:
+      review_gerrit:
         verified: -1
-      smtp:
+      my_smtp:
         to: you@example.com
     start:
-      gerrit:
+      review_gerrit:
         verified: 0
     precedence: high
 

tests/fixtures/layouts/good_require_approvals.yaml

@@ -5,7 +5,7 @@ pipelines:
   - name: check
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: comment-added
           require-approval:
             - username: jenkins
@@ -23,10 +23,10 @@ pipelines:
               username: jenkins
               email: jenkins@example.com
     success:
-      gerrit:
+      review_gerrit:
         verified: 1
     failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
 projects:

tests/fixtures/layouts/good_swift.yaml

@@ -2,13 +2,13 @@ pipelines:
   - name: check
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
     success:
-      gerrit:
+      review_gerrit:
         verified: 1
     failure:
-      gerrit:
+      review_gerrit:
         verified: -1
 
 jobs:

tests/fixtures/layouts/good_template1.yaml

@@ -2,7 +2,7 @@ pipelines:
   - name: 'check'
     manager: IndependentPipelineManager
     trigger:
-      gerrit:
+      review_gerrit:
         - event: patchset-created
 
 project-templates:

tests/fixtures/layouts/zuul_default.conf

@@ -0,0 +1,36 @@
+[gearman]
+server=127.0.0.1
+
+[zuul]
+layout_config=layout.yaml
+url_pattern=http://logs.example.com/{change.number}/{change.patchset}/{pipeline.name}/{job.name}/{build.number}
+job_name_in_report=true
+
+[merger]
+git_dir=/tmp/zuul-test/git
+git_user_email=zuul@example.com
+git_user_name=zuul
+zuul_url=http://zuul.example.com/p
+
+[swift]
+authurl=https://identity.api.example.org/v2.0/
+user=username
+key=password
+tenant_name=" "
+
+default_container=logs
+region_name=EXP
+logserver_prefix=http://logs.example.org/server.app/
+
+[connection review_gerrit]
+driver=gerrit
+server=review.example.com
+user=jenkins
+sshkey=none
+
+[connection my_smtp]
+driver=smtp
+server=localhost
+port=25
+default_from=zuul@example.com
+default_to=you@example.com

tests/fixtures/zuul.conf

@@ -1,11 +1,6 @@
 [gearman]
 server=127.0.0.1
 
-[gerrit]
-server=review.example.com
-user=jenkins
-sshkey=none
-
 [zuul]
 layout_config=layout.yaml
 url_pattern=http://logs.example.com/{change.number}/{change.patchset}/{pipeline.name}/{job.name}/{build.number}
@@ -17,12 +12,6 @@ git_user_email=zuul@example.com
 git_user_name=zuul
 zuul_url=http://zuul.example.com/p
 
-[smtp]
-server=localhost
-port=25
-default_from=zuul@example.com
-default_to=you@example.com
-
 [swift]
 authurl=https://identity.api.example.org/v2.0/
 user=username
@@ -32,3 +21,16 @@ tenant_name=" "
 default_container=logs
 region_name=EXP
 logserver_prefix=http://logs.example.org/server.app/
+
+[connection gerrit]
+driver=gerrit
+server=review.example.com
+user=jenkins
+sshkey=none
+
+[connection smtp]
+driver=smtp
+server=localhost
+port=25
+default_from=zuul@example.com
+default_to=you@example.com

tests/test_connection.py

@@ -0,0 +1,26 @@
+# Copyright 2014 Rackspace Australia
+#
+# 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 logging
+import testtools
+
+import zuul.connection.gerrit
+
+
+class TestGerritConnection(testtools.TestCase):
+    log = logging.getLogger("zuul.test_connection")
+
+    def test_driver_name(self):
+        self.assertEqual('gerrit',
+                         zuul.connection.gerrit.GerritConnection.driver_name)

tests/test_gerrit.py

@@ -21,7 +21,7 @@ except ImportError:
     import mock
 
 from tests.base import BaseTestCase
-from zuul.lib.gerrit import Gerrit
+from zuul.connection.gerrit import GerritConnection
 
 FIXTURE_DIR = os.path.join(os.path.dirname(__file__), 'fixtures/gerrit')
 
@@ -46,9 +46,13 @@ def read_fixtures(files):
 
 class TestGerrit(BaseTestCase):
 
-    @mock.patch('zuul.lib.gerrit.Gerrit._ssh')
+    @mock.patch('zuul.connection.gerrit.GerritConnection._ssh')
     def run_query(self, files, expected_patches, _ssh_mock):
-        gerrit = Gerrit('localhost', 'user')
+        gerrit_config = {
+            'user': 'gerrit',
+            'server': 'localhost',
+        }
+        gerrit = GerritConnection('review_gerrit', gerrit_config)
 
         calls, values = read_fixtures(files)
         _ssh_mock.side_effect = values

tests/test_layoutvalidator.py

@@ -14,6 +14,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import ConfigParser
 import os
 import re
 
@@ -22,6 +23,7 @@ import voluptuous
 import yaml
 
 import zuul.layoutvalidator
+import zuul.lib.connections
 
 FIXTURE_DIR = os.path.join(os.path.dirname(__file__),
                            'fixtures')
@@ -38,19 +40,31 @@ class TestLayoutValidator(testtools.TestCase):
             if not m:
                 continue
             print fn
+
+            # Load any .conf file by the same name but .conf extension.
+            config_file = ("%s.conf" %
+                           os.path.join(FIXTURE_DIR, 'layouts',
+                                        fn.split('.yaml')[0]))
+            if not os.path.isfile(config_file):
+                config_file = os.path.join(FIXTURE_DIR, 'layouts',
+                                           'zuul_default.conf')
+            config = ConfigParser.ConfigParser()
+            config.read(config_file)
+            connections = zuul.lib.connections.configure_connections(config)
+
             layout = os.path.join(FIXTURE_DIR, 'layouts', fn)
             data = yaml.load(open(layout))
             validator = zuul.layoutvalidator.LayoutValidator()
             if m.group(1) == 'good':
                 try:
-                    validator.validate(data)
+                    validator.validate(data, connections)
                 except voluptuous.Invalid as e:
                     raise Exception(
                         'Unexpected YAML syntax error in %s:\n  %s' %
                         (fn, str(e)))
             else:
                 try:
-                    validator.validate(data)
+                    validator.validate(data, connections)
                     raise Exception("Expected a YAML syntax error in %s." %
                                     fn)
                 except voluptuous.Invalid as e:

tests/test_reporter.py

@@ -26,7 +26,7 @@ class TestSMTPReporter(testtools.TestCase):
 
     def test_reporter_abc(self):
         # We only need to instantiate a class for this
-        reporter = zuul.reporter.smtp.SMTPReporter('', '')  # noqa
+        reporter = zuul.reporter.smtp.SMTPReporter({})  # noqa
 
     def test_reporter_name(self):
         self.assertEqual('smtp', zuul.reporter.smtp.SMTPReporter.name)

tests/test_scheduler.py

@@ -61,7 +61,7 @@ jobs:
     """.strip()
         data = yaml.load(job_yaml)
         config_job = data.get('jobs')[0]
-        sched = zuul.scheduler.Scheduler()
+        sched = zuul.scheduler.Scheduler({})
         cm = zuul.change_matcher
         expected = cm.MatchAny([
             cm.MatchAll([
@@ -762,9 +762,9 @@ class TestScheduler(ZuulTestCase):
         self.fake_gerrit.addEvent(B.addApproval('APRV', 1))
         self.waitUntilSettled()
 
-        self.log.debug("len %s" % self.gerrit_source._change_cache.keys())
+        self.log.debug("len %s" % self.fake_gerrit._change_cache.keys())
         # there should still be changes in the cache
-        self.assertNotEqual(len(self.gerrit_source._change_cache.keys()), 0)
+        self.assertNotEqual(len(self.fake_gerrit._change_cache.keys()), 0)
 
         self.worker.hold_jobs_in_build = False
         self.worker.release()
@@ -1469,7 +1469,7 @@ class TestScheduler(ZuulTestCase):
         "Test that the merger works with large changes after a repack"
         # https://bugs.launchpad.net/zuul/+bug/1078946
         # This test assumes the repo is already cloned; make sure it is
-        url = self.sched.sources['gerrit'].getGitUrl(
+        url = self.fake_gerrit.getGitUrl(
             self.sched.layout.projects['org/project1'])
         self.merge_server.merger.addProject('org/project1', url)
         A = self.fake_gerrit.addFakeChange('org/project1', 'master', 'A')
@@ -2164,7 +2164,8 @@ class TestScheduler(ZuulTestCase):
 
     def test_test_config(self):
         "Test that we can test the config"
-        self.sched.testConfig(self.config.get('zuul', 'layout_config'))
+        self.sched.testConfig(self.config.get('zuul', 'layout_config'),
+                              self.connections)
 
     def test_build_description(self):
         "Test that build descriptions update"

tests/test_trigger.py

@@ -23,8 +23,7 @@ class TestGerritTrigger(testtools.TestCase):
 
     def test_trigger_abc(self):
         # We only need to instantiate a class for this
-        trigger = zuul.trigger.gerrit.GerritTrigger(None, None, None,  # noqa
-                                                    None)
+        zuul.trigger.gerrit.GerritTrigger({})
 
     def test_trigger_name(self):
         self.assertEqual('gerrit', zuul.trigger.gerrit.GerritTrigger.name)
@@ -35,7 +34,7 @@ class TestTimerTrigger(testtools.TestCase):
 
     def test_trigger_abc(self):
         # We only need to instantiate a class for this
-        trigger = zuul.trigger.timer.TimerTrigger(None, None)  # noqa
+        zuul.trigger.timer.TimerTrigger({})
 
     def test_trigger_name(self):
         self.assertEqual('timer', zuul.trigger.timer.TimerTrigger.name)
@@ -46,7 +45,7 @@ class TestZuulTrigger(testtools.TestCase):
 
     def test_trigger_abc(self):
         # We only need to instantiate a class for this
-        trigger = zuul.trigger.zuultrigger.ZuulTrigger(None, None)  # noqa
+        zuul.trigger.zuultrigger.ZuulTrigger({})
 
     def test_trigger_name(self):
         self.assertEqual('zuul', zuul.trigger.zuultrigger.ZuulTrigger.name)

zuul/cmd/__init__.py

@@ -26,7 +26,9 @@ import traceback
 
 yappi = extras.try_import('yappi')
 
-# No zuul imports here because they pull in paramiko which must not be
+import zuul.lib.connections
+
+# Do not import modules that will pull in paramiko which must not be
 # imported until after the daemonization.
 # https://github.com/paramiko/paramiko/issues/59
 # Similar situation with gear and statsd.
@@ -59,6 +61,7 @@ class ZuulApp(object):
     def __init__(self):
         self.args = None
         self.config = None
+        self.connections = {}
 
     def _get_version(self):
         from zuul.version import version_info as zuul_version_info
@@ -86,3 +89,7 @@ class ZuulApp(object):
             logging.config.fileConfig(fp)
         else:
             logging.basicConfig(level=logging.DEBUG)
+
+    def configure_connections(self):
+        self.connections = zuul.lib.connections.configure_connections(
+            self.config)

zuul/cmd/merger.py

@@ -58,7 +58,8 @@ class Merger(zuul.cmd.ZuulApp):
 
         self.setup_logging('merger', 'log_config')
 
-        self.merger = zuul.merger.server.MergeServer(self.config)
+        self.merger = zuul.merger.server.MergeServer(self.config,
+                                                     self.connections)
         self.merger.start()
 
         signal.signal(signal.SIGUSR1, self.exit_handler)
@@ -76,6 +77,7 @@ def main():
     server.parse_arguments()
 
     server.read_config()
+    server.configure_connections()
 
     if server.config.has_option('zuul', 'state_dir'):
         state_dir = os.path.expanduser(server.config.get('zuul', 'state_dir'))

zuul/cmd/server.py

@@ -60,9 +60,13 @@ class Server(zuul.cmd.ZuulApp):
 
     def reconfigure_handler(self, signum, frame):
         signal.signal(signal.SIGHUP, signal.SIG_IGN)
+        self.log.debug("Reconfiguration triggered")
+        self.sched.stopConnections()
         self.read_config()
         self.setup_logging('zuul', 'log_config')
         try:
+            self.configure_connections()
+            self.sched.registerConnections(self.connections)
             self.sched.reconfigure(self.config)
         except Exception:
             self.log.exception("Reconfiguration failed:")
@@ -85,14 +89,11 @@ class Server(zuul.cmd.ZuulApp):
         import zuul.trigger.gerrit
 
         logging.basicConfig(level=logging.DEBUG)
-        self.sched = zuul.scheduler.Scheduler()
-        self.sched.registerReporter(None, 'gerrit')
-        self.sched.registerReporter(None, 'smtp')
-        self.sched.registerTrigger(None, 'gerrit')
-        self.sched.registerTrigger(None, 'timer')
-        self.sched.registerTrigger(None, 'zuul')
+        self.sched = zuul.scheduler.Scheduler(self.config)
+        self.configure_connections()
         layout = self.sched.testConfig(self.config.get('zuul',
-                                                       'layout_config'))
+                                                       'layout_config'),
+                                       self.connections)
         if not job_list_path:
             return False
 
@@ -144,51 +145,6 @@ class Server(zuul.cmd.ZuulApp):
         if self.gear_server_pid:
             os.kill(self.gear_server_pid, signal.SIGKILL)
 
-    def register_sources(self):
-        # Register the available sources
-        # See comment at top of file about zuul imports
-        import zuul.source.gerrit
-        self.gerrit_source = zuul.source.gerrit.GerritSource(self.config,
-                                                             self.sched)
-
-        self.sched.registerSource(self.gerrit_source)
-
-    def register_triggers(self):
-        # Register the available triggers
-        # See comment at top of file about zuul imports
-        import zuul.trigger.gerrit
-        import zuul.trigger.timer
-        import zuul.trigger.zuultrigger
-        self.gerrit_trigger = zuul.trigger.gerrit.GerritTrigger(
-            self.gerrit, self.config, self.sched, self.gerrit_source)
-        timer = zuul.trigger.timer.TimerTrigger(self.config, self.sched)
-        zuultrigger = zuul.trigger.zuultrigger.ZuulTrigger(
-            self.config, self.sched)
-
-        self.sched.registerTrigger(self.gerrit_trigger)
-        self.sched.registerTrigger(timer)
-        self.sched.registerTrigger(zuultrigger)
-
-    def register_reporters(self):
-        # Register the available reporters
-        # See comment at top of file about zuul imports
-        import zuul.reporter.gerrit
-        import zuul.reporter.smtp
-        gerrit_reporter = zuul.reporter.gerrit.GerritReporter(self.gerrit)
-        smtp_reporter = zuul.reporter.smtp.SMTPReporter(
-            self.config.get('smtp', 'default_from')
-            if self.config.has_option('smtp', 'default_from') else 'zuul',
-            self.config.get('smtp', 'default_to')
-            if self.config.has_option('smtp', 'default_to') else 'zuul',
-            self.config.get('smtp', 'server')
-            if self.config.has_option('smtp', 'server') else 'localhost',
-            self.config.get('smtp', 'port')
-            if self.config.has_option('smtp', 'port') else 25
-        )
-
-        self.sched.registerReporter(gerrit_reporter)
-        self.sched.registerReporter(smtp_reporter)
-
     def main(self):
         # See comment at top of file about zuul imports
         import zuul.scheduler
@@ -206,7 +162,8 @@ class Server(zuul.cmd.ZuulApp):
         self.setup_logging('zuul', 'log_config')
         self.log = logging.getLogger("zuul.Server")
 
-        self.sched = zuul.scheduler.Scheduler()
+        self.sched = zuul.scheduler.Scheduler(self.config)
+        # TODO(jhesketh): Move swift into a connection?
         self.swift = zuul.lib.swift.Swift(self.config)
 
         gearman = zuul.launcher.gearman.Gearman(self.config, self.sched,
@@ -220,17 +177,13 @@ class Server(zuul.cmd.ZuulApp):
         webapp = zuul.webapp.WebApp(self.sched, cache_expiry=cache_expiry)
         rpc = zuul.rpclistener.RPCListener(self.config, self.sched)
 
+        self.configure_connections()
         self.sched.setLauncher(gearman)
         self.sched.setMerger(merger)
-        self.register_sources()
-        # TODO(jhesketh): Use connections instead of grabbing the gerrit lib
-        #                 from the source
-        self.gerrit = self.gerrit_source.gerrit
-        self.register_triggers()
-        self.register_reporters()
 
         self.log.info('Starting scheduler')
         self.sched.start()
+        self.sched.registerConnections(self.connections)
         self.sched.reconfigure(self.config)
         self.sched.resume()
         self.log.info('Starting Webapp')

zuul/connection/__init__.py

@@ -0,0 +1,53 @@
+# Copyright 2014 Rackspace Australia
+#
+# 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 six
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BaseConnection(object):
+    """Base class for connections.
+
+    A connection is a shared object that sources, triggers and reporters can
+    use to speak with a remote API without needing to establish a new
+    connection each time or without having to authenticate each time.
+
+    Multiple instances of the same connection may exist with different
+    credentials, for example, thus allowing for different pipelines to operate
+    on different Gerrit installations or post back as a different user etc.
+
+    Connections can implement their own public methods. Required connection
+    methods are validated by the {trigger, source, reporter} they are loaded
+    into. For example, a trigger will likely require some kind of query method
+    while a reporter may need a review method."""
+
+    def __init__(self, connection_name, connection_config):
+        # connection_name is the name given to this connection in zuul.ini
+        # connection_config is a dictionary of config_section from zuul.ini for
+        # this connection.
+        # __init__ shouldn't make the actual connection in case this connection
+        # isn't used in the layout.
+        self.connection_name = connection_name
+        self.connection_config = connection_config
+
+    def onLoad(self):
+        pass
+
+    def onStop(self):
+        pass
+
+    def registerScheduler(self, sched):
+        self.sched = sched

zuul/connection/gerrit.py

@@ -0,0 +1,465 @@
+# Copyright 2011 OpenStack, LLC.
+# Copyright 2012 Hewlett-Packard Development Company, L.P.
+#
+# 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 threading
+import select
+import json
+import time
+from six.moves import queue as Queue
+import paramiko
+import logging
+import pprint
+import voluptuous as v
+import urllib2
+
+from zuul.connection import BaseConnection
+from zuul.model import TriggerEvent
+
+
+class GerritEventConnector(threading.Thread):
+    """Move events from Gerrit to the scheduler."""
+
+    log = logging.getLogger("zuul.GerritEventConnector")
+    delay = 5.0
+
+    def __init__(self, connection):
+        super(GerritEventConnector, self).__init__()
+        self.daemon = True
+        self.connection = connection
+        self._stopped = False
+
+    def stop(self):
+        self._stopped = True
+        self.connection.addEvent(None)
+
+    def _handleEvent(self):
+        ts, data = self.connection.getEvent()
+        if self._stopped:
+            self.connection.eventDone()
+            return
+        # Gerrit can produce inconsistent data immediately after an
+        # event, So ensure that we do not deliver the event to Zuul
+        # until at least a certain amount of time has passed.  Note
+        # that if we receive several events in succession, we will
+        # only need to delay for the first event.  In essence, Zuul
+        # should always be a constant number of seconds behind Gerrit.
+        now = time.time()
+        time.sleep(max((ts + self.delay) - now, 0.0))
+        event = TriggerEvent()
+        event.type = data.get('type')
+        event.trigger_name = 'gerrit'
+        change = data.get('change')
+        if change:
+            event.project_name = change.get('project')
+            event.branch = change.get('branch')
+            event.change_number = change.get('number')
+            event.change_url = change.get('url')
+            patchset = data.get('patchSet')
+            if patchset:
+                event.patch_number = patchset.get('number')
+                event.refspec = patchset.get('ref')
+            event.approvals = data.get('approvals', [])
+            event.comment = data.get('comment')
+        refupdate = data.get('refUpdate')
+        if refupdate:
+            event.project_name = refupdate.get('project')
+            event.ref = refupdate.get('refName')
+            event.oldrev = refupdate.get('oldRev')
+            event.newrev = refupdate.get('newRev')
+        # Map the event types to a field name holding a Gerrit
+        # account attribute. See Gerrit stream-event documentation
+        # in cmd-stream-events.html
+        accountfield_from_type = {
+            'patchset-created': 'uploader',
+            'draft-published': 'uploader',  # Gerrit 2.5/2.6
+            'change-abandoned': 'abandoner',
+            'change-restored': 'restorer',
+            'change-merged': 'submitter',
+            'merge-failed': 'submitter',  # Gerrit 2.5/2.6
+            'comment-added': 'author',
+            'ref-updated': 'submitter',
+            'reviewer-added': 'reviewer',  # Gerrit 2.5/2.6
+        }
+        try:
+            event.account = data.get(accountfield_from_type[event.type])
+        except KeyError:
+            self.log.error("Received unrecognized event type '%s' from Gerrit.\
+                    Can not get account information." % event.type)
+            event.account = None
+
+        if (event.change_number and
+            self.connection.sched.getProject(event.project_name)):
+            # Mark the change as needing a refresh in the cache
+            event._needs_refresh = True
+
+        self.connection.sched.addEvent(event)
+        self.connection.eventDone()
+
+    def run(self):
+        while True:
+            if self._stopped:
+                return
+            try:
+                self._handleEvent()
+            except:
+                self.log.exception("Exception moving Gerrit event:")
+
+
+class GerritWatcher(threading.Thread):
+    log = logging.getLogger("gerrit.GerritWatcher")
+
+    def __init__(self, gerrit_connection, username, hostname, port=29418,
+                 keyfile=None):
+        threading.Thread.__init__(self)
+        self.username = username
+        self.keyfile = keyfile
+        self.hostname = hostname
+        self.port = port
+        self.gerrit_connection = gerrit_connection
+        self._stopped = False
+
+    def _read(self, fd):
+        l = fd.readline()
+        data = json.loads(l)
+        self.log.debug("Received data from Gerrit event stream: \n%s" %
+                       pprint.pformat(data))
+        self.gerrit_connection.addEvent(data)
+
+    def _listen(self, stdout, stderr):
+        poll = select.poll()
+        poll.register(stdout.channel)
+        while not self._stopped:
+            ret = poll.poll()
+            for (fd, event) in ret:
+                if fd == stdout.channel.fileno():
+                    if event == select.POLLIN:
+                        self._read(stdout)
+                    else:
+                        raise Exception("event on ssh connection")
+
+    def _run(self):
+        try:
+            client = paramiko.SSHClient()
+            client.load_system_host_keys()
+            client.set_missing_host_key_policy(paramiko.WarningPolicy())
+            client.connect(self.hostname,
+                           username=self.username,
+                           port=self.port,
+                           key_filename=self.keyfile)
+
+            stdin, stdout, stderr = client.exec_command("gerrit stream-events")
+
+            self._listen(stdout, stderr)
+
+            if not stdout.channel.exit_status_ready():
+                # The stream-event is still running but we are done polling
+                # on stdout most likely due to being asked to stop.
+                # Try to stop the stream-events command sending Ctrl-C
+                stdin.write("\x03")
+                time.sleep(.2)
+                if not stdout.channel.exit_status_ready():
+                    # we're still not ready to exit, lets force the channel
+                    # closed now.
+                    stdout.channel.close()
+            ret = stdout.channel.recv_exit_status()
+            self.log.debug("SSH exit status: %s" % ret)
+            client.close()
+
+            if ret and ret not in [-1, 130]:
+                raise Exception("Gerrit error executing stream-events")
+        except:
+            self.log.exception("Exception on ssh event stream:")
+            time.sleep(5)
+
+    def run(self):
+        while not self._stopped:
+            self._run()
+
+    def stop(self):
+        self.log.debug("Stopping watcher")
+        self._stopped = True
+
+
+class GerritConnection(BaseConnection):
+    driver_name = 'gerrit'
+    log = logging.getLogger("connection.gerrit")
+
+    def __init__(self, connection_name, connection_config):
+        super(GerritConnection, self).__init__(connection_name,
+                                               connection_config)
+        if 'server' not in self.connection_config:
+            raise Exception('server is required for gerrit connections in '
+                            '%s' % self.connection_name)
+        if 'user' not in self.connection_config:
+            raise Exception('user is required for gerrit connections in '
+                            '%s' % self.connection_name)
+
+        self.user = self.connection_config.get('user')
+        self.server = self.connection_config.get('server')
+        self.port = int(self.connection_config.get('port', 29418))
+        self.keyfile = self.connection_config.get('sshkey', None)
+        self.watcher_thread = None
+        self.event_queue = None
+        self.client = None
+
+        self.baseurl = self.connection_config.get('baseurl',
+                                                  'https://%s' % self.server)
+
+        self._change_cache = {}
+        self.gerrit_event_connector = None
+
+    def getCachedChange(self, key):
+        if key in self._change_cache:
+            return self._change_cache.get(key)
+        return None
+
+    def updateChangeCache(self, key, value):
+        self._change_cache[key] = value
+
+    def deleteCachedChange(self, key):
+        if key in self._change_cache:
+            del self._change_cache[key]
+
+    def maintainCache(self, relevant):
+        # This lets the user supply a list of change objects that are
+        # still in use.  Anything in our cache that isn't in the supplied
+        # list should be safe to remove from the cache.
+        remove = []
+        for key, change in self._change_cache.items():
+            if change not in relevant:
+                remove.append(key)
+        for key in remove:
+            del self._change_cache[key]
+
+    def addEvent(self, data):
+        return self.event_queue.put((time.time(), data))
+
+    def getEvent(self):
+        return self.event_queue.get()
+
+    def eventDone(self):
+        self.event_queue.task_done()
+
+    def review(self, project, change, message, action={}):
+        cmd = 'gerrit review --project %s' % project
+        if message:
+            cmd += ' --message "%s"' % message
+        for key, val in action.items():
+            if val is True:
+                cmd += ' --%s' % key
+            else:
+                cmd += ' --%s %s' % (key, val)
+        cmd += ' %s' % change
+        out, err = self._ssh(cmd)
+        return err
+
+    def query(self, query):
+        args = '--all-approvals --comments --commit-message'
+        args += ' --current-patch-set --dependencies --files'
+        args += ' --patch-sets --submit-records'
+        cmd = 'gerrit query --format json %s %s' % (
+            args, query)
+        out, err = self._ssh(cmd)
+        if not out:
+            return False
+        lines = out.split('\n')
+        if not lines:
+            return False
+        data = json.loads(lines[0])
+        if not data:
+            return False
+        self.log.debug("Received data from Gerrit query: \n%s" %
+                       (pprint.pformat(data)))
+        return data
+
+    def simpleQuery(self, query):
+        def _query_chunk(query):
+            args = '--commit-message --current-patch-set'
+
+            cmd = 'gerrit query --format json %s %s' % (
+                args, query)
+            out, err = self._ssh(cmd)
+            if not out:
+                return False
+            lines = out.split('\n')
+            if not lines:
+                return False
+
+            # filter out blank lines
+            data = [json.loads(line) for line in lines
+                    if line.startswith('{')]
+
+            # check last entry for more changes
+            more_changes = None
+            if 'moreChanges' in data[-1]:
+                more_changes = data[-1]['moreChanges']
+
+            # we have to remove the statistics line
+            del data[-1]
+
+            if not data:
+                return False, more_changes
+            self.log.debug("Received data from Gerrit query: \n%s" %
+                           (pprint.pformat(data)))
+            return data, more_changes
+
+        # gerrit returns 500 results by default, so implement paging
+        # for large projects like nova
+        alldata = []
+        chunk, more_changes = _query_chunk(query)
+        while(chunk):
+            alldata.extend(chunk)
+            if more_changes is None:
+                # continue sortKey based (before Gerrit 2.9)
+                resume = "resume_sortkey:'%s'" % chunk[-1]["sortKey"]
+            elif more_changes:
+                # continue moreChanges based (since Gerrit 2.9)
+                resume = "-S %d" % len(alldata)
+            else:
+                # no more changes
+                break
+
+            chunk, more_changes = _query_chunk("%s %s" % (query, resume))
+        return alldata
+
+    def _open(self):
+        client = paramiko.SSHClient()
+        client.load_system_host_keys()
+        client.set_missing_host_key_policy(paramiko.WarningPolicy())
+        client.connect(self.server,
+                       username=self.user,
+                       port=self.port,
+                       key_filename=self.keyfile)
+        self.client = client
+
+    def _ssh(self, command, stdin_data=None):
+        if not self.client:
+            self._open()
+
+        try:
+            self.log.debug("SSH command:\n%s" % command)
+            stdin, stdout, stderr = self.client.exec_command(command)
+        except:
+            self._open()
+            stdin, stdout, stderr = self.client.exec_command(command)
+
+        if stdin_data:
+            stdin.write(stdin_data)
+
+        out = stdout.read()
+        self.log.debug("SSH received stdout:\n%s" % out)
+
+        ret = stdout.channel.recv_exit_status()
+        self.log.debug("SSH exit status: %s" % ret)
+
+        err = stderr.read()
+        self.log.debug("SSH received stderr:\n%s" % err)
+        if ret:
+            raise Exception("Gerrit error executing %s" % command)
+        return (out, err)
+
+    def getInfoRefs(self, project):
+        url = "%s/p/%s/info/refs?service=git-upload-pack" % (
+            self.baseurl, project)
+        try:
+            data = urllib2.urlopen(url).read()
+        except:
+            self.log.error("Cannot get references from %s" % url)
+            raise  # keeps urllib2 error informations
+        ret = {}
+        read_headers = False
+        read_advertisement = False
+        if data[4] != '#':
+            raise Exception("Gerrit repository does not support "
+                            "git-upload-pack")
+        i = 0
+        while i < len(data):
+            if len(data) - i < 4:
+                raise Exception("Invalid length in info/refs")
+            plen = int(data[i:i + 4], 16)
+            i += 4
+            # It's the length of the packet, including the 4 bytes of the
+            # length itself, unless it's null, in which case the length is
+            # not included.
+            if plen > 0:
+                plen -= 4
+            if len(data) - i < plen:
+                raise Exception("Invalid data in info/refs")
+            line = data[i:i + plen]
+            i += plen
+            if not read_headers:
+                if plen == 0:
+                    read_headers = True
+                continue
+            if not read_advertisement:
+                read_advertisement = True
+                continue
+            if plen == 0:
+                # The terminating null
+                continue
+            line = line.strip()
+            revision, ref = line.split()
+            ret[ref] = revision
+        return ret
+
+    def getGitUrl(self, project):
+        url = 'ssh://%s@%s:%s/%s' % (self.user, self.server, self.port,
+                                     project.name)
+        return url
+
+    def getGitwebUrl(self, project, sha=None):
+        url = '%s/gitweb?p=%s.git' % (self.baseurl, project)
+        if sha:
+            url += ';a=commitdiff;h=' + sha
+        return url
+
+    def onLoad(self):
+        self.log.debug("Starting Gerrit Conncetion/Watchers")
+        self._start_watcher_thread()
+        self._start_event_connector()
+
+    def onStop(self):
+        self.log.debug("Stopping Gerrit Conncetion/Watchers")
+        self._stop_watcher_thread()
+        self._stop_event_connector()
+
+    def _stop_watcher_thread(self):
+        if self.watcher_thread:
+            self.watcher_thread.stop()
+            self.watcher_thread.join()
+
+    def _start_watcher_thread(self):
+        self.event_queue = Queue.Queue()
+        self.watcher_thread = GerritWatcher(
+            self,
+            self.user,
+            self.server,
+            self.port,
+            keyfile=self.keyfile)
+        self.watcher_thread.start()
+
+    def _stop_event_connector(self):
+        if self.gerrit_event_connector:
+            self.gerrit_event_connector.stop()
+            self.gerrit_event_connector.join()
+
+    def _start_event_connector(self):
+        self.gerrit_event_connector = GerritEventConnector(self)
+        self.gerrit_event_connector.start()
+
+
+def getSchema():
+    gerrit_connection = v.Any(str, v.Schema({}, extra=True))
+    return gerrit_connection

zuul/connection/smtp.py

@@ -0,0 +1,63 @@
+# Copyright 2014 Rackspace Australia
+#
+# 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 logging
+import voluptuous as v
+import smtplib
+
+from email.mime.text import MIMEText
+
+from zuul.connection import BaseConnection
+
+
+class SMTPConnection(BaseConnection):
+    driver_name = 'smtp'
+    log = logging.getLogger("connection.smtp")
+
+    def __init__(self, connection_name, connection_config):
+
+        super(SMTPConnection, self).__init__(connection_name,
+                                             connection_config)
+
+        self.smtp_server = self.connection_config.get(
+            'server', 'localhost')
+        self.smtp_port = self.connection_config.get('port', 25)
+        self.smtp_default_from = self.connection_config.get(
+            'default_from', 'zuul')
+        self.smtp_default_to = self.connection_config.get(
+            'default_to', 'zuul')
+
+    def sendMail(self, subject, message, from_email=None, to_email=None):
+        # Create a text/plain email message
+        from_email = from_email \
+            if from_email is not None else self.smtp_default_from
+        to_email = to_email if to_email is not None else self.smtp_default_to
+
+        msg = MIMEText(message)
+        msg['Subject'] = subject
+        msg['From'] = from_email
+        msg['To'] = to_email
+
+        try:
+            s = smtplib.SMTP(self.smtp_server, self.smtp_port)
+            s.sendmail(from_email, to_email.split(','), msg.as_string())
+            s.quit()
+        except:
+            return "Could not send email via SMTP"
+        return
+
+
+def getSchema():
+    smtp_connection = v.Any(str, v.Schema({}, extra=True))
+    return smtp_connection

zuul/exceptions.py

@@ -0,0 +1,26 @@
+# Copyright 2015 Rackspace Australia
+#
+# 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.
+
+
+class ChangeNotFound(Exception):
+    def __init__(self, number, ps):
+        self.number = number
+        self.ps = ps
+        self.change = "%s,%s" % (str(number), str(ps))
+        message = "Change %s not found" % self.change
+        super(ChangeNotFound, self).__init__(message)
+
+
+class MergeFailure(Exception):
+    pass

zuul/layoutvalidator.py

@@ -18,8 +18,6 @@
 import voluptuous as v
 import string
 
-from zuul.trigger import gerrit
-
 
 # Several forms accept either a single item or a list, this makes
 # specifying that in the schema easy (and explicit).
@@ -36,8 +34,6 @@ class LayoutSchema(object):
 
     precedence = v.Any('normal', 'low', 'high')
 
-    variable_dict = v.Schema({}, extra=True)
-
     approval = v.Schema({'username': str,
                          'email-filter': str,
                          'email': str,
@@ -45,49 +41,6 @@ class LayoutSchema(object):
                          'newer-than': str,
                          }, extra=True)
 
-    gerrit_trigger = {v.Required('event'):
-                      toList(v.Any('patchset-created',
-                                   'draft-published',
-                                   'change-abandoned',
-                                   'change-restored',
-                                   'change-merged',
-                                   'comment-added',
-                                   'ref-updated')),
-                      'comment_filter': toList(str),
-                      'comment': toList(str),
-                      'email_filter': toList(str),
-                      'email': toList(str),
-                      'username_filter': toList(str),
-                      'username': toList(str),
-                      'branch': toList(str),
-                      'ref': toList(str),
-                      'ignore-deletes': bool,
-                      'approval': toList(variable_dict),
-                      'require-approval': toList(approval),
-                      'reject-approval': toList(approval),
-                      }
-
-    timer_trigger = {v.Required('time'): str}
-
-    zuul_trigger = {v.Required('event'):
-                    toList(v.Any('parent-change-enqueued',
-                                 'project-change-merged')),
-                    'pipeline': toList(str),
-                    'require-approval': toList(approval),
-                    'reject-approval': toList(approval),
-                    }
-
-    trigger = v.Required({'gerrit': toList(gerrit_trigger),
-                          'timer': toList(timer_trigger),
-                          'zuul': toList(zuul_trigger)})
-
-    report_actions = {'gerrit': variable_dict,
-                      'smtp': {'to': str,
-                               'from': str,
-                               'subject': str,
-                               },
-                      }
-
     require = {'approval': toList(approval),
                'open': bool,
                'current-patchset': bool,
@@ -102,7 +55,7 @@ class LayoutSchema(object):
 
     pipeline = {v.Required('name'): str,
                 v.Required('manager'): manager,
-                'source': v.Any('gerrit'),
+                'source': str,
                 'precedence': precedence,
                 'description': str,
                 'require': require,
@@ -113,12 +66,6 @@ class LayoutSchema(object):
                 'footer-message': str,
                 'dequeue-on-new-patchset': bool,
                 'ignore-dependencies': bool,
-                'trigger': trigger,
-                'success': report_actions,
-                'failure': report_actions,
-                'merge-failure': report_actions,
-                'start': report_actions,
-                'disabled': report_actions,
                 'disable-after-consecutive-failures':
                     v.All(int, v.Range(min=1)),
                 'window': window,
@@ -128,7 +75,6 @@ class LayoutSchema(object):
                 'window-decrease-type': window_type,
                 'window-decrease-factor': window_factor,
                 }
-    pipelines = [pipeline]
 
     project_template = {v.Required('name'): str}
     project_templates = [project_template]
@@ -209,7 +155,42 @@ class LayoutSchema(object):
 
         return parameters
 
-    def getSchema(self, data):
+    def getDriverSchema(self, dtype, connections):
+        # TODO(jhesketh): Make the driver discovery dynamic
+        connection_drivers = {
+            'trigger': {
+                'gerrit': 'zuul.trigger.gerrit',
+            },
+            'reporter': {
+                'gerrit': 'zuul.reporter.gerrit',
+                'smtp': 'zuul.reporter.smtp',
+            },
+        }
+        standard_drivers = {
+            'trigger': {
+                'timer': 'zuul.trigger.timer',
+                'zuul': 'zuul.trigger.zuultrigger',
+            }
+        }
+
+        schema = {}
+        # Add the configured connections as available layout options
+        for connection_name, connection in connections.items():
+            for dname, dmod in connection_drivers.get(dtype, {}).items():
+                if connection.driver_name == dname:
+                    schema[connection_name] = toList(__import__(
+                        connection_drivers[dtype][dname],
+                        fromlist=['']).getSchema())
+
+        # Standard drivers are always available and don't require a unique
+        # (connection) name
+        for dname, dmod in standard_drivers.get(dtype, {}).items():
+            schema[dname] = toList(__import__(
+                standard_drivers[dtype][dname], fromlist=['']).getSchema())
+
+        return schema
+
+    def getSchema(self, data, connections=None):
         if not isinstance(data, dict):
             raise Exception("Malformed layout configuration: top-level type "
                             "should be a dictionary")
@@ -263,9 +244,31 @@ class LayoutSchema(object):
         for p in pipelines:
             project_template[p] = self.validateJob
         project_templates = [project_template]
+
+        # TODO(jhesketh): source schema is still defined above as sources
+        # currently aren't key/value so there is nothing to validate. Need to
+        # revisit this and figure out how to allow drivers with and without
+        # params. eg support all:
+        #   source: gerrit
+        # and
+        #   source:
+        #     gerrit:
+        #       - val
+        #       - val2
+        # and
+        #   source:
+        #     gerrit: something
+        # etc...
+        self.pipeline['trigger'] = v.Required(
+            self.getDriverSchema('trigger', connections))
+        for action in ['start', 'success', 'failure', 'merge-failure',
+                       'disabled']:
+            self.pipeline[action] = self.getDriverSchema('reporter',
+                                                         connections)
+
         # Gather our sub schemas
         schema = v.Schema({'includes': self.includes,
-                           v.Required('pipelines'): self.pipelines,
+                           v.Required('pipelines'): [self.pipeline],
                            'jobs': self.jobs,
                            'project-templates': project_templates,
                            v.Required('projects'): projects,
@@ -282,8 +285,45 @@ class LayoutValidator(object):
                                 path + [i])
             items.append(item['name'])
 
-    def validate(self, data):
-        schema = LayoutSchema().getSchema(data)
+    def extraDriverValidation(self, dtype, driver_data, connections=None):
+        # Some drivers may have extra validation to run on the layout
+        # TODO(jhesketh): Make the driver discovery dynamic
+        connection_drivers = {
+            'trigger': {
+                'gerrit': 'zuul.trigger.gerrit',
+            },
+            'reporter': {
+                'gerrit': 'zuul.reporter.gerrit',
+                'smtp': 'zuul.reporter.smtp',
+            },
+        }
+        standard_drivers = {
+            'trigger': {
+                'timer': 'zuul.trigger.timer',
+                'zuul': 'zuul.trigger.zuultrigger',
+            }
+        }
+
+        for dname, d_conf in driver_data.items():
+            for connection_name, connection in connections.items():
+                if connection_name == dname:
+                    if (connection.driver_name in
+                        connection_drivers.get(dtype, {}).keys()):
+                        module = __import__(
+                            connection_drivers[dtype][connection.driver_name],
+                            fromlist=['']
+                        )
+                        if 'validate_conf' in dir(module):
+                            module.validate_conf(d_conf)
+                    break
+            if dname in standard_drivers.get(dtype, {}).keys():
+                module = __import__(standard_drivers[dtype][dname],
+                                    fromlist=[''])
+                if 'validate_conf' in dir(module):
+                    module.validate_conf(d_conf)
+
+    def validate(self, data, connections=None):
+        schema = LayoutSchema().getSchema(data, connections)
         schema(data)
         self.checkDuplicateNames(data['pipelines'], ['pipelines'])
         if 'jobs' in data:
@@ -292,6 +332,11 @@ class LayoutValidator(object):
         if 'project-templates' in data:
             self.checkDuplicateNames(
                 data['project-templates'], ['project-templates'])
+
         for pipeline in data['pipelines']:
-            if 'gerrit' in pipeline['trigger']:
-                gerrit.validate_trigger(pipeline['trigger'])
+            self.extraDriverValidation('trigger', pipeline['trigger'],
+                                       connections)
+            for action in ['start', 'success', 'failure', 'merge-failure']:
+                if action in pipeline:
+                    self.extraDriverValidation('reporter', pipeline[action],
+                                               connections)

zuul/lib/connections.py

@@ -0,0 +1,66 @@
+# Copyright 2015 Rackspace Australia
+#
+# 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 re
+
+import zuul.connection.gerrit
+import zuul.connection.smtp
+
+
+def configure_connections(config):
+    # Register connections from the config
+
+    # TODO(jhesketh): import connection modules dynamically
+    connections = {}
+
+    for section_name in config.sections():
+        con_match = re.match(r'^connection ([\'\"]?)(.*)(\1)$',
+                             section_name, re.I)
+        if not con_match:
+            continue
+        con_name = con_match.group(2)
+        con_config = dict(config.items(section_name))
+
+        if 'driver' not in con_config:
+            raise Exception("No driver specified for connection %s."
+                            % con_name)
+
+        con_driver = con_config['driver']
+
+        # TODO(jhesketh): load the required class automatically
+        if con_driver == 'gerrit':
+            connections[con_name] = \
+                zuul.connection.gerrit.GerritConnection(con_name,
+                                                        con_config)
+        elif con_driver == 'smtp':
+            connections[con_name] = \
+                zuul.connection.smtp.SMTPConnection(con_name, con_config)
+        else:
+            raise Exception("Unknown driver, %s, for connection %s"
+                            % (con_config['driver'], con_name))
+
+    # If the [gerrit] or [smtp] sections still exist, load them in as a
+    # connection named 'gerrit' or 'smtp' respectfully
+
+    if 'gerrit' in config.sections():
+        connections['gerrit'] = \
+            zuul.connection.gerrit.GerritConnection(
+                '_legacy_gerrit', dict(config.items('gerrit')))
+
+    if 'smtp' in config.sections():
+        connections['smtp'] = \
+            zuul.connection.smtp.SMTPConnection(
+                '_legacy_smtp', dict(config.items('smtp')))
+
+    return connections

zuul/lib/gerrit.py

@@ -1,231 +0,0 @@
-# Copyright 2011 OpenStack, LLC.
-# Copyright 2012 Hewlett-Packard Development Company, L.P.
-#
-# 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 threading
-import select
-import json
-import time
-from six.moves import queue as Queue
-import paramiko
-import logging
-import pprint
-
-
-class GerritWatcher(threading.Thread):
-    log = logging.getLogger("gerrit.GerritWatcher")
-
-    def __init__(self, gerrit, username, hostname, port=29418, keyfile=None):
-        threading.Thread.__init__(self)
-        self.username = username
-        self.keyfile = keyfile
-        self.hostname = hostname
-        self.port = port
-        self.gerrit = gerrit
-
-    def _read(self, fd):
-        l = fd.readline()
-        data = json.loads(l)
-        self.log.debug("Received data from Gerrit event stream: \n%s" %
-                       pprint.pformat(data))
-        self.gerrit.addEvent((time.time(), data))
-
-    def _listen(self, stdout, stderr):
-        poll = select.poll()
-        poll.register(stdout.channel)
-        while True:
-            ret = poll.poll()
-            for (fd, event) in ret:
-                if fd == stdout.channel.fileno():
-                    if event == select.POLLIN:
-                        self._read(stdout)
-                    else:
-                        raise Exception("event on ssh connection")
-
-    def _run(self):
-        try:
-            client = paramiko.SSHClient()
-            client.load_system_host_keys()
-            client.set_missing_host_key_policy(paramiko.WarningPolicy())
-            client.connect(self.hostname,
-                           username=self.username,
-                           port=self.port,
-                           key_filename=self.keyfile)
-
-            stdin, stdout, stderr = client.exec_command("gerrit stream-events")
-
-            self._listen(stdout, stderr)
-
-            ret = stdout.channel.recv_exit_status()
-            self.log.debug("SSH exit status: %s" % ret)
-
-            if ret:
-                raise Exception("Gerrit error executing stream-events")
-        except:
-            self.log.exception("Exception on ssh event stream:")
-            time.sleep(5)
-
-    def run(self):
-        while True:
-            self._run()
-
-
-class Gerrit(object):
-    log = logging.getLogger("gerrit.Gerrit")
-
-    def __init__(self, hostname, username, port=29418, keyfile=None):
-        self.username = username
-        self.hostname = hostname
-        self.port = port
-        self.keyfile = keyfile
-        self.watcher_thread = None
-        self.event_queue = None
-        self.client = None
-
-    def startWatching(self):
-        self.event_queue = Queue.Queue()
-        self.watcher_thread = GerritWatcher(
-            self,
-            self.username,
-            self.hostname,
-            self.port,
-            keyfile=self.keyfile)
-        self.watcher_thread.start()
-
-    def addEvent(self, data):
-        return self.event_queue.put(data)
-
-    def getEvent(self):
-        return self.event_queue.get()
-
-    def eventDone(self):
-        self.event_queue.task_done()
-
-    def review(self, project, change, message, action={}):
-        cmd = 'gerrit review --project %s' % project
-        if message:
-            cmd += ' --message "%s"' % message
-        for k, v in action.items():
-            if v is True:
-                cmd += ' --%s' % k
-            else:
-                cmd += ' --label %s=%s' % (k, v)
-        cmd += ' %s' % change
-        out, err = self._ssh(cmd)
-        return err
-
-    def query(self, query):
-        args = '--all-approvals --comments --commit-message'
-        args += ' --current-patch-set --dependencies --files'
-        args += ' --patch-sets --submit-records'
-        cmd = 'gerrit query --format json %s %s' % (
-            args, query)
-        out, err = self._ssh(cmd)
-        if not out:
-            return False
-        lines = out.split('\n')
-        if not lines:
-            return False
-        data = json.loads(lines[0])
-        if not data:
-            return False
-        self.log.debug("Received data from Gerrit query: \n%s" %
-                       (pprint.pformat(data)))
-        return data
-
-    def simpleQuery(self, query):
-        def _query_chunk(query):
-            args = '--commit-message --current-patch-set'
-
-            cmd = 'gerrit query --format json %s %s' % (
-                args, query)
-            out, err = self._ssh(cmd)
-            if not out:
-                return False
-            lines = out.split('\n')
-            if not lines:
-                return False
-
-            # filter out blank lines
-            data = [json.loads(line) for line in lines
-                    if line.startswith('{')]
-
-            # check last entry for more changes
-            more_changes = None
-            if 'moreChanges' in data[-1]:
-                more_changes = data[-1]['moreChanges']
-
-            # we have to remove the statistics line
-            del data[-1]
-
-            if not data:
-                return False, more_changes
-            self.log.debug("Received data from Gerrit query: \n%s" %
-                           (pprint.pformat(data)))
-            return data, more_changes
-
-        # gerrit returns 500 results by default, so implement paging
-        # for large projects like nova
-        alldata = []
-        chunk, more_changes = _query_chunk(query)
-        while(chunk):
-            alldata.extend(chunk)
-            if more_changes is None:
-                # continue sortKey based (before Gerrit 2.9)
-                resume = "resume_sortkey:'%s'" % chunk[-1]["sortKey"]
-            elif more_changes:
-                # continue moreChanges based (since Gerrit 2.9)
-                resume = "-S %d" % len(alldata)
-            else:
-                # no more changes
-                break
-
-            chunk, more_changes = _query_chunk("%s %s" % (query, resume))
-        return alldata
-
-    def _open(self):
-        client = paramiko.SSHClient()
-        client.load_system_host_keys()
-        client.set_missing_host_key_policy(paramiko.WarningPolicy())
-        client.connect(self.hostname,
-                       username=self.username,
-                       port=self.port,
-                       key_filename=self.keyfile)
-        self.client = client
-
-    def _ssh(self, command, stdin_data=None):
-        if not self.client:
-            self._open()
-
-        try:
-            self.log.debug("SSH command:\n%s" % command)
-            stdin, stdout, stderr = self.client.exec_command(command)
-        except:
-            self._open()
-            stdin, stdout, stderr = self.client.exec_command(command)
-
-        if stdin_data:
-            stdin.write(stdin_data)
-
-        out = stdout.read()
-        self.log.debug("SSH received stdout:\n%s" % out)
-
-        ret = stdout.channel.recv_exit_status()
-        self.log.debug("SSH exit status: %s" % ret)
-
-        err = stderr.read()
-        self.log.debug("SSH received stderr:\n%s" % err)
-        if ret:
-            raise Exception("Gerrit error executing %s" % command)
-        return (out, err)

zuul/merger/merger.py

@@ -188,24 +188,29 @@ class Repo(object):
 class Merger(object):
     log = logging.getLogger("zuul.Merger")
 
-    def __init__(self, working_root, sshkey, email, username):
+    def __init__(self, working_root, connections, email, username):
         self.repos = {}
         self.working_root = working_root
         if not os.path.exists(working_root):
             os.makedirs(working_root)
-        if sshkey:
-            self._makeSSHWrapper(sshkey)
+        self._makeSSHWrappers(working_root, connections)
         self.email = email
         self.username = username
 
-    def _makeSSHWrapper(self, key):
-        name = os.path.join(self.working_root, '.ssh_wrapper')
+    def _makeSSHWrappers(self, working_root, connections):
+        for connection_name, connection in connections.items():
+            sshkey = connection.connection_config.get('sshkey')
+            if sshkey:
+                self._makeSSHWrapper(sshkey, working_root, connection_name)
+
+    def _makeSSHWrapper(self, key, merge_root, connection_name='default'):
+        wrapper_name = '.ssh_wrapper_%s' % connection_name
+        name = os.path.join(merge_root, wrapper_name)
         fd = open(name, 'w')
         fd.write('#!/bin/bash\n')
         fd.write('ssh -i %s $@\n' % key)
         fd.close()
         os.chmod(name, 0755)
-        os.environ['GIT_SSH'] = name
 
     def addProject(self, project, url):
         repo = None
@@ -263,10 +268,19 @@ class Merger(object):
 
         return commit
 
+    def _setGitSsh(self, connection_name):
+        wrapper_name = '.ssh_wrapper_%s' % connection_name
+        name = os.path.join(self.working_root, wrapper_name)
+        if os.path.isfile(name):
+            os.environ['GIT_SSH'] = name
+        elif 'GIT_SSH' in os.environ:
+            del os.environ['GIT_SSH']
+
     def _mergeItem(self, item, recent):
         self.log.debug("Processing refspec %s for project %s / %s ref %s" %
                        (item['refspec'], item['project'], item['branch'],
                         item['ref']))
+        self._setGitSsh(item['connection_name'])
         repo = self.getRepo(item['project'], item['url'])
         key = (item['project'], item['branch'])
         # See if we have a commit for this change already in this repo

zuul/merger/server.py

@@ -25,7 +25,7 @@ import merger
 class MergeServer(object):
     log = logging.getLogger("zuul.MergeServer")
 
-    def __init__(self, config):
+    def __init__(self, config, connections={}):
         self.config = config
         self.zuul_url = config.get('merger', 'zuul_url')
 
@@ -44,13 +44,8 @@ class MergeServer(object):
         else:
             merge_name = None
 
-        if self.config.has_option('gerrit', 'sshkey'):
-            sshkey = self.config.get('gerrit', 'sshkey')
-        else:
-            sshkey = None
-
-        self.merger = merger.Merger(merge_root, sshkey,
-                                    merge_email, merge_name)
+        self.merger = merger.Merger(merge_root, connections, merge_email,
+                                    merge_name)
 
     def start(self):
         self._running = True

zuul/model.py

@@ -81,10 +81,11 @@ class Pipeline(object):
         self.queues = []
         self.precedence = PRECEDENCE_NORMAL
         self.source = None
-        self.start_actions = None
-        self.success_actions = None
-        self.failure_actions = None
-        self.disabled_actions = None
+        self.start_actions = []
+        self.success_actions = []
+        self.failure_actions = []
+        self.merge_failure_actions = []
+        self.disabled_actions = []
         self.disable_at = None
         self._consecutive_failures = 0
         self._disabled = False
@@ -295,22 +296,21 @@ class ActionReporter(object):
     """An ActionReporter has a reporter and its configured parameters"""
 
     def __repr__(self):
-        return '<ActionReporter %s, %s>' % (self.reporter, self.params)
+        return '<ActionReporter %s>' % (self.reporter)
 
-    def __init__(self, reporter, params):
+    def __init__(self, reporter):
         self.reporter = reporter
-        self.params = params
 
     def report(self, source, change, message):
         """Sends the built message off to the configured reporter.
         Takes the change and message and adds the configured parameters.
         """
-        return self.reporter.report(source, change, message, self.params)
+        return self.reporter.report(source, change, message)
 
     def getSubmitAllowNeeds(self):
         """Gets the submit allow needs from the reporter based off the
         parameters."""
-        return self.reporter.getSubmitAllowNeeds(self.params)
+        return self.reporter.getSubmitAllowNeeds()
 
 
 class ChangeQueue(object):
@@ -1026,6 +1026,9 @@ class TriggerEvent(object):
         # an admin command, etc):
         self.forced_pipeline = None
 
+        # Internal mechanism to track if the change needs a refresh from cache
+        self._needs_refresh = False
+
     def __repr__(self):
         ret = '<TriggerEvent %s %s' % (self.type, self.project_name)
 

zuul/reporter/__init__.py

@@ -24,19 +24,25 @@ class BaseReporter(object):
     Defines the exact public methods that must be supplied.
     """
 
-    @abc.abstractmethod
-    def __init__(self, *args, **kwargs):
-        # TODO(jhesketh): Fix *args to just a connection
-        pass
+    def __init__(self, reporter_config={}, sched=None, connection=None):
+        self.reporter_config = reporter_config
+        self.sched = sched
+        self.connection = connection
+
+    def stop(self):
+        """Stop the reporter."""
 
     @abc.abstractmethod
-    def report(self, source, change, message, params):
+    def report(self, source, change, message):
         """Send the compiled report message."""
 
-    def getSubmitAllowNeeds(self, params):
+    def getSubmitAllowNeeds(self):
         """Get a list of code review labels that are allowed to be
         "needed" in the submit records for a change, with respect
         to this queue.  In other words, the list of review labels
         this reporter itself is likely to set before submitting.
         """
         return []
+
+    def postConfig(self):
+        """Run tasks after configuration is reloaded"""

zuul/reporter/gerrit.py

@@ -13,6 +13,7 @@
 # under the License.
 
 import logging
+import voluptuous as v
 
 
 from zuul.reporter import BaseReporter
@@ -24,25 +25,26 @@ class GerritReporter(BaseReporter):
     name = 'gerrit'
     log = logging.getLogger("zuul.reporter.gerrit.Reporter")
 
-    def __init__(self, gerrit):
-        """Set up the reporter."""
-        # TODO: make default_gerrit come from a connection
-        self.default_gerrit = gerrit
-
-    def report(self, source, change, message, params):
+    def report(self, source, change, message):
         """Send a message to gerrit."""
         self.log.debug("Report change %s, params %s, message: %s" %
-                       (change, params, message))
+                       (change, self.reporter_config, message))
         changeid = '%s,%s' % (change.number, change.patchset)
         change._ref_sha = source.getRefSha(change.project.name,
                                            'refs/heads/' + change.branch)
-        return self.default_gerrit.review(
-            change.project.name, changeid, message, params)
 
-    def getSubmitAllowNeeds(self, params):
+        return self.connection.review(change.project.name, changeid, message,
+                                      self.reporter_config)
+
+    def getSubmitAllowNeeds(self):
         """Get a list of code review labels that are allowed to be
         "needed" in the submit records for a change, with respect
         to this queue.  In other words, the list of review labels
         this reporter itself is likely to set before submitting.
         """
-        return params
+        return self.reporter_config
+
+
+def getSchema():
+    gerrit_reporter = v.Any(str, v.Schema({}, extra=True))
+    return gerrit_reporter

zuul/reporter/smtp.py

@@ -13,9 +13,7 @@
 # under the License.
 
 import logging
-import smtplib
-
-from email.mime.text import MIMEText
+import voluptuous as v
 
 from zuul.reporter import BaseReporter
 
@@ -26,40 +24,29 @@ class SMTPReporter(BaseReporter):
     name = 'smtp'
     log = logging.getLogger("zuul.reporter.smtp.Reporter")
 
-    def __init__(self, smtp_default_from, smtp_default_to,
-                 smtp_server='localhost', smtp_port=25):
-        """Set up the reporter.
-
-        Takes parameters for the smtp server.
-        """
-        self.smtp_server = smtp_server
-        self.smtp_port = smtp_port
-        self.smtp_default_from = smtp_default_from
-        self.smtp_default_to = smtp_default_to
-
-    def report(self, source, change, message, params):
+    def report(self, source, change, message):
         """Send the compiled report message via smtp."""
         self.log.debug("Report change %s, params %s, message: %s" %
-                       (change, params, message))
+                       (change, self.reporter_config, message))
 
-        # Create a text/plain email message
-        from_email = params['from']\
-            if 'from' in params else self.smtp_default_from
-        to_email = params['to']\
-            if 'to' in params else self.smtp_default_to
-        msg = MIMEText(message)
-        if 'subject' in params:
-            subject = params['subject'].format(change=change)
+        from_email = self.reporter_config['from'] \
+            if 'from' in self.reporter_config else None
+        to_email = self.reporter_config['to'] \
+            if 'to' in self.reporter_config else None
+
+        if 'subject' in self.reporter_config:
+            subject = self.reporter_config['subject'].format(change=change)
         else:
             subject = "Report for change %s" % change
-        msg['Subject'] = subject
-        msg['From'] = from_email
-        msg['To'] = to_email
 
-        try:
-            s = smtplib.SMTP(self.smtp_server, self.smtp_port)
-            s.sendmail(from_email, to_email.split(','), msg.as_string())
-            s.quit()
-        except:
-            return "Could not send email via SMTP"
-        return
+        self.connection.sendMail(subject, message, from_email, to_email)
+
+
+def getSchema():
+    smtp_reporter = v.Schema({
+        'connection': str,
+        'to': str,
+        'from': str,
+        'subject': str,
+    })
+    return smtp_reporter

zuul/scheduler.py

@@ -31,7 +31,7 @@ import layoutvalidator
 import model
 from model import ActionReporter, Pipeline, Project, ChangeQueue
 from model import ChangeishFilter, NullChange
-from zuul import change_matcher
+from zuul import change_matcher, exceptions
 from zuul import version as zuul_version
 
 statsd = extras.try_import('statsd.statsd')
@@ -59,10 +59,6 @@ def deep_format(obj, paramdict):
     return ret
 
 
-class MergeFailure(Exception):
-    pass
-
-
 class ManagementEvent(object):
     """An event that should be processed within the main queue run loop"""
     def __init__(self):
@@ -178,7 +174,7 @@ def toList(item):
 class Scheduler(threading.Thread):
     log = logging.getLogger("zuul.Scheduler")
 
-    def __init__(self):
+    def __init__(self, config):
         threading.Thread.__init__(self)
         self.daemon = True
         self.wake_event = threading.Event()
@@ -189,10 +185,14 @@ class Scheduler(threading.Thread):
         self._stopped = False
         self.launcher = None
         self.merger = None
-        self.sources = dict()
+        self.connections = dict()
+        # Despite triggers being part of the pipeline, there is one trigger set
+        # per scheduler. The pipeline handles the trigger filters but since
+        # the events are handled by the scheduler itself it needs to handle
+        # the loading of the triggers.
+        # self.triggers['connection_name'] = triggerObject
         self.triggers = dict()
-        self.reporters = dict()
-        self.config = None
+        self.config = config
 
         self.trigger_event_queue = Queue.Queue()
         self.result_event_queue = Queue.Queue()
@@ -204,10 +204,14 @@ class Scheduler(threading.Thread):
 
     def stop(self):
         self._stopped = True
+        self._unloadDrivers()
+        self.stopConnections()
         self.wake_event.set()
 
-    def testConfig(self, config_path):
-        return self._parseConfig(config_path)
+    def testConfig(self, config_path, connections):
+        # Take the list of set up connections directly here rather than with
+        # registerConnections as we don't want to do the onLoad event yet.
+        return self._parseConfig(config_path, connections)
 
     def _parseSkipIf(self, config_job):
         cm = change_matcher
@@ -237,7 +241,76 @@ class Scheduler(threading.Thread):
             # Any skip-if predicate can be matched to trigger a skip
             return cm.MatchAny(skip_matchers)
 
-    def _parseConfig(self, config_path):
+    def registerConnections(self, connections):
+        self.connections = connections
+        for connection_name, connection in self.connections.items():
+            connection.registerScheduler(self)
+            connection.onLoad()
+
+    def stopConnections(self):
+        for connection_name, connection in self.connections.items():
+            connection.onStop()
+
+    def _unloadDrivers(self):
+        for trigger in self.triggers.values():
+            trigger.stop()
+        for pipeline in self.layout.pipelines.values():
+            pipeline.source.stop()
+            for action in ['start_actions', 'success_actions',
+                           'failure_actions', 'merge_failure_actions']:
+                for action_reporter in pipeline.__getattribute__(action):
+                    action_reporter.reporter.stop()
+
+    def _getDriver(self, dtype, connection_name, driver_config={}):
+        # Instantiate a driver such as a trigger, source or reporter
+        # TODO(jhesketh): Make this list dynamic or use entrypoints etc.
+        # Stevedore was not a good fit here due to the nature of triggers.
+        # Specifically we don't want to load a trigger per a pipeline as one
+        # trigger can listen to a stream (from gerrit, for example) and the
+        # scheduler decides which eventfilter to use. As such we want to load
+        # trigger+connection pairs uniquely.
+        drivers = {
+            'source': {
+                'gerrit': 'zuul.source.gerrit:GerritSource',
+            },
+            'trigger': {
+                'gerrit': 'zuul.trigger.gerrit:GerritTrigger',
+                'timer': 'zuul.trigger.timer:TimerTrigger',
+                'zuul': 'zuul.trigger.zuultrigger:ZuulTrigger',
+            },
+            'reporter': {
+                'gerrit': 'zuul.reporter.gerrit:GerritReporter',
+                'smtp': 'zuul.reporter.smtp:SMTPReporter',
+            },
+        }
+
+        # TODO(jhesketh): Check the connection_name exists
+        if connection_name in self.connections.keys():
+            driver_name = self.connections[connection_name].driver_name
+            connection = self.connections[connection_name]
+        else:
+            # In some cases a driver may not be related to a connection. For
+            # example, the 'timer' or 'zuul' triggers.
+            driver_name = connection_name
+            connection = None
+        driver = drivers[dtype][driver_name].split(':')
+        driver_instance = getattr(
+            __import__(driver[0], fromlist=['']), driver[1])(
+                driver_config, self, connection
+        )
+
+        return driver_instance
+
+    def _getSourceDriver(self, connection_name):
+        return self._getDriver('source', connection_name)
+
+    def _getReporterDriver(self, connection_name, driver_config={}):
+        return self._getDriver('reporter', connection_name, driver_config)
+
+    def _getTriggerDriver(self, connection_name, driver_config={}):
+        return self._getDriver('trigger', connection_name, driver_config)
+
+    def _parseConfig(self, config_path, connections):
         layout = model.Layout()
         project_templates = {}
 
@@ -250,7 +323,7 @@ class Scheduler(threading.Thread):
             data = yaml.load(config_file)
 
         validator = layoutvalidator.LayoutValidator()
-        validator.validate(data)
+        validator.validate(data, connections)
 
         config_env = {}
         for include in data.get('includes', []):
@@ -266,8 +339,8 @@ class Scheduler(threading.Thread):
             pipeline = Pipeline(conf_pipeline['name'])
             pipeline.description = conf_pipeline.get('description')
             # TODO(jeblair): remove backwards compatibility:
-            pipeline.source = self.sources[conf_pipeline.get('source',
-                                                             'gerrit')]
+            pipeline.source = self._getSourceDriver(
+                conf_pipeline.get('source', 'gerrit'))
             precedence = model.PRECEDENCE_MAP[conf_pipeline.get('precedence')]
             pipeline.precedence = precedence
             pipeline.failure_message = conf_pipeline.get('failure-message',
@@ -293,12 +366,10 @@ class Scheduler(threading.Thread):
                 if conf_pipeline.get(action):
                     for reporter_name, params \
                         in conf_pipeline.get(action).items():
-                        if reporter_name in self.reporters.keys():
-                            action_reporters[action].append(ActionReporter(
-                                self.reporters[reporter_name], params))
-                        else:
-                            self.log.error('Invalid reporter name %s' %
-                                           reporter_name)
+                        reporter = self._getReporterDriver(reporter_name,
+                                                           params)
+                        action_reporters[action].append(ActionReporter(
+                            reporter))
             pipeline.start_actions = action_reporters['start']
             pipeline.success_actions = action_reporters['success']
             pipeline.failure_actions = action_reporters['failure']
@@ -339,10 +410,16 @@ class Scheduler(threading.Thread):
                 )
                 manager.changeish_filters.append(f)
 
-            # TODO(jhesketh): Allow multiple triggers per pipeline
-            for trigger in self.triggers.values():
-                manager.event_filters += \
-                    trigger.getEventFilters(conf_pipeline['trigger'])
+            for trigger_name, trigger_config\
+                in conf_pipeline.get('trigger').items():
+                if trigger_name not in self.triggers.keys():
+                    self.triggers[trigger_name] = \
+                        self._getTriggerDriver(trigger_name, trigger_config)
+
+            for trigger_name, trigger in self.triggers.items():
+                if trigger_name in conf_pipeline['trigger']:
+                    manager.event_filters += trigger.getEventFilters(
+                        conf_pipeline['trigger'][trigger_name])
 
         for project_template in data.get('project-templates', []):
             # Make sure the template only contains valid pipelines
@@ -461,21 +538,6 @@ class Scheduler(threading.Thread):
     def setMerger(self, merger):
         self.merger = merger
 
-    def registerSource(self, source, name=None):
-        if name is None:
-            name = source.name
-        self.sources[name] = source
-
-    def registerTrigger(self, trigger, name=None):
-        if name is None:
-            name = trigger.name
-        self.triggers[name] = trigger
-
-    def registerReporter(self, reporter, name=None):
-        if name is None:
-            name = reporter.name
-        self.reporters[name] = reporter
-
     def getProject(self, name, create_foreign=False):
         self.layout_lock.acquire()
         p = None
@@ -651,8 +713,9 @@ class Scheduler(threading.Thread):
         self.config = event.config
         try:
             self.log.debug("Performing reconfiguration")
+            self._unloadDrivers()
             layout = self._parseConfig(
-                self.config.get('zuul', 'layout_config'))
+                self.config.get('zuul', 'layout_config'), self.connections)
             for name, new_pipeline in layout.pipelines.items():
                 old_pipeline = self.layout.pipelines.get(name)
                 if not old_pipeline:
@@ -709,8 +772,12 @@ class Scheduler(threading.Thread):
             self.maintainTriggerCache()
             for trigger in self.triggers.values():
                 trigger.postConfig()
-            for source in self.sources.values():
-                source.postConfig()
+            for pipeline in self.layout.pipelines.values():
+                pipeline.source.postConfig()
+                for action in ['start_actions', 'success_actions',
+                               'failure_actions', 'merge_failure_actions']:
+                    for action_reporter in pipeline.__getattribute__(action):
+                        action_reporter.reporter.postConfig()
             if statsd:
                 try:
                     for pipeline in self.layout.pipelines.values():
@@ -840,10 +907,9 @@ class Scheduler(threading.Thread):
             for item in pipeline.getAllItems():
                 relevant.add(item.change)
                 relevant.update(item.change.getRelatedChanges())
+            pipeline.source.maintainCache(relevant)
             self.log.debug("End maintain trigger cache for: %s" % pipeline)
         self.log.debug("Trigger cache size: %s" % len(relevant))
-        for source in self.sources.values():
-            source.maintainCache(relevant)
 
     def process_event_queue(self):
         self.log.debug("Fetching trigger event")
@@ -851,12 +917,22 @@ class Scheduler(threading.Thread):
         self.log.debug("Processing trigger event %s" % event)
         try:
             project = self.layout.projects.get(event.project_name)
-            if not project or project.foreign:
-                self.log.debug("Project %s not found" % event.project_name)
-                return
 
             for pipeline in self.layout.pipelines.values():
-                change = pipeline.source.getChange(event, project)
+                # Get the change even if the project is unknown to us for the
+                # use of updating the cache if there is another change
+                # depending on this foreign one.
+                try:
+                    change = pipeline.source.getChange(event, project)
+                except exceptions.ChangeNotFound as e:
+                    self.log.debug("Unable to get change %s from source %s. "
+                                   "(most likely looking for a change from "
+                                   "another connection trigger)",
+                                   e.change, pipeline.source)
+                    continue
+                if not project or project.foreign:
+                    self.log.debug("Project %s not found" % event.project_name)
+                    continue
                 if event.type == 'patchset-created':
                     pipeline.manager.removeOldVersionsOfChange(change)
                 elif event.type == 'change-abandoned':
@@ -1248,8 +1324,8 @@ class BasePipelineManager(object):
             self.reportStats(item)
             self.enqueueChangesBehind(change, quiet, ignore_requirements,
                                       change_queue)
-            self.sched.triggers['zuul'].onChangeEnqueued(item.change,
-                                                         self.pipeline)
+            for trigger in self.sched.triggers.values():
+                trigger.onChangeEnqueued(item.change, self.pipeline)
             return True
 
     def dequeueItem(self, item):
@@ -1278,9 +1354,11 @@ class BasePipelineManager(object):
         elif hasattr(item.change, 'newrev'):
             oldrev = item.change.oldrev
             newrev = item.change.newrev
+        connection_name = self.pipeline.source.connection.connection_name
         return dict(project=item.change.project.name,
                     url=self.pipeline.source.getGitUrl(
                         item.change.project),
+                    connection_name=connection_name,
                     merge_mode=item.change.project.merge_mode,
                     refspec=item.change.refspec,
                     branch=item.change.branch,
@@ -1379,7 +1457,7 @@ class BasePipelineManager(object):
             if item.live:
                 try:
                     self.reportItem(item)
-                except MergeFailure:
+                except exceptions.MergeFailure:
                     pass
             return (True, nnfi)
         dep_items = self.getFailingDependentItems(item)
@@ -1420,7 +1498,7 @@ class BasePipelineManager(object):
             and item.live):
             try:
                 self.reportItem(item)
-            except MergeFailure:
+            except exceptions.MergeFailure:
                 failing_reasons.append("it did not merge")
                 for item_behind in item.items_behind:
                     self.log.info("Resetting builds for change %s because the "
@@ -1518,12 +1596,15 @@ class BasePipelineManager(object):
                 change_queue.decreaseWindowSize()
                 self.log.debug("%s window size decreased to %s" %
                                (change_queue, change_queue.window))
-                raise MergeFailure("Change %s failed to merge" % item.change)
+                raise exceptions.MergeFailure(
+                    "Change %s failed to merge" % item.change)
             else:
                 change_queue.increaseWindowSize()
                 self.log.debug("%s window size increased to %s" %
                                (change_queue, change_queue.window))
-                self.sched.triggers['zuul'].onChangeMerged(item.change)
+
+                for trigger in self.sched.triggers.values():
+                    trigger.onChangeMerged(item.change, self.pipeline.source)
 
     def _reportItem(self, item):
         self.log.debug("Reporting change %s" % item.change)

zuul/source/__init__.py

@@ -27,9 +27,13 @@ class BaseSource(object):
 
     Defines the exact public methods that must be supplied."""
 
-    @abc.abstractmethod
-    def __init__(self, config, sched):
-        """Constructor."""
+    def __init__(self, source_config={}, sched=None, connection=None):
+        self.source_config = source_config
+        self.sched = sched
+        self.connection = connection
+
+    def stop(self):
+        """Stop the source."""
 
     @abc.abstractmethod
     def getRefSha(self, project, ref):

zuul/source/gerrit.py

@@ -15,8 +15,7 @@
 import logging
 import re
 import time
-import urllib2
-from zuul.lib import gerrit
+from zuul import exceptions
 from zuul.model import Change, Ref, NullChange
 from zuul.source import BaseSource
 
@@ -30,75 +29,10 @@ class GerritSource(BaseSource):
     depends_on_re = re.compile(r"^Depends-On: (I[0-9a-f]{40})\s*$",
                                re.MULTILINE | re.IGNORECASE)
 
-    def __init__(self, config, sched):
-        self._change_cache = {}
-        self.sched = sched
-        self.config = config
-        self.server = config.get('gerrit', 'server')
-        if config.has_option('gerrit', 'baseurl'):
-            self.baseurl = config.get('gerrit', 'baseurl')
-        else:
-            self.baseurl = 'https://%s' % self.server
-        user = config.get('gerrit', 'user')
-        if config.has_option('gerrit', 'sshkey'):
-            sshkey = config.get('gerrit', 'sshkey')
-        else:
-            sshkey = None
-        if config.has_option('gerrit', 'port'):
-            port = int(config.get('gerrit', 'port'))
-        else:
-            port = 29418
-        self.gerrit = gerrit.Gerrit(self.server, user, port, sshkey)
-        self.gerrit.startWatching()
-
-    def _getInfoRefs(self, project):
-        url = "%s/p/%s/info/refs?service=git-upload-pack" % (
-            self.baseurl, project)
-        try:
-            data = urllib2.urlopen(url).read()
-        except:
-            self.log.error("Cannot get references from %s" % url)
-            raise  # keeps urllib2 error informations
-        ret = {}
-        read_headers = False
-        read_advertisement = False
-        if data[4] != '#':
-            raise Exception("Gerrit repository does not support "
-                            "git-upload-pack")
-        i = 0
-        while i < len(data):
-            if len(data) - i < 4:
-                raise Exception("Invalid length in info/refs")
-            plen = int(data[i:i + 4], 16)
-            i += 4
-            # It's the length of the packet, including the 4 bytes of the
-            # length itself, unless it's null, in which case the length is
-            # not included.
-            if plen > 0:
-                plen -= 4
-            if len(data) - i < plen:
-                raise Exception("Invalid data in info/refs")
-            line = data[i:i + plen]
-            i += plen
-            if not read_headers:
-                if plen == 0:
-                    read_headers = True
-                continue
-            if not read_advertisement:
-                read_advertisement = True
-                continue
-            if plen == 0:
-                # The terminating null
-                continue
-            line = line.strip()
-            revision, ref = line.split()
-            ret[ref] = revision
-        return ret
-
     def getRefSha(self, project, ref):
         refs = {}
         try:
-            refs = self._getInfoRefs(project)
+            refs = self.connection.getInfoRefs(project)
         except:
             self.log.exception("Exception looking for ref %s" %
                                ref)
@@ -123,7 +57,7 @@ class GerritSource(BaseSource):
             # means it's merged.
             return True
 
-        data = self.gerrit.query(change.number)
+        data = self.connection.query(change.number)
         change._data = data
         change.is_merged = self._isMerged(change)
         if not head:
@@ -189,23 +123,17 @@ class GerritSource(BaseSource):
             return False
         return True
 
-    def maintainCache(self, relevant):
-        # This lets the user supply a list of change objects that are
-        # still in use.  Anything in our cache that isn't in the supplied
-        # list should be safe to remove from the cache.
-        remove = []
-        for key, change in self._change_cache.items():
-            if change not in relevant:
-                remove.append(key)
-        for key in remove:
-            del self._change_cache[key]
-
     def postConfig(self):
         pass
 
     def getChange(self, event, project):
         if event.change_number:
-            change = self._getChange(event.change_number, event.patch_number)
+            refresh = False
+            if event._needs_refresh:
+                refresh = True
+                event._needs_refresh = False
+            change = self._getChange(event.change_number, event.patch_number,
+                                     refresh=refresh)
         elif event.ref:
             change = Ref(project)
             change.ref = event.ref
@@ -218,21 +146,19 @@ class GerritSource(BaseSource):
 
     def _getChange(self, number, patchset, refresh=False, history=None):
         key = '%s,%s' % (number, patchset)
-        change = None
-        if key in self._change_cache:
-            change = self._change_cache.get(key)
-            if not refresh:
-                return change
+        change = self.connection.getCachedChange(key)
+        if change and not refresh:
+            return change
         if not change:
             change = Change(None)
             change.number = number
             change.patchset = patchset
         key = '%s,%s' % (change.number, change.patchset)
-        self._change_cache[key] = change
+        self.connection.updateChangeCache(key, change)
         try:
             self._updateChange(change, history)
         except Exception:
-            del self._change_cache[key]
+            self.connection.deleteCachedChange(key)
             raise
         return change
 
@@ -242,7 +168,7 @@ class GerritSource(BaseSource):
         query = "project:%s status:open" % (project.name,)
         self.log.debug("Running query %s to get project open changes" %
                        (query,))
-        data = self.gerrit.simpleQuery(query)
+        data = self.connection.simpleQuery(query)
         changes = []
         for record in data:
             try:
@@ -266,7 +192,7 @@ class GerritSource(BaseSource):
             query = "change:%s" % (match,)
             self.log.debug("Running query %s to find needed changes" %
                            (query,))
-            records.extend(self.gerrit.simpleQuery(query))
+            records.extend(self.connection.simpleQuery(query))
         return records
 
     def _getNeededByFromCommit(self, change_id):
@@ -275,7 +201,7 @@ class GerritSource(BaseSource):
         query = 'message:%s' % change_id
         self.log.debug("Running query %s to find changes needed-by" %
                        (query,))
-        results = self.gerrit.simpleQuery(query)
+        results = self.connection.simpleQuery(query)
         for result in results:
             for match in self.depends_on_re.findall(
                 result['commitMessage']):
@@ -293,15 +219,14 @@ class GerritSource(BaseSource):
     def _updateChange(self, change, history=None):
         self.log.info("Updating information for %s,%s" %
                       (change.number, change.patchset))
-        data = self.gerrit.query(change.number)
+        data = self.connection.query(change.number)
         change._data = data
 
         if change.patchset is None:
             change.patchset = data['currentPatchSet']['number']
 
         if 'project' not in data:
-            raise Exception("Change %s,%s not found" % (change.number,
-                                                        change.patchset))
+            raise exceptions.ChangeNotFound(change.number, change.patchset)
         # If updated changed came as a dependent on
         # and its project is not defined,
         # then create a 'foreign' project for it in layout
@@ -393,17 +318,10 @@ class GerritSource(BaseSource):
         return change
 
     def getGitUrl(self, project):
-        server = self.config.get('gerrit', 'server')
-        user = self.config.get('gerrit', 'user')
-        if self.config.has_option('gerrit', 'port'):
-            port = int(self.config.get('gerrit', 'port'))
-        else:
-            port = 29418
-        url = 'ssh://%s@%s:%s/%s' % (user, server, port, project.name)
-        return url
+        return self.connection.getGitUrl(project)
 
     def _getGitwebUrl(self, project, sha=None):
-        url = '%s/gitweb?p=%s.git' % (self.baseurl, project)
-        if sha:
-            url += ';a=commitdiff;h=' + sha
-        return url
+        return self.connection.getGitwebUrl(project, sha)
+
+    def maintainCache(self, relevant):
+        self.connection.maintainCache(relevant)

zuul/trigger/__init__.py

@@ -23,9 +23,10 @@ class BaseTrigger(object):
 
     Defines the exact public methods that must be supplied."""
 
-    @abc.abstractmethod
-    def __init__(self, *args, **kwargs):
-        """Constructor."""
+    def __init__(self, trigger_config={}, sched=None, connection=None):
+        self.trigger_config = trigger_config
+        self.sched = sched
+        self.connection = connection
 
     def stop(self):
         """Stop the trigger."""
@@ -38,7 +39,7 @@ class BaseTrigger(object):
     def postConfig(self):
         """Called after config is loaded."""
 
-    def onChangeMerged(self, change):
+    def onChangeMerged(self, change, source):
         """Called when a change has been merged."""
 
     def onChangeEnqueued(self, change, pipeline):

zuul/trigger/gerrit.py

@@ -13,129 +13,15 @@
 # under the License.
 
 import logging
-import threading
-import time
-import voluptuous
-from zuul.model import EventFilter, TriggerEvent
+import voluptuous as v
+from zuul.model import EventFilter
 from zuul.trigger import BaseTrigger
 
 
-class GerritEventConnector(threading.Thread):
-    """Move events from Gerrit to the scheduler."""
-
-    log = logging.getLogger("zuul.GerritEventConnector")
-    delay = 5.0
-
-    def __init__(self, gerrit, sched, trigger, source):
-        super(GerritEventConnector, self).__init__()
-        self.daemon = True
-        self.gerrit = gerrit
-        self.sched = sched
-        self.trigger = trigger
-        self.source = source
-        self._stopped = False
-
-    def stop(self):
-        self._stopped = True
-        self.gerrit.addEvent((None, None))
-
-    def _handleEvent(self):
-        ts, data = self.gerrit.getEvent()
-        if self._stopped:
-            return
-        # Gerrit can produce inconsistent data immediately after an
-        # event, So ensure that we do not deliver the event to Zuul
-        # until at least a certain amount of time has passed.  Note
-        # that if we receive several events in succession, we will
-        # only need to delay for the first event.  In essence, Zuul
-        # should always be a constant number of seconds behind Gerrit.
-        now = time.time()
-        time.sleep(max((ts + self.delay) - now, 0.0))
-        event = TriggerEvent()
-        event.type = data.get('type')
-        event.trigger_name = self.trigger.name
-        change = data.get('change')
-        if change:
-            event.project_name = change.get('project')
-            event.branch = change.get('branch')
-            event.change_number = change.get('number')
-            event.change_url = change.get('url')
-            patchset = data.get('patchSet')
-            if patchset:
-                event.patch_number = patchset.get('number')
-                event.refspec = patchset.get('ref')
-            event.approvals = data.get('approvals', [])
-            event.comment = data.get('comment')
-        refupdate = data.get('refUpdate')
-        if refupdate:
-            event.project_name = refupdate.get('project')
-            event.ref = refupdate.get('refName')
-            event.oldrev = refupdate.get('oldRev')
-            event.newrev = refupdate.get('newRev')
-        # Map the event types to a field name holding a Gerrit
-        # account attribute. See Gerrit stream-event documentation
-        # in cmd-stream-events.html
-        accountfield_from_type = {
-            'patchset-created': 'uploader',
-            'draft-published': 'uploader',  # Gerrit 2.5/2.6
-            'change-abandoned': 'abandoner',
-            'change-restored': 'restorer',
-            'change-merged': 'submitter',
-            'merge-failed': 'submitter',  # Gerrit 2.5/2.6
-            'comment-added': 'author',
-            'ref-updated': 'submitter',
-            'reviewer-added': 'reviewer',  # Gerrit 2.5/2.6
-        }
-        try:
-            event.account = data.get(accountfield_from_type[event.type])
-        except KeyError:
-            self.log.error("Received unrecognized event type '%s' from Gerrit.\
-                    Can not get account information." % event.type)
-            event.account = None
-
-        if event.change_number and self.sched.getProject(event.project_name):
-            # Call _getChange for the side effect of updating the
-            # cache.  Note that this modifies Change objects outside
-            # the main thread.
-            self.source._getChange(event.change_number,
-                                   event.patch_number,
-                                   refresh=True)
-
-        self.sched.addEvent(event)
-
-    def run(self):
-        while True:
-            if self._stopped:
-                return
-            try:
-                self._handleEvent()
-            except:
-                self.log.exception("Exception moving Gerrit event:")
-            finally:
-                self.gerrit.eventDone()
-
-
 class GerritTrigger(BaseTrigger):
     name = 'gerrit'
     log = logging.getLogger("zuul.trigger.Gerrit")
 
-    def __init__(self, gerrit, config, sched, source):
-        self.sched = sched
-        # TODO(jhesketh): Make 'gerrit' come from a connection (rather than the
-        #                 source)
-        # TODO(jhesketh): Remove the requirement for a gerrit source (currently
-        #                 it is needed so on a trigger event the cache is
-        #                 updated. However if we share a connection object the
-        #                 cache could be stored there)
-        self.config = config
-        self.gerrit_connector = GerritEventConnector(gerrit, sched, self,
-                                                     source)
-        self.gerrit_connector.start()
-
-    def stop(self):
-        self.gerrit_connector.stop()
-        self.gerrit_connector.join()
-
     def getEventFilters(self, trigger_conf):
         def toList(item):
             if not item:
@@ -145,50 +31,87 @@ class GerritTrigger(BaseTrigger):
             return [item]
 
         efilters = []
-        if 'gerrit' in trigger_conf:
-            for trigger in toList(trigger_conf['gerrit']):
-                approvals = {}
-                for approval_dict in toList(trigger.get('approval')):
-                    for k, v in approval_dict.items():
-                        approvals[k] = v
-                # Backwards compat for *_filter versions of these args
-                comments = toList(trigger.get('comment'))
-                if not comments:
-                    comments = toList(trigger.get('comment_filter'))
-                emails = toList(trigger.get('email'))
-                if not emails:
-                    emails = toList(trigger.get('email_filter'))
-                usernames = toList(trigger.get('username'))
-                if not usernames:
-                    usernames = toList(trigger.get('username_filter'))
-                ignore_deletes = trigger.get('ignore-deletes', True)
-                f = EventFilter(
-                    trigger=self,
-                    types=toList(trigger['event']),
-                    branches=toList(trigger.get('branch')),
-                    refs=toList(trigger.get('ref')),
-                    event_approvals=approvals,
-                    comments=comments,
-                    emails=emails,
-                    usernames=usernames,
-                    required_approvals=(
-                        toList(trigger.get('require-approval'))
-                    ),
-                    reject_approvals=toList(
-                        trigger.get('reject-approval')
-                    ),
-                    ignore_deletes=ignore_deletes
-                )
-                efilters.append(f)
+        for trigger in toList(trigger_conf):
+            approvals = {}
+            for approval_dict in toList(trigger.get('approval')):
+                for key, val in approval_dict.items():
+                    approvals[key] = val
+            # Backwards compat for *_filter versions of these args
+            comments = toList(trigger.get('comment'))
+            if not comments:
+                comments = toList(trigger.get('comment_filter'))
+            emails = toList(trigger.get('email'))
+            if not emails:
+                emails = toList(trigger.get('email_filter'))
+            usernames = toList(trigger.get('username'))
+            if not usernames:
+                usernames = toList(trigger.get('username_filter'))
+            ignore_deletes = trigger.get('ignore-deletes', True)
+            f = EventFilter(
+                trigger=self,
+                types=toList(trigger['event']),
+                branches=toList(trigger.get('branch')),
+                refs=toList(trigger.get('ref')),
+                event_approvals=approvals,
+                comments=comments,
+                emails=emails,
+                usernames=usernames,
+                required_approvals=(
+                    toList(trigger.get('require-approval'))
+                ),
+                reject_approvals=toList(
+                    trigger.get('reject-approval')
+                ),
+                ignore_deletes=ignore_deletes
+            )
+            efilters.append(f)
 
         return efilters
 
 
-def validate_trigger(trigger_data):
+def validate_conf(trigger_conf):
     """Validates the layout's trigger data."""
     events_with_ref = ('ref-updated', )
-    for event in trigger_data['gerrit']:
+    for event in trigger_conf:
         if event['event'] not in events_with_ref and event.get('ref', False):
-            raise voluptuous.Invalid(
+            raise v.Invalid(
                 "The event %s does not include ref information, Zuul cannot "
                 "use ref filter 'ref: %s'" % (event['event'], event['ref']))
+
+
+def getSchema():
+    def toList(x):
+        return v.Any([x], x)
+    variable_dict = v.Schema({}, extra=True)
+
+    approval = v.Schema({'username': str,
+                         'email-filter': str,
+                         'email': str,
+                         'older-than': str,
+                         'newer-than': str,
+                         }, extra=True)
+
+    gerrit_trigger = {
+        v.Required('event'):
+            toList(v.Any('patchset-created',
+                         'draft-published',
+                         'change-abandoned',
+                         'change-restored',
+                         'change-merged',
+                         'comment-added',
+                         'ref-updated')),
+        'comment_filter': toList(str),
+        'comment': toList(str),
+        'email_filter': toList(str),
+        'email': toList(str),
+        'username_filter': toList(str),
+        'username': toList(str),
+        'branch': toList(str),
+        'ref': toList(str),
+        'ignore-deletes': bool,
+        'approval': toList(variable_dict),
+        'require-approval': toList(approval),
+        'reject-approval': toList(approval),
+    }
+
+    return gerrit_trigger

zuul/trigger/timer.py

@@ -15,6 +15,7 @@
 
 import apscheduler.scheduler
 import logging
+import voluptuous as v
 from zuul.model import EventFilter, TriggerEvent
 from zuul.trigger import BaseTrigger
 
@@ -23,9 +24,8 @@ class TimerTrigger(BaseTrigger):
     name = 'timer'
     log = logging.getLogger("zuul.Timer")
 
-    def __init__(self, config, sched):
-        self.sched = sched
-        self.config = config
+    def __init__(self, trigger_config={}, sched=None, connection=None):
+        super(TimerTrigger, self).__init__(trigger_config, sched, connection)
         self.apsched = apscheduler.scheduler.Scheduler()
         self.apsched.start()
 
@@ -39,8 +39,8 @@ class TimerTrigger(BaseTrigger):
             self.log.debug("Adding event %s" % event)
             self.sched.addEvent(event)
 
-    def stop(self):
-        self.apsched.shutdown()
+    def _shutdown(self):
+        self.apsched.stop()
 
     def getEventFilters(self, trigger_conf):
         def toList(item):
@@ -51,13 +51,12 @@ class TimerTrigger(BaseTrigger):
             return [item]
 
         efilters = []
-        if 'timer' in trigger_conf:
-            for trigger in toList(trigger_conf['timer']):
-                f = EventFilter(trigger=self,
-                                types=['timer'],
-                                timespecs=toList(trigger['time']))
+        for trigger in toList(trigger_conf):
+            f = EventFilter(trigger=self,
+                            types=['timer'],
+                            timespecs=toList(trigger['time']))
 
-                efilters.append(f)
+            efilters.append(f)
 
         return efilters
 
@@ -90,3 +89,8 @@ class TimerTrigger(BaseTrigger):
                                               second=second,
                                               args=(pipeline.name,
                                                     timespec,))
+
+
+def getSchema():
+    timer_trigger = {v.Required('time'): str}
+    return timer_trigger

zuul/trigger/zuultrigger.py

@@ -14,6 +14,7 @@
 # under the License.
 
 import logging
+import voluptuous as v
 from zuul.model import EventFilter, TriggerEvent
 from zuul.trigger import BaseTrigger
 
@@ -22,9 +23,8 @@ class ZuulTrigger(BaseTrigger):
     name = 'zuul'
     log = logging.getLogger("zuul.ZuulTrigger")
 
-    def __init__(self, config, sched):
-        self.sched = sched
-        self.config = config
+    def __init__(self, trigger_config={}, sched=None, connection=None):
+        super(ZuulTrigger, self).__init__(trigger_config, sched, connection)
         self._handle_parent_change_enqueued_events = False
         self._handle_project_change_merged_events = False
 
@@ -37,28 +37,27 @@ class ZuulTrigger(BaseTrigger):
             return [item]
 
         efilters = []
-        if 'zuul' in trigger_conf:
-            for trigger in toList(trigger_conf['zuul']):
-                f = EventFilter(
-                    trigger=self,
-                    types=toList(trigger['event']),
-                    pipelines=toList(trigger.get('pipeline')),
-                    required_approvals=(
-                        toList(trigger.get('require-approval'))
-                    ),
-                    reject_approvals=toList(
-                        trigger.get('reject-approval')
-                    ),
-                )
-                efilters.append(f)
+        for trigger in toList(trigger_conf):
+            f = EventFilter(
+                trigger=self,
+                types=toList(trigger['event']),
+                pipelines=toList(trigger.get('pipeline')),
+                required_approvals=(
+                    toList(trigger.get('require-approval'))
+                ),
+                reject_approvals=toList(
+                    trigger.get('reject-approval')
+                ),
+            )
+            efilters.append(f)
 
         return efilters
 
-    def onChangeMerged(self, change):
+    def onChangeMerged(self, change, source):
         # Called each time zuul merges a change
         if self._handle_project_change_merged_events:
             try:
-                self._createProjectChangeMergedEvents(change)
+                self._createProjectChangeMergedEvents(change, source)
             except Exception:
                 self.log.exception(
                     "Unable to create project-change-merged events for "
@@ -74,8 +73,8 @@ class ZuulTrigger(BaseTrigger):
                     "Unable to create parent-change-enqueued events for "
                     "%s in %s" % (change, pipeline))
 
-    def _createProjectChangeMergedEvents(self, change):
-        changes = self.sched.sources['gerrit'].getProjectOpenChanges(
+    def _createProjectChangeMergedEvents(self, change, source):
+        changes = source.getProjectOpenChanges(
             change.project)
         for open_change in changes:
             self._createProjectChangeMergedEvent(open_change)
@@ -124,3 +123,26 @@ class ZuulTrigger(BaseTrigger):
                     self._handle_parent_change_enqueued_events = True
                 elif 'project-change-merged' in ef._types:
                     self._handle_project_change_merged_events = True
+
+
+def getSchema():
+    def toList(x):
+        return v.Any([x], x)
+
+    approval = v.Schema({'username': str,
+                         'email-filter': str,
+                         'email': str,
+                         'older-than': str,
+                         'newer-than': str,
+                         }, extra=True)
+
+    zuul_trigger = {
+        v.Required('event'):
+        toList(v.Any('parent-change-enqueued',
+                     'project-change-merged')),
+        'pipeline': toList(str),
+        'require-approval': toList(approval),
+        'reject-approval': toList(approval),
+    }
+
+    return zuul_trigger