diff --git a/.zuul.yaml b/.zuul.yaml
index 2eadd4f..ede4391 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -45,8 +45,7 @@
     name: openstack-infra/zuul
     check:
       jobs:
-        - build-openstack-infra-sphinx-docs:
-            success-url: 'html/feature/zuulv3/'
+        - build-openstack-sphinx-docs:
             irrelevant-files:
               - zuul/cmd/migrate.py
               - playbooks/zuul-migrate/.*
@@ -67,8 +66,10 @@
               - playbooks/zuul-migrate/.*
     gate:
       jobs:
-        - build-openstack-infra-sphinx-docs:
-            success-url: 'html/feature/zuulv3/'
+        - build-openstack-sphinx-docs:
+            irrelevant-files:
+              - zuul/cmd/migrate.py
+              - playbooks/zuul-migrate/.*
         - tox-pep8
         - tox-py35:
             irrelevant-files:
@@ -77,5 +78,5 @@
         - zuul-stream-functional
     post:
       jobs:
-        - publish-openstack-python-docs-infra
+        - publish-openstack-sphinx-docs-infra
         - publish-openstack-python-branch-tarball
diff --git a/doc/source/admin/components.rst b/doc/source/admin/components.rst
index fbb8cbc..b20aba7 100644
--- a/doc/source/admin/components.rst
+++ b/doc/source/admin/components.rst
@@ -101,6 +101,27 @@
 
       An openssl file containing the client private key in PEM format.
 
+.. attr:: statsd
+
+   Information about the optional statsd server.  If the ``statsd``
+   python module is installed and this section is configured,
+   statistics will be reported to statsd.  See :ref:`statsd` for more
+   information.
+
+   .. attr:: server
+
+      Hostname or IP address of the statsd server.
+
+   .. attr:: port
+      :default: 8125
+
+      The UDP port on which the statsd server is listening.
+
+   .. attr:: prefix
+
+      If present, this will be prefixed to all of the keys before
+      transmitting to the statsd server.
+
 .. NOTE: this is a white lie at this point, since only the scheduler
    uses this, however, we expect other components to use it later, so
    it's reasonable for admins to plan for this now.
@@ -115,6 +136,11 @@
       A list of zookeeper hosts for Zuul to use when communicating
       with Nodepool.
 
+   .. attr:: session_timeout
+      :default: 10.0
+
+      The ZooKeeper session timeout, in seconds.
+
 
 .. _scheduler:
 
@@ -260,6 +286,22 @@
 
       Directory in which Zuul should clone git repositories.
 
+   .. attr:: git_http_low_speed_limit
+      :default: 1000
+
+      If the HTTP transfer speed is less then git_http_low_speed_limit for
+      longer then git_http_low_speed_time, the transfer is aborted.
+
+      Value in bytes, setting to 0 will disable.
+
+   .. attr:: git_http_low_speed_time
+      :default: 30
+
+      If the HTTP transfer speed is less then git_http_low_speed_limit for
+      longer then git_http_low_speed_time, the transfer is aborted.
+
+      Value in seconds, setting to 0 will disable.
+
    .. attr:: git_user_email
 
       Value to pass to `git config user.email
@@ -463,6 +505,32 @@
       `bubblewrap` has become integral to securely operating Zuul.  If you
       have a valid use case for it, we encourage you to let us know.
 
+   .. attr:: load_multiplier
+      :default: 2.5
+
+      When an executor host gets too busy, the system may suffer
+      timeouts and other ill effects. The executor will stop accepting
+      more than 1 job at a time until load has lowered below a safe
+      level.  This level is determined by multiplying the number of
+      CPU's by `load_multiplier`.
+
+      So for example, if the system has 2 CPUs, and load_multiplier
+      is 2.5, the safe load for the system is 5.00. Any time the
+      system load average is over 5.00, the executor will quit
+      accepting multiple jobs at one time.
+
+      The executor will observe system load and determine whether
+      to accept more jobs every 30 seconds.
+
+   .. attr:: hostname
+      :default: hostname of the server
+
+      The executor needs to know its hostname under which it is reachable by
+      zuul-web. Otherwise live console log streaming doesn't work. In most cases
+      This is automatically detected correctly. But when running in environments
+      where it cannot determine its hostname correctly this can be overridden
+      here.
+
 .. attr:: merger
 
    .. attr:: git_user_email
diff --git a/doc/source/admin/monitoring.rst b/doc/source/admin/monitoring.rst
index 4fed1f9..55f1908 100644
--- a/doc/source/admin/monitoring.rst
+++ b/doc/source/admin/monitoring.rst
@@ -3,6 +3,8 @@
 Monitoring
 ==========
 
+.. _statsd:
+
 Statsd reporting
 ----------------
 
@@ -13,37 +15,24 @@
 Configuration
 ~~~~~~~~~~~~~
 
-Statsd support uses the statsd python module. Note that Zuul will start without
-the statsd python module, so an existing Zuul installation may be missing it.
+Statsd support uses the ``statsd`` python module.  Note that support
+is optional and Zuul will start without the statsd python module
+present.
 
-The configuration is done via environment variables STATSD_HOST and
-STATSD_PORT. They are interpreted by the statsd module directly and there is no
-such parameter in zuul.conf yet. Your init script will have to initialize both
-of them before executing Zuul.
-
-Your init script most probably loads a configuration file named
-``/etc/default/zuul`` which would contain the environment variables::
-
-  $ cat /etc/default/zuul
-  STATSD_HOST=10.0.0.1
-  STATSD_PORT=8125
+Configuration is in the :attr:`statsd` section of ``zuul.conf``.
 
 Metrics
 ~~~~~~~
 
 These metrics are emitted by the Zuul :ref:`scheduler`:
 
-.. stat:: gerrit.event.<type>
+.. stat:: zuul.event.<driver>.event.<type>
    :type: counter
 
-   Gerrit emits different kinds of messages over its `stream-events`
-   interface.  Zuul will report counters for each type of event it
-   receives from Gerrit.
+   Zuul will report counters for each type of event it receives from
+   each of its configured drivers.
 
-   Refer to your Gerrit installation documentation for a complete
-   list of Gerrit event types.
-
-.. stat:: zuul.pipeline
+.. stat:: zuul.tenant.<tenant>.pipeline
 
    Holds metrics specific to jobs. This hierarchy includes:
 
@@ -63,22 +52,60 @@
          The number of items currently being processed by this
          pipeline.
 
-      .. stat:: job
+      .. stat:: project
 
-         Subtree detailing per jobs statistics:
+         This hierarchy holds more specific metrics for each project
+         participating in the pipeline.
 
-         .. stat:: <jobname>
+         .. stat:: <canonical_hostname>
 
-            The triggered job name.
+            The canonical hostname for the triggering project.
+            Embedded ``.`` characters will be translated to ``_``.
 
-            .. stat:: <result>
-               :type: counter, timer
+            .. stat:: <project>
 
-               A counter for each type of result (e.g., ``SUCCESS`` or
-               ``FAILURE``, ``ERROR``, etc.) for the job.  If the
-               result is ``SUCCESS`` or ``FAILURE``, Zuul will
-               additionally report the duration of the build as a
-               timer.
+               The name of the triggering project.  Embedded ``/`` or
+               ``.`` characters will be translated to ``_``.
+
+               .. stat:: <branch>
+
+                  The name of the triggering branch.  Embedded ``/`` or
+                  ``.`` characters will be translated to ``_``.
+
+                  .. stat:: job
+
+                     Subtree detailing per-project job statistics:
+
+                     .. stat:: <jobname>
+
+                        The triggered job name.
+
+                        .. stat:: <result>
+                           :type: counter, timer
+
+                           A counter for each type of result (e.g., ``SUCCESS`` or
+                           ``FAILURE``, ``ERROR``, etc.) for the job.  If the
+                           result is ``SUCCESS`` or ``FAILURE``, Zuul will
+                           additionally report the duration of the build as a
+                           timer.
+
+                  .. stat:: current_changes
+                     :type: gauge
+
+                     The number of items of this project currently being
+                     processed by this pipeline.
+
+                  .. stat:: resident_time
+                     :type: timer
+
+                     A timer metric reporting how long each item for this
+                     project has been in the pipeline.
+
+                  .. stat:: total_changes
+                     :type: counter
+
+                     The number of changes for this project processed by the
+                     pipeline since Zuul started.
 
       .. stat:: resident_time
          :type: timer
@@ -98,34 +125,111 @@
          How long each item spent in the pipeline before its first job
          started.
 
-      .. stat:: <project>
+.. stat:: zuul.executor.<executor>
 
-         This hierarchy holds more specific metrics for each project
-         participating in the pipeline.  If the project name contains
-         a ``/`` character, it will be replaced with a ``.``.
+   Holds metrics emitted by individual executors.  The ``<executor>``
+   component of the key will be replaced with the hostname of the
+   executor.
 
-         .. stat:: current_changes
-            :type: gauge
+   .. stat:: builds
+      :type: counter
 
-            The number of items of this project currently being
-            processed by this pipeline.
+      Incremented each time the executor starts a build.
 
-         .. stat:: resident_time
-            :type: timer
+   .. stat:: running_builds
+      :type: gauge
 
-            A timer metric reporting how long each item for this
-            project has been in the pipeline.
+      The number of builds currently running on this executor.
 
-         .. stat:: total_changes
-            :type: counter
+   .. stat:: load_average
+      :type: gauge
 
-            The number of changes for this project processed by the
-            pipeline since Zuul started.
+      The one-minute load average of this executor, multiplied by 100.
 
-As an example, given a job named `myjob` triggered by the `gate` pipeline
-which took 40 seconds to build, the Zuul scheduler will emit the following
-statsd events:
+.. stat:: zuul.nodepool
 
-  * ``zuul.pipeline.gate.job.myjob.SUCCESS`` +1
-  * ``zuul.pipeline.gate.job.myjob``  40 seconds
-  * ``zuul.pipeline.gate.all_jobs`` +1
+   Holds metrics related to Zuul requests from Nodepool.
+
+   .. stat:: requested
+      :type: counter
+
+      Incremented each time a node request is submitted to Nodepool.
+
+      .. stat:: label.<label>
+         :type: counter
+
+         Incremented each time a request for a specific label is
+         submitted to Nodepool.
+
+      .. stat:: size.<size>
+         :type: counter
+
+         Incremented each time a request of a specific size is submitted
+         to Nodepool.  For example, a request for 3 nodes would use the
+         key ``zuul.nodepool.requested.size.3``.
+
+   .. stat:: canceled
+      :type: counter, timer
+
+      The counter is incremented each time a node request is canceled
+      by Zuul.  The timer records the elapsed time from request to
+      cancelation.
+
+      .. stat:: label.<label>
+         :type: counter, timer
+
+         The same, for a specific label.
+
+      .. stat:: size.<size>
+         :type: counter, timer
+
+         The same, for a specific request size.
+
+   .. stat:: fulfilled
+      :type: counter, timer
+
+      The counter is incremented each time a node request is fulfilled
+      by Nodepool.  The timer records the elapsed time from request to
+      fulfillment.
+
+      .. stat:: label.<label>
+         :type: counter, timer
+
+         The same, for a specific label.
+
+      .. stat:: size.<size>
+         :type: counter, timer
+
+         The same, for a specific request size.
+
+   .. stat:: failed
+      :type: counter, timer
+
+      The counter is incremented each time Nodepool fails to fulfill a
+      node request.  The timer records the elapsed time from request
+      to failure.
+
+      .. stat:: label.<label>
+         :type: counter, timer
+
+         The same, for a specific label.
+
+      .. stat:: size.<size>
+         :type: counter, timer
+
+         The same, for a specific request size.
+
+   .. stat:: current_requests
+      :type: gauge
+
+      The number of outstanding nodepool requests from Zuul.
+
+
+As an example, given a job named `myjob` in `mytenant` triggered by a
+change to `myproject` on the `master` branch in the `gate` pipeline
+which took 40 seconds to build, the Zuul scheduler will emit the
+following statsd events:
+
+  * ``zuul.tenant.mytenant.pipeline.gate.project.example_com.myproject.master.job.myjob.SUCCESS`` +1
+  * ``zuul.tenant.mytenant.pipeline.gate.project.example_com.myproject.master.job.myjob.SUCCESS``  40 seconds
+  * ``zuul.tenant.mytenant.pipeline.gate.all_jobs`` +1
diff --git a/doc/source/user/config.rst b/doc/source/user/config.rst
index 025ea71..c4014e2 100644
--- a/doc/source/user/config.rst
+++ b/doc/source/user/config.rst
@@ -653,10 +653,22 @@
         configuration, Zuul reads the ``master`` branch of a given
         project first, then other branches in alphabetical order.
 
+      * In the case of a job variant defined within a :ref:`project`,
+        if the project definition is in a :term:`config-project`, no
+        implied branch specifier is used.  If it appears in an
+        :term:`untrusted-project`, with no branch specifier, the
+        branch containing the project definition is used as an implied
+        branch specifier.
+
+      * In the case of a job variant defined within a
+        :ref:`project-template`, if no branch specifier appears, the
+        implied branch specifier for the :ref:`project` definition which
+        uses the project-template will be used.
+
       * Any further job variants other than the reference definition
         in an untrusted-project will, if they do not have a branch
-        specifier, will have an implied branch specifier for the
-        current branch applied.
+        specifier, have an implied branch specifier for the current
+        branch applied.
 
       This allows for the very simple and expected workflow where if a
       project defines a job on the ``master`` branch with no branch
@@ -940,12 +952,12 @@
 
       A boolean value which indicates whether this job may only be
       used in pipelines where :attr:`pipeline.post-review` is
-      ``true``.  This is automatically set to ``true`` if this job is
-      defined in a :term:`untrusted-project`.  It may be explicitly
-      set to obtain the same behavior for jobs defined in
-      :term:`config projects <config-project>`.  Once this is set to
-      ``true`` anywhere in the inheritance hierarchy for a job, it
-      will remain set for all child jobs and variants (it can not be
+      ``true``.  This is automatically set to ``true`` if this job
+      uses a :ref:`secret` and is defined in a :term:`untrusted-project`.
+      It may be explicitly set to obtain the same behavior for jobs
+      defined in :term:`config projects <config-project>`.  Once this
+      is set to ``true`` anywhere in the inheritance hierarchy for a job,
+      it will remain set for all child jobs and variants (it can not be
       set to ``false``).
 
 .. _project:
diff --git a/doc/source/user/jobs.rst b/doc/source/user/jobs.rst
index 3d24f5d..cf607b9 100644
--- a/doc/source/user/jobs.rst
+++ b/doc/source/user/jobs.rst
@@ -244,6 +244,18 @@
          The path to the source code, relative to the work dir.  E.g.,
          `src/git.example.com/org/project`.
 
+      .. var:: required
+
+         A boolean indicating whether this project appears in the
+         :attr:`job.required-projects` list for this job.
+
+   .. var:: _projects
+      :type: dict
+
+      The same as ``projects`` but a dictionary indexed by the
+      ``name`` value of each entry.  ``projects`` will be converted to
+      this.
+
    .. var:: tenant
 
       The name of the current Zuul tenant.
diff --git a/etc/status/public_html/jquery.zuul.js b/etc/status/public_html/jquery.zuul.js
index 70e999e..50dbed5 100644
--- a/etc/status/public_html/jquery.zuul.js
+++ b/etc/status/public_html/jquery.zuul.js
@@ -53,6 +53,7 @@
             'msg_id': '#zuul_msg',
             'pipelines_id': '#zuul_pipelines',
             'queue_events_num': '#zuul_queue_events_num',
+            'queue_management_events_num': '#zuul_queue_management_events_num',
             'queue_results_num': '#zuul_queue_results_num',
         }, options);
 
@@ -281,38 +282,38 @@
 
             change_header: function(change) {
                 var change_id = change.id || 'NA';
-                if (change_id.length === 40) {
-                    change_id = change_id.substr(0, 7);
-                }
 
                 var $change_link = $('<small />');
                 if (change.url !== null) {
-                    var github_id = change.id.match(/^([0-9]+),([0-9a-f]{40})$/);
+                    var github_id = change_id.match(/^([0-9]+),([0-9a-f]{40})$/);
                     if (github_id) {
                         $change_link.append(
                             $('<a />').attr('href', change.url).append(
                                 $('<abbr />')
-                                    .attr('title', change.id)
+                                    .attr('title', change_id)
                                     .text('#' + github_id[1])
                             )
                         );
-                    } else if (/^[0-9a-f]{40}$/.test(change.id)) {
-                        var change_id_short = change.id.slice(0, 7);
+                    } else if (/^[0-9a-f]{40}$/.test(change_id)) {
+                        var change_id_short = change_id.slice(0, 7);
                         $change_link.append(
                             $('<a />').attr('href', change.url).append(
                                 $('<abbr />')
-                                    .attr('title', change.id)
+                                    .attr('title', change_id)
                                     .text(change_id_short)
                             )
                         );
                     }
                     else {
                         $change_link.append(
-                            $('<a />').attr('href', change.url).text(change.id)
+                            $('<a />').attr('href', change.url).text(change_id)
                         );
                     }
                 }
                 else {
+                    if (change_id.length === 40) {
+                        change_id = change_id.substr(0, 7);
+                    }
                     $change_link.text(change_id);
                 }
 
@@ -713,6 +714,10 @@
                             data.trigger_event_queue ?
                                 data.trigger_event_queue.length : '0'
                         );
+                        $(options.queue_management_events_num).text(
+                            data.management_event_queue ?
+                                data.management_event_queue.length : '0'
+                        );
                         $(options.queue_results_num).text(
                             data.result_event_queue ?
                                 data.result_event_queue.length : '0'
diff --git a/etc/status/public_html/zuul.app.js b/etc/status/public_html/zuul.app.js
index ae950e8..7ceb2dd 100644
--- a/etc/status/public_html/zuul.app.js
+++ b/etc/status/public_html/zuul.app.js
@@ -33,7 +33,7 @@
         + '<div class="zuul-container" id="zuul-container">'
         + '<div style="display: none;" class="alert" id="zuul_msg"></div>'
         + '<button class="btn pull-right zuul-spinner">updating <span class="glyphicon glyphicon-refresh"></span></button>'
-        + '<p>Queue lengths: <span id="zuul_queue_events_num">0</span> events, <span id="zuul_queue_results_num">0</span> results.</p>'
+        + '<p>Queue lengths: <span id="zuul_queue_events_num">0</span> events, <span id="zuul_queue_management_events_num">0</span> management events, <span id="zuul_queue_results_num">0</span> results.</p>'
         + '<div id="zuul_controls"></div>'
         + '<div id="zuul_pipelines" class="row"></div>'
         + '<p>Zuul version: <span id="zuul-version-span"></span></p>'
diff --git a/etc/zuul.conf-sample b/etc/zuul.conf-sample
index ba7aace..76494ad 100644
--- a/etc/zuul.conf-sample
+++ b/etc/zuul.conf-sample
@@ -5,6 +5,9 @@
 ;ssl_cert=/path/to/client.pem
 ;ssl_key=/path/to/client.key
 
+[statsd]
+server=127.0.0.1
+
 [zookeeper]
 hosts=127.0.0.1:2181
 
diff --git a/requirements.txt b/requirements.txt
index cdffda2..4b8be3c 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,12 +2,16 @@
 
 # pull from master until https://github.com/sigmavirus24/github3.py/pull/671
 # is in a release
--e git+https://github.com/sigmavirus24/github3.py.git@develop#egg=Github3.py
+git+https://github.com/sigmavirus24/github3.py.git@develop#egg=Github3.py
 PyYAML>=3.1.0
 Paste
 WebOb>=1.2.3
 paramiko>=1.8.0,<2.0.0
-GitPython>=0.3.3,<2.1.2
+# Using a local fork of gitpython until at least these changes are in a
+# release.
+# https://github.com/gitpython-developers/GitPython/pull/682
+# https://github.com/gitpython-developers/GitPython/pull/686
+git+https://github.com/jeblair/GitPython.git@zuul#egg=GitPython
 python-daemon>=2.0.4,<2.1.0
 extras
 statsd>=1.0.0,<3.0
diff --git a/tests/base.py b/tests/base.py
index 2e3d682..a02ee5a 100755
--- a/tests/base.py
+++ b/tests/base.py
@@ -20,7 +20,6 @@
 import datetime
 import gc
 import hashlib
-import importlib
 from io import StringIO
 import json
 import logging
@@ -48,7 +47,6 @@
 import kazoo.client
 import kazoo.exceptions
 import pymysql
-import statsd
 import testtools
 import testtools.content
 import testtools.content_type
@@ -1363,6 +1361,63 @@
         return repos
 
 
+class RecordingAnsibleJob(zuul.executor.server.AnsibleJob):
+    def doMergeChanges(self, merger, items, repo_state):
+        # Get a merger in order to update the repos involved in this job.
+        commit = super(RecordingAnsibleJob, self).doMergeChanges(
+            merger, items, repo_state)
+        if not commit:  # merge conflict
+            self.recordResult('MERGER_FAILURE')
+        return commit
+
+    def recordResult(self, result):
+        build = self.executor_server.job_builds[self.job.unique]
+        self.executor_server.lock.acquire()
+        self.executor_server.build_history.append(
+            BuildHistory(name=build.name, result=result, changes=build.changes,
+                         node=build.node, uuid=build.unique,
+                         ref=build.parameters['zuul']['ref'],
+                         parameters=build.parameters, jobdir=build.jobdir,
+                         pipeline=build.parameters['zuul']['pipeline'])
+        )
+        self.executor_server.running_builds.remove(build)
+        del self.executor_server.job_builds[self.job.unique]
+        self.executor_server.lock.release()
+
+    def runPlaybooks(self, args):
+        build = self.executor_server.job_builds[self.job.unique]
+        build.jobdir = self.jobdir
+
+        result = super(RecordingAnsibleJob, self).runPlaybooks(args)
+        self.recordResult(result)
+        return result
+
+    def runAnsible(self, cmd, timeout, playbook, wrapped=True):
+        build = self.executor_server.job_builds[self.job.unique]
+
+        if self.executor_server._run_ansible:
+            result = super(RecordingAnsibleJob, self).runAnsible(
+                cmd, timeout, playbook, wrapped)
+        else:
+            if playbook.path:
+                result = build.run()
+            else:
+                result = (self.RESULT_NORMAL, 0)
+        return result
+
+    def getHostList(self, args):
+        self.log.debug("hostlist")
+        hosts = super(RecordingAnsibleJob, self).getHostList(args)
+        for host in hosts:
+            host['host_vars']['ansible_connection'] = 'local'
+
+        hosts.append(dict(
+            name='localhost',
+            host_vars=dict(ansible_connection='local'),
+            host_keys=[]))
+        return hosts
+
+
 class RecordingExecutorServer(zuul.executor.server.ExecutorServer):
     """An Ansible executor to be used in tests.
 
@@ -1374,6 +1429,9 @@
         be explicitly released.
 
     """
+
+    _job_class = RecordingAnsibleJob
+
     def __init__(self, *args, **kw):
         self._run_ansible = kw.pop('_run_ansible', False)
         self._test_root = kw.pop('_test_root', False)
@@ -1428,8 +1486,7 @@
         args = json.loads(job.arguments)
         args['zuul']['_test'] = dict(test_root=self._test_root)
         job.arguments = json.dumps(args)
-        self.job_workers[job.unique] = RecordingAnsibleJob(self, job)
-        self.job_workers[job.unique].run()
+        super(RecordingExecutorServer, self).executeJob(job)
 
     def stopJob(self, job):
         self.log.debug("handle stop")
@@ -1447,63 +1504,6 @@
         super(RecordingExecutorServer, self).stop()
 
 
-class RecordingAnsibleJob(zuul.executor.server.AnsibleJob):
-    def doMergeChanges(self, merger, items, repo_state):
-        # Get a merger in order to update the repos involved in this job.
-        commit = super(RecordingAnsibleJob, self).doMergeChanges(
-            merger, items, repo_state)
-        if not commit:  # merge conflict
-            self.recordResult('MERGER_FAILURE')
-        return commit
-
-    def recordResult(self, result):
-        build = self.executor_server.job_builds[self.job.unique]
-        self.executor_server.lock.acquire()
-        self.executor_server.build_history.append(
-            BuildHistory(name=build.name, result=result, changes=build.changes,
-                         node=build.node, uuid=build.unique,
-                         ref=build.parameters['zuul']['ref'],
-                         parameters=build.parameters, jobdir=build.jobdir,
-                         pipeline=build.parameters['zuul']['pipeline'])
-        )
-        self.executor_server.running_builds.remove(build)
-        del self.executor_server.job_builds[self.job.unique]
-        self.executor_server.lock.release()
-
-    def runPlaybooks(self, args):
-        build = self.executor_server.job_builds[self.job.unique]
-        build.jobdir = self.jobdir
-
-        result = super(RecordingAnsibleJob, self).runPlaybooks(args)
-        self.recordResult(result)
-        return result
-
-    def runAnsible(self, cmd, timeout, playbook):
-        build = self.executor_server.job_builds[self.job.unique]
-
-        if self.executor_server._run_ansible:
-            result = super(RecordingAnsibleJob, self).runAnsible(
-                cmd, timeout, playbook)
-        else:
-            if playbook.path:
-                result = build.run()
-            else:
-                result = (self.RESULT_NORMAL, 0)
-        return result
-
-    def getHostList(self, args):
-        self.log.debug("hostlist")
-        hosts = super(RecordingAnsibleJob, self).getHostList(args)
-        for host in hosts:
-            host['host_vars']['ansible_connection'] = 'local'
-
-        hosts.append(dict(
-            name='localhost',
-            host_vars=dict(ansible_connection='local'),
-            host_keys=[]))
-        return hosts
-
-
 class FakeGearmanServer(gear.Server):
     """A Gearman server for use in tests.
 
@@ -1518,6 +1518,7 @@
 
     def __init__(self, use_ssl=False):
         self.hold_jobs_in_queue = False
+        self.hold_merge_jobs_in_queue = False
         if use_ssl:
             ssl_ca = os.path.join(FIXTURE_DIR, 'gearman/root-ca.pem')
             ssl_cert = os.path.join(FIXTURE_DIR, 'gearman/server.pem')
@@ -1537,6 +1538,8 @@
                 if not hasattr(job, 'waiting'):
                     if job.name.startswith(b'executor:execute'):
                         job.waiting = self.hold_jobs_in_queue
+                    elif job.name.startswith(b'merger:'):
+                        job.waiting = self.hold_merge_jobs_in_queue
                     else:
                         job.waiting = False
                 if job.waiting:
@@ -1562,10 +1565,15 @@
                 len(self.low_queue))
         self.log.debug("releasing queued job %s (%s)" % (regex, qlen))
         for job in self.getQueue():
-            if job.name != b'executor:execute':
-                continue
-            parameters = json.loads(job.arguments.decode('utf8'))
-            if not regex or re.match(regex, parameters.get('job')):
+            match = False
+            if job.name == b'executor:execute':
+                parameters = json.loads(job.arguments.decode('utf8'))
+                if not regex or re.match(regex, parameters.get('job')):
+                    match = True
+            if job.name.startswith(b'merger:'):
+                if not regex:
+                    match = True
+            if match:
                 self.log.debug("releasing queued job %s" %
                                job.unique)
                 job.waiting = False
@@ -2039,14 +2047,9 @@
         self.config.set('executor', 'state_dir', self.executor_state_root)
 
         self.statsd = FakeStatsd()
-        # note, use 127.0.0.1 rather than localhost to avoid getting ipv6
-        # see: https://github.com/jsocol/pystatsd/issues/61
-        os.environ['STATSD_HOST'] = '127.0.0.1'
-        os.environ['STATSD_PORT'] = str(self.statsd.port)
+        if self.config.has_section('statsd'):
+            self.config.set('statsd', 'port', str(self.statsd.port))
         self.statsd.start()
-        # the statsd client object is configured in the statsd module import
-        importlib.reload(statsd)
-        importlib.reload(zuul.scheduler)
 
         self.gearman_server = FakeGearmanServer(self.use_ssl)
 
@@ -2381,7 +2384,7 @@
         # before noticing they should exit, but they should exit on their own.
         # Further the pydevd threads also need to be whitelisted so debugging
         # e.g. in PyCharm is possible without breaking shutdown.
-        whitelist = ['executor-watchdog',
+        whitelist = ['watchdog',
                      'pydevd.CommandThread',
                      'pydevd.Reader',
                      'pydevd.Writer',
@@ -2556,6 +2559,26 @@
             return False
         return True
 
+    def areAllMergeJobsWaiting(self):
+        for client_job in list(self.merge_client.jobs):
+            if not client_job.handle:
+                self.log.debug("%s has no handle" % client_job)
+                return False
+            server_job = self.gearman_server.jobs.get(client_job.handle)
+            if not server_job:
+                self.log.debug("%s is not known to the gearman server" %
+                               client_job)
+                return False
+            if not hasattr(server_job, 'waiting'):
+                self.log.debug("%s is being enqueued" % server_job)
+                return False
+            if server_job.waiting:
+                self.log.debug("%s is waiting" % server_job)
+                continue
+            self.log.debug("%s is not waiting" % server_job)
+            return False
+        return True
+
     def eventQueuesEmpty(self):
         for event_queue in self.event_queues:
             yield event_queue.empty()
@@ -2592,7 +2615,7 @@
                 # processed
                 self.eventQueuesJoin()
                 self.sched.run_handler_lock.acquire()
-                if (not self.merge_client.jobs and
+                if (self.areAllMergeJobsWaiting() and
                     self.haveAllBuildsReported() and
                     self.areAllBuildsWaiting() and
                     self.areAllNodeRequestsComplete() and
@@ -2620,6 +2643,13 @@
                 return build
         raise Exception("Unable to find build %s" % name)
 
+    def assertJobNotInHistory(self, name, project=None):
+        for job in self.history:
+            if (project is None or
+                job.parameters['zuul']['project']['name'] == project):
+                self.assertNotEqual(job.name, name,
+                                    'Job %s found in history' % name)
+
     def getJobFromHistory(self, name, project=None):
         for job in self.history:
             if (job.name == name and
diff --git a/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/post-logs.yaml b/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/post-logs.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/post-logs.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/post-ssh.yaml b/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/post-ssh.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/post-ssh.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/pre.yaml b/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/pre.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/project-config/playbooks/base/pre.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/branch-variants/git/project-config/zuul.yaml b/tests/fixtures/config/branch-variants/git/project-config/zuul.yaml
new file mode 100644
index 0000000..89d98a9
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/project-config/zuul.yaml
@@ -0,0 +1,25 @@
+- pipeline:
+    name: check
+    manager: independent
+    trigger:
+      gerrit:
+        - event: patchset-created
+    success:
+      gerrit:
+        Verified: 1
+    failure:
+      gerrit:
+        Verified: -1
+
+- job:
+    name: base
+    parent: null
+    pre-run: playbooks/base/pre
+    post-run:
+      - playbooks/base/post-ssh
+      - playbooks/base/post-logs
+
+- project:
+    name: project-config
+    check:
+      jobs: []
diff --git a/tests/fixtures/config/branch-variants/git/puppet-integration/.zuul.yaml b/tests/fixtures/config/branch-variants/git/puppet-integration/.zuul.yaml
new file mode 100644
index 0000000..2545208
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/puppet-integration/.zuul.yaml
@@ -0,0 +1,24 @@
+- job:
+    name: puppet-base
+    pre-run: playbooks/prepare-node-common
+
+- job:
+    name: puppet-module-base
+    parent: puppet-base
+    pre-run: playbooks/prepare-node-unit
+
+- job:
+    name: puppet-lint
+    parent: puppet-module-base
+    run: playbooks/run-lint
+
+- project-template:
+    name: puppet-check-jobs
+    check:
+      jobs:
+        - puppet-lint
+    
+- project:
+    name: puppet-integration
+    templates:
+      - puppet-check-jobs
diff --git a/tests/fixtures/config/branch-variants/git/puppet-integration/README b/tests/fixtures/config/branch-variants/git/puppet-integration/README
new file mode 100644
index 0000000..9daeafb
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/puppet-integration/README
@@ -0,0 +1 @@
+test
diff --git a/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/prepare-node-common.yaml b/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/prepare-node-common.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/prepare-node-common.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/prepare-node-unit.yaml b/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/prepare-node-unit.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/prepare-node-unit.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/run-lint.yaml b/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/run-lint.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/git/puppet-integration/playbooks/run-lint.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/branch-variants/main.yaml b/tests/fixtures/config/branch-variants/main.yaml
new file mode 100644
index 0000000..ad2b2f6
--- /dev/null
+++ b/tests/fixtures/config/branch-variants/main.yaml
@@ -0,0 +1,8 @@
+- tenant:
+    name: tenant-one
+    source:
+      gerrit:
+        config-projects:
+          - project-config
+        untrusted-projects:
+          - puppet-integration
diff --git a/tests/fixtures/config/job-output/git/common-config/playbooks/job-output-failure-post.yaml b/tests/fixtures/config/job-output/git/common-config/playbooks/job-output-failure-post.yaml
new file mode 100644
index 0000000..bd46718
--- /dev/null
+++ b/tests/fixtures/config/job-output/git/common-config/playbooks/job-output-failure-post.yaml
@@ -0,0 +1,4 @@
+- hosts: all
+  tasks:
+    - shell: echo "Failure test {{ zuul.executor.src_root }}"
+    - shell: exit 1
diff --git a/tests/fixtures/config/job-output/git/common-config/zuul.yaml b/tests/fixtures/config/job-output/git/common-config/zuul.yaml
index a83f0bc..f182d8d 100644
--- a/tests/fixtures/config/job-output/git/common-config/zuul.yaml
+++ b/tests/fixtures/config/job-output/git/common-config/zuul.yaml
@@ -20,8 +20,19 @@
     parent: base
     name: job-output
 
+- job:
+    name: job-output-failure
+    run: playbooks/job-output
+    post-run: playbooks/job-output-failure-post
+
 - project:
     name: org/project
     check:
       jobs:
         - job-output
+
+- project:
+    name: org/project2
+    check:
+      jobs:
+        - job-output-failure
diff --git a/tests/fixtures/config/job-output/git/org_project2/README b/tests/fixtures/config/job-output/git/org_project2/README
new file mode 100644
index 0000000..9daeafb
--- /dev/null
+++ b/tests/fixtures/config/job-output/git/org_project2/README
@@ -0,0 +1 @@
+test
diff --git a/tests/fixtures/config/job-output/main.yaml b/tests/fixtures/config/job-output/main.yaml
index 208e274..14b382f 100644
--- a/tests/fixtures/config/job-output/main.yaml
+++ b/tests/fixtures/config/job-output/main.yaml
@@ -6,3 +6,4 @@
           - common-config
         untrusted-projects:
           - org/project
+          - org/project2
diff --git a/tests/fixtures/config/post-playbook/git/common-config/playbooks/post.yaml b/tests/fixtures/config/post-playbook/git/common-config/playbooks/post.yaml
new file mode 100644
index 0000000..40b3ca7
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/git/common-config/playbooks/post.yaml
@@ -0,0 +1,13 @@
+- hosts: all
+  tasks:
+    - debug: var=waitpath
+    - file:
+        path: "{{zuul._test.test_root}}/{{zuul.build}}.post_start.flag"
+        state: touch
+    # Do not finish until test creates the flag file
+    - wait_for:
+        state: present
+        path: "{{waitpath}}"
+    - file:
+        path: "{{zuul._test.test_root}}/{{zuul.build}}.post_end.flag"
+        state: touch
diff --git a/tests/fixtures/config/post-playbook/git/common-config/playbooks/pre.yaml b/tests/fixtures/config/post-playbook/git/common-config/playbooks/pre.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/git/common-config/playbooks/pre.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/post-playbook/git/common-config/playbooks/python27.yaml b/tests/fixtures/config/post-playbook/git/common-config/playbooks/python27.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/git/common-config/playbooks/python27.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/post-playbook/git/common-config/zuul.yaml b/tests/fixtures/config/post-playbook/git/common-config/zuul.yaml
new file mode 100644
index 0000000..92a5515
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/git/common-config/zuul.yaml
@@ -0,0 +1,24 @@
+- pipeline:
+    name: check
+    manager: independent
+    post-review: true
+    trigger:
+      gerrit:
+        - event: patchset-created
+    success:
+      gerrit:
+        Verified: 1
+    failure:
+      gerrit:
+        Verified: -1
+
+- job:
+    name: base
+    parent: null
+
+- job:
+    name: python27
+    pre-run: playbooks/pre
+    post-run: playbooks/post
+    vars:
+      waitpath: '{{zuul._test.test_root}}/{{zuul.build}}/test_wait'
diff --git a/tests/fixtures/config/post-playbook/git/org_project/.zuul.yaml b/tests/fixtures/config/post-playbook/git/org_project/.zuul.yaml
new file mode 100644
index 0000000..89a5674
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/git/org_project/.zuul.yaml
@@ -0,0 +1,5 @@
+- project:
+    name: org/project
+    check:
+      jobs:
+        - python27
diff --git a/tests/fixtures/config/post-playbook/git/org_project/README b/tests/fixtures/config/post-playbook/git/org_project/README
new file mode 100644
index 0000000..9daeafb
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/git/org_project/README
@@ -0,0 +1 @@
+test
diff --git a/tests/fixtures/config/post-playbook/main.yaml b/tests/fixtures/config/post-playbook/main.yaml
new file mode 100644
index 0000000..6033879
--- /dev/null
+++ b/tests/fixtures/config/post-playbook/main.yaml
@@ -0,0 +1,9 @@
+- tenant:
+    name: tenant-one
+    source:
+      gerrit:
+        config-projects:
+          - common-config
+        untrusted-projects:
+          - org/project
+
diff --git a/tests/fixtures/config/single-tenant/git/common-config/zuul.yaml b/tests/fixtures/config/single-tenant/git/common-config/zuul.yaml
index 14f43f4..2160ef9 100644
--- a/tests/fixtures/config/single-tenant/git/common-config/zuul.yaml
+++ b/tests/fixtures/config/single-tenant/git/common-config/zuul.yaml
@@ -39,6 +39,7 @@
       gerrit:
         - event: ref-updated
           ref: ^(?!refs/).*$
+    precedence: low
 
 - job:
     name: base
diff --git a/tests/fixtures/config/sql-driver/git/common-config/playbooks/project-test3.yaml b/tests/fixtures/config/sql-driver/git/common-config/playbooks/project-test3.yaml
new file mode 100644
index 0000000..f679dce
--- /dev/null
+++ b/tests/fixtures/config/sql-driver/git/common-config/playbooks/project-test3.yaml
@@ -0,0 +1,2 @@
+- hosts: all
+  tasks: []
diff --git a/tests/fixtures/config/sql-driver/git/common-config/zuul.yaml b/tests/fixtures/config/sql-driver/git/common-config/zuul.yaml
index b8f4d67..8fce9e7 100644
--- a/tests/fixtures/config/sql-driver/git/common-config/zuul.yaml
+++ b/tests/fixtures/config/sql-driver/git/common-config/zuul.yaml
@@ -27,6 +27,9 @@
 - job:
     name: project-test2
 
+- job:
+    name: project-test3
+
 - project:
     name: org/project
     check:
@@ -36,3 +39,9 @@
             dependencies: project-merge
         - project-test2:
             dependencies: project-merge
+        # Make sure we have a "SKIPPED" result
+        - project-test3:
+            dependencies: project-test1
+        # The noop job can have timing quirks
+        - noop:
+            dependencies: project-test2
diff --git a/tests/fixtures/config/templated-project/git/untrusted-config/zuul.d/project.yaml b/tests/fixtures/config/templated-project/git/untrusted-config/zuul.d/project.yaml
new file mode 100644
index 0000000..6d1892c
--- /dev/null
+++ b/tests/fixtures/config/templated-project/git/untrusted-config/zuul.d/project.yaml
@@ -0,0 +1,4 @@
+- project:
+    name: untrusted-config
+    templates:
+      - test-one-and-two
diff --git a/tests/fixtures/config/templated-project/git/common-config/zuul.d/templates.yaml b/tests/fixtures/config/templated-project/git/untrusted-config/zuul.d/templates.yaml
similarity index 100%
rename from tests/fixtures/config/templated-project/git/common-config/zuul.d/templates.yaml
rename to tests/fixtures/config/templated-project/git/untrusted-config/zuul.d/templates.yaml
diff --git a/tests/fixtures/config/templated-project/main.yaml b/tests/fixtures/config/templated-project/main.yaml
index e59b396..bb59838 100644
--- a/tests/fixtures/config/templated-project/main.yaml
+++ b/tests/fixtures/config/templated-project/main.yaml
@@ -5,5 +5,6 @@
         config-projects:
           - common-config
         untrusted-projects:
+          - untrusted-config
           - org/templated-project
           - org/layered-project
diff --git a/tests/fixtures/fake_git.sh b/tests/fixtures/fake_git.sh
new file mode 100755
index 0000000..5b787b7
--- /dev/null
+++ b/tests/fixtures/fake_git.sh
@@ -0,0 +1,14 @@
+#!/bin/sh
+
+echo $*
+case "$1" in
+    clone)
+        dest=$3
+        mkdir -p $dest/.git
+        ;;
+    version)
+        echo "git version 1.0.0"
+        exit 0
+        ;;
+esac
+sleep 30
diff --git a/tests/fixtures/layouts/delayed-repo-init.yaml b/tests/fixtures/layouts/delayed-repo-init.yaml
index e97d37a..c89e2fa 100644
--- a/tests/fixtures/layouts/delayed-repo-init.yaml
+++ b/tests/fixtures/layouts/delayed-repo-init.yaml
@@ -67,7 +67,7 @@
             dependencies: project-merge
     gate:
       jobs:
-        - project-merge:
+        - project-merge
         - project-test1:
             dependencies: project-merge
         - project-test2:
diff --git a/tests/fixtures/layouts/job-vars.yaml b/tests/fixtures/layouts/job-vars.yaml
new file mode 100644
index 0000000..22fc5c2
--- /dev/null
+++ b/tests/fixtures/layouts/job-vars.yaml
@@ -0,0 +1,75 @@
+- pipeline:
+    name: check
+    manager: independent
+    trigger:
+      gerrit:
+        - event: patchset-created
+    success:
+      gerrit:
+        Verified: 1
+    failure:
+      gerrit:
+        Verified: -1
+
+- job:
+    name: base
+    parent: null
+
+- job:
+    name: parentjob
+    parent: base
+    required-projects:
+      - org/project0
+    vars:
+      override: 0
+      child1override: 0
+      parent: 0
+
+- job:
+    name: child1
+    parent: parentjob
+    required-projects:
+      - org/project1
+    vars:
+      override: 1
+      child1override: 1
+      child1: 1
+
+- job:
+    name: child2
+    parent: parentjob
+    required-projects:
+      - org/project2
+    vars:
+      override: 2
+      child2: 2
+
+- job:
+    name: child3
+    parent: parentjob
+
+- project:
+    name: org/project
+    check:
+      jobs:
+        - parentjob
+        - child1
+        - child2
+        - child3:
+            required-projects:
+              - org/project3
+            vars:
+              override: 3
+              child3: 3
+
+- project:
+    name: org/project0
+
+- project:
+    name: org/project1
+
+- project:
+    name: org/project2
+
+- project:
+    name: org/project3
diff --git a/tests/fixtures/layouts/matcher-test.yaml b/tests/fixtures/layouts/matcher-test.yaml
new file mode 100644
index 0000000..b511a2f
--- /dev/null
+++ b/tests/fixtures/layouts/matcher-test.yaml
@@ -0,0 +1,63 @@
+- pipeline:
+    name: check
+    manager: independent
+    trigger:
+      gerrit:
+        - event: patchset-created
+    success:
+      gerrit:
+        Verified: 1
+    failure:
+      gerrit:
+        Verified: -1
+
+- pipeline:
+    name: gate
+    manager: dependent
+    success-message: Build succeeded (gate).
+    trigger:
+      gerrit:
+        - event: comment-added
+          approval:
+            - Approved: 1
+    success:
+      gerrit:
+        Verified: 2
+        submit: true
+    failure:
+      gerrit:
+        Verified: -2
+    start:
+      gerrit:
+        Verified: 0
+    precedence: high
+
+- job:
+    name: base
+    parent: null
+
+- job:
+    name: project-test1
+    nodeset:
+      nodes:
+        - name: controller
+          label: label1
+
+- job:
+    name: ignore-branch
+    branches: ^(?!featureA).*$
+    nodeset:
+      nodes:
+        - name: controller
+          label: label2
+
+- project:
+    name: org/project
+    check:
+      jobs:
+        - project-test1
+        - ignore-branch
+    gate:
+      jobs:
+        - project-test1
+        - ignore-branch
diff --git a/tests/fixtures/layouts/multiple-templates.yaml b/tests/fixtures/layouts/multiple-templates.yaml
new file mode 100644
index 0000000..7272cad
--- /dev/null
+++ b/tests/fixtures/layouts/multiple-templates.yaml
@@ -0,0 +1,44 @@
+- pipeline:
+    name: check
+    manager: independent
+    trigger:
+      gerrit:
+        - event: patchset-created
+    success:
+      gerrit:
+        Verified: 1
+    failure:
+      gerrit:
+        Verified: -1
+
+- job:
+    name: base
+    parent: null
+
+- job:
+    name: py27
+
+- project-template:
+    name: python-jobs
+    check:
+      jobs:
+        - py27
+
+- project-template:
+    name: python-trusty-jobs
+    check:
+      jobs:
+        - py27:
+            tags:
+              - trusty
+
+- project:
+    name: org/project1
+    templates:
+      - python-jobs
+      - python-trusty-jobs
+
+- project:
+    name: org/project2
+    templates:
+      - python-jobs
diff --git a/tests/fixtures/zuul-executor-hostname.conf b/tests/fixtures/zuul-executor-hostname.conf
new file mode 100644
index 0000000..7db144d
--- /dev/null
+++ b/tests/fixtures/zuul-executor-hostname.conf
@@ -0,0 +1,32 @@
+[gearman]
+server=127.0.0.1
+
+[statsd]
+# note, use 127.0.0.1 rather than localhost to avoid getting ipv6
+# see: https://github.com/jsocol/pystatsd/issues/61
+server=127.0.0.1
+
+[scheduler]
+tenant_config=main.yaml
+
+[merger]
+git_dir=/tmp/zuul-test/merger-git
+git_user_email=zuul@example.com
+git_user_name=zuul
+
+[executor]
+git_dir=/tmp/zuul-test/executor-git
+hostname=test-executor-hostname.openstack.org
+
+[connection gerrit]
+driver=gerrit
+server=review.example.com
+user=jenkins
+sshkey=fake_id_rsa_path
+
+[connection smtp]
+driver=smtp
+server=localhost
+port=25
+default_from=zuul@example.com
+default_to=you@example.com
diff --git a/tests/fixtures/zuul.conf b/tests/fixtures/zuul.conf
index 7bc8c59..e6f997c 100644
--- a/tests/fixtures/zuul.conf
+++ b/tests/fixtures/zuul.conf
@@ -1,6 +1,11 @@
 [gearman]
 server=127.0.0.1
 
+[statsd]
+# note, use 127.0.0.1 rather than localhost to avoid getting ipv6
+# see: https://github.com/jsocol/pystatsd/issues/61
+server=127.0.0.1
+
 [scheduler]
 tenant_config=main.yaml
 
diff --git a/tests/print_layout.py b/tests/print_layout.py
index a295886..055270f 100644
--- a/tests/print_layout.py
+++ b/tests/print_layout.py
@@ -59,6 +59,14 @@
             if fn in ['zuul.yaml', '.zuul.yaml']:
                 print_file('File: ' + os.path.join(gitrepo, fn),
                            os.path.join(reporoot, fn))
+        for subdir in ['.zuul.d', 'zuul.d']:
+            zuuld = os.path.join(reporoot, subdir)
+            if not os.path.exists(zuuld):
+                continue
+            filenames = os.listdir(zuuld)
+            for fn in filenames:
+                print_file('File: ' + os.path.join(gitrepo, subdir, fn),
+                           os.path.join(zuuld, fn))
 
 
 if __name__ == '__main__':
diff --git a/tests/unit/test_change_matcher.py b/tests/unit/test_change_matcher.py
index 6b161a1..3d5345f 100644
--- a/tests/unit/test_change_matcher.py
+++ b/tests/unit/test_change_matcher.py
@@ -60,7 +60,7 @@
         self.assertTrue(self.matcher.matches(self.change))
 
     def test_matches_returns_true_on_matching_ref(self):
-        self.change.branch = 'bar'
+        delattr(self.change, 'branch')
         self.change.ref = 'foo'
         self.assertTrue(self.matcher.matches(self.change))
 
@@ -69,11 +69,6 @@
         self.change.ref = 'baz'
         self.assertFalse(self.matcher.matches(self.change))
 
-    def test_matches_returns_false_for_missing_attrs(self):
-        delattr(self.change, 'branch')
-        # ref is by default not an attribute
-        self.assertFalse(self.matcher.matches(self.change))
-
 
 class TestFileMatcher(BaseTestMatcher):
 
diff --git a/tests/unit/test_connection.py b/tests/unit/test_connection.py
index de3b7cf..3774cc3 100644
--- a/tests/unit/test_connection.py
+++ b/tests/unit/test_connection.py
@@ -74,6 +74,7 @@
 
     def test_sql_results(self):
         "Test results are entered into an sql table"
+        self.executor_server.hold_jobs_in_build = True
         # Grab the sa tables
         tenant = self.sched.abide.tenants.get('tenant-one')
         reporter = _get_reporter_from_connection_name(
@@ -85,6 +86,8 @@
         A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
         self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
         self.waitUntilSettled()
+        self.orderedRelease()
+        self.waitUntilSettled()
 
         # Add a failed result
         B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B')
@@ -92,6 +95,8 @@
         self.executor_server.failJob('project-test1', B)
         self.fake_gerrit.addEvent(B.getPatchsetCreatedEvent(1))
         self.waitUntilSettled()
+        self.orderedRelease()
+        self.waitUntilSettled()
 
         conn = self.connections.connections['resultsdb'].engine.connect()
         result = conn.execute(
@@ -143,15 +148,15 @@
             )
         ).fetchall()
 
-        # Check the second last result, which should be the project-test1 job
+        # Check the second result, which should be the project-test1 job
         # which failed
-        self.assertEqual('project-test1', buildset1_builds[-2]['job_name'])
-        self.assertEqual("FAILURE", buildset1_builds[-2]['result'])
+        self.assertEqual('project-test1', buildset1_builds[1]['job_name'])
+        self.assertEqual("FAILURE", buildset1_builds[1]['result'])
         self.assertEqual(
             'finger://{hostname}/{uuid}'.format(
                 hostname=self.executor_server.hostname,
-                uuid=buildset1_builds[-2]['uuid']),
-            buildset1_builds[-2]['log_url'])
+                uuid=buildset1_builds[1]['uuid']),
+            buildset1_builds[1]['log_url'])
 
     def test_multiple_sql_connections(self):
         "Test putting results in different databases"
diff --git a/tests/unit/test_executor.py b/tests/unit/test_executor.py
index 9c45645..ac7ae34 100755
--- a/tests/unit/test_executor.py
+++ b/tests/unit/test_executor.py
@@ -401,3 +401,12 @@
         node['ssh_port'] = 22022
         keys = self.test_job.getHostList({'nodes': [node]})[0]['host_keys']
         self.assertEqual(keys[0], '[localhost]:22022 fake-host-key')
+
+
+class TestExecutorHostname(ZuulTestCase):
+    config_file = 'zuul-executor-hostname.conf'
+    tenant_config_file = 'config/single-tenant/main.yaml'
+
+    def test_executor_hostname(self):
+        self.assertEqual('test-executor-hostname.openstack.org',
+                         self.executor_server.hostname)
diff --git a/tests/unit/test_merger_repo.py b/tests/unit/test_merger_repo.py
index 8aafabf..ec30a2b 100644
--- a/tests/unit/test_merger_repo.py
+++ b/tests/unit/test_merger_repo.py
@@ -19,9 +19,10 @@
 import os
 
 import git
+import testtools
 
 from zuul.merger.merger import Repo
-from tests.base import ZuulTestCase
+from tests.base import ZuulTestCase, FIXTURE_DIR
 
 
 class TestMergerRepo(ZuulTestCase):
@@ -49,7 +50,7 @@
             msg='.git file in submodule should be a file')
 
         work_repo = Repo(parent_path, self.workspace_root,
-                         'none@example.org', 'User Name')
+                         'none@example.org', 'User Name', '0', '0')
         self.assertTrue(
             os.path.isdir(os.path.join(self.workspace_root, 'subdir')),
             msg='Cloned repository has a submodule placeholder directory')
@@ -60,7 +61,7 @@
         sub_repo = Repo(
             os.path.join(self.upstream_root, 'org/project2'),
             os.path.join(self.workspace_root, 'subdir'),
-            'none@example.org', 'User Name')
+            'none@example.org', 'User Name', '0', '0')
         self.assertTrue(os.path.exists(
             os.path.join(self.workspace_root, 'subdir', '.git')),
             msg='Cloned over the submodule placeholder')
@@ -74,3 +75,28 @@
             os.path.join(self.upstream_root, 'org/project2'),
             sub_repo.createRepoObject().remotes[0].url,
             message="Sub repository points to upstream project2")
+
+    def test_clone_timeout(self):
+        parent_path = os.path.join(self.upstream_root, 'org/project1')
+        self.patch(git.Git, 'GIT_PYTHON_GIT_EXECUTABLE',
+                   os.path.join(FIXTURE_DIR, 'fake_git.sh'))
+        work_repo = Repo(parent_path, self.workspace_root,
+                         'none@example.org', 'User Name', '0', '0',
+                         git_timeout=0.001)
+        # TODO: have the merger and repo classes catch fewer
+        # exceptions, including this one on initialization.  For the
+        # test, we try cloning again.
+        with testtools.ExpectedException(git.exc.GitCommandError,
+                                         '.*exit code\(-9\)'):
+            work_repo._ensure_cloned()
+
+    def test_fetch_timeout(self):
+        parent_path = os.path.join(self.upstream_root, 'org/project1')
+        work_repo = Repo(parent_path, self.workspace_root,
+                         'none@example.org', 'User Name', '0', '0')
+        work_repo.git_timeout = 0.001
+        self.patch(git.Git, 'GIT_PYTHON_GIT_EXECUTABLE',
+                   os.path.join(FIXTURE_DIR, 'fake_git.sh'))
+        with testtools.ExpectedException(git.exc.GitCommandError,
+                                         '.*exit code\(-9\)'):
+            work_repo.update()
diff --git a/tests/unit/test_model.py b/tests/unit/test_model.py
index c457ff0..628a45c 100644
--- a/tests/unit/test_model.py
+++ b/tests/unit/test_model.py
@@ -246,7 +246,11 @@
         })
         layout.addJob(python27essex)
 
-        project_config = configloader.ProjectParser.fromYaml(tenant, layout, [{
+        project_template_parser = configloader.ProjectTemplateParser(
+            tenant, layout)
+        project_parser = configloader.ProjectParser(
+            tenant, layout, project_template_parser)
+        project_config = project_parser.fromYaml([{
             '_source_context': self.context,
             '_start_mark': self.start_mark,
             'name': 'project',
@@ -262,7 +266,7 @@
         # Test master
         change.branch = 'master'
         item = queue.enqueueChange(change)
-        item.current_build_set.layout = layout
+        item.layout = layout
 
         self.assertTrue(base.changeMatches(change))
         self.assertTrue(python27.changeMatches(change))
@@ -291,7 +295,7 @@
         # Test diablo
         change.branch = 'stable/diablo'
         item = queue.enqueueChange(change)
-        item.current_build_set.layout = layout
+        item.layout = layout
 
         self.assertTrue(base.changeMatches(change))
         self.assertTrue(python27.changeMatches(change))
@@ -321,7 +325,7 @@
         # Test essex
         change.branch = 'stable/essex'
         item = queue.enqueueChange(change)
-        item.current_build_set.layout = layout
+        item.layout = layout
 
         self.assertTrue(base.changeMatches(change))
         self.assertTrue(python27.changeMatches(change))
@@ -505,6 +509,7 @@
     def test_job_inheritance_job_tree(self):
         tenant = model.Tenant('tenant')
         layout = model.Layout(tenant)
+
         tpc = model.TenantProjectConfig(self.project)
         tenant.addUntrustedProject(tpc)
 
@@ -539,7 +544,11 @@
         })
         layout.addJob(python27diablo)
 
-        project_config = configloader.ProjectParser.fromYaml(tenant, layout, [{
+        project_template_parser = configloader.ProjectTemplateParser(
+            tenant, layout)
+        project_parser = configloader.ProjectParser(
+            tenant, layout, project_template_parser)
+        project_config = project_parser.fromYaml([{
             '_source_context': self.context,
             '_start_mark': self.start_mark,
             'name': 'project',
@@ -554,7 +563,7 @@
         change = model.Change(self.project)
         change.branch = 'master'
         item = queue.enqueueChange(change)
-        item.current_build_set.layout = layout
+        item.layout = layout
 
         self.assertTrue(base.changeMatches(change))
         self.assertTrue(python27.changeMatches(change))
@@ -568,7 +577,7 @@
 
         change.branch = 'stable/diablo'
         item = queue.enqueueChange(change)
-        item.current_build_set.layout = layout
+        item.layout = layout
 
         self.assertTrue(base.changeMatches(change))
         self.assertTrue(python27.changeMatches(change))
@@ -609,7 +618,11 @@
         })
         layout.addJob(python27)
 
-        project_config = configloader.ProjectParser.fromYaml(tenant, layout, [{
+        project_template_parser = configloader.ProjectTemplateParser(
+            tenant, layout)
+        project_parser = configloader.ProjectParser(
+            tenant, layout, project_template_parser)
+        project_config = project_parser.fromYaml([{
             '_source_context': self.context,
             '_start_mark': self.start_mark,
             'name': 'project',
@@ -625,7 +638,7 @@
         change.branch = 'master'
         change.files = ['/COMMIT_MSG', 'ignored-file']
         item = queue.enqueueChange(change)
-        item.current_build_set.layout = layout
+        item.layout = layout
 
         self.assertTrue(base.changeMatches(change))
         self.assertFalse(python27.changeMatches(change))
@@ -682,8 +695,12 @@
         context2 = model.SourceContext(project2, 'master',
                                        'test', True)
 
-        project2_config = configloader.ProjectParser.fromYaml(
-            self.tenant, self.layout, [{
+        project_template_parser = configloader.ProjectTemplateParser(
+            self.tenant, self.layout)
+        project_parser = configloader.ProjectParser(
+            self.tenant, self.layout, project_template_parser)
+        project2_config = project_parser.fromYaml(
+            [{
                 '_source_context': context2,
                 '_start_mark': self.start_mark,
                 'name': 'project2',
@@ -700,7 +717,7 @@
         # Test master
         change.branch = 'master'
         item = self.queue.enqueueChange(change)
-        item.current_build_set.layout = self.layout
+        item.layout = self.layout
         with testtools.ExpectedException(
                 Exception,
                 "Project project2 is not allowed to run job job"):
@@ -718,8 +735,12 @@
 
         self.layout.addJob(job)
 
-        project_config = configloader.ProjectParser.fromYaml(
-            self.tenant, self.layout, [{
+        project_template_parser = configloader.ProjectTemplateParser(
+            self.tenant, self.layout)
+        project_parser = configloader.ProjectParser(
+            self.tenant, self.layout, project_template_parser)
+        project_config = project_parser.fromYaml(
+            [{
                 '_source_context': self.context,
                 '_start_mark': self.start_mark,
                 'name': 'project',
@@ -736,7 +757,7 @@
         # Test master
         change.branch = 'master'
         item = self.queue.enqueueChange(change)
-        item.current_build_set.layout = self.layout
+        item.layout = self.layout
         with testtools.ExpectedException(
                 Exception,
                 "Pre-review pipeline gate does not allow post-review job"):
diff --git a/tests/unit/test_nodepool.py b/tests/unit/test_nodepool.py
index ba7523c..d3f9ddb 100644
--- a/tests/unit/test_nodepool.py
+++ b/tests/unit/test_nodepool.py
@@ -29,6 +29,7 @@
     def setUp(self):
         super(TestNodepool, self).setUp()
 
+        self.statsd = None
         self.zk_chroot_fixture = self.useFixture(
             ChrootedKazooFixture(self.id()))
         self.zk_config = '%s:%s%s' % (
@@ -76,7 +77,7 @@
         self.assertEqual(request.state, 'fulfilled')
 
         # Accept the nodes
-        self.nodepool.acceptNodes(request)
+        self.nodepool.acceptNodes(request, request.id)
         nodeset = request.nodeset
 
         for node in nodeset.getNodes():
@@ -125,3 +126,47 @@
 
         self.waitForRequests()
         self.assertEqual(len(self.provisioned_requests), 0)
+
+    def test_accept_nodes_resubmitted(self):
+        # Test that a resubmitted request would not lock nodes
+
+        nodeset = model.NodeSet()
+        nodeset.addNode(model.Node('controller', 'ubuntu-xenial'))
+        nodeset.addNode(model.Node('compute', 'ubuntu-xenial'))
+        job = model.Job('testjob')
+        job.nodeset = nodeset
+        request = self.nodepool.requestNodes(None, job)
+        self.waitForRequests()
+        self.assertEqual(len(self.provisioned_requests), 1)
+        self.assertEqual(request.state, 'fulfilled')
+
+        # Accept the nodes, passing a different ID
+        self.nodepool.acceptNodes(request, "invalid")
+        nodeset = request.nodeset
+
+        for node in nodeset.getNodes():
+            self.assertIsNone(node.lock)
+            self.assertEqual(node.state, 'ready')
+
+    def test_accept_nodes_lost_request(self):
+        # Test that a lost request would not lock nodes
+
+        nodeset = model.NodeSet()
+        nodeset.addNode(model.Node('controller', 'ubuntu-xenial'))
+        nodeset.addNode(model.Node('compute', 'ubuntu-xenial'))
+        job = model.Job('testjob')
+        job.nodeset = nodeset
+        request = self.nodepool.requestNodes(None, job)
+        self.waitForRequests()
+        self.assertEqual(len(self.provisioned_requests), 1)
+        self.assertEqual(request.state, 'fulfilled')
+
+        self.zk.deleteNodeRequest(request)
+
+        # Accept the nodes
+        self.nodepool.acceptNodes(request, request.id)
+        nodeset = request.nodeset
+
+        for node in nodeset.getNodes():
+            self.assertIsNone(node.lock)
+            self.assertEqual(node.state, 'ready')
diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py
index 2dcd9bf..3608ef0 100755
--- a/tests/unit/test_scheduler.py
+++ b/tests/unit/test_scheduler.py
@@ -89,25 +89,44 @@
         self.assertEqual(self.getJobFromHistory('project-test2').node,
                          'label1')
 
+        for stat in self.statsd.stats:
+            k, v = stat.decode('utf-8').split(':')
+            self.log.debug('stat %s:%s', k, v)
         # TODOv3(jeblair): we may want to report stats by tenant (also?).
         # Per-driver
         self.assertReportedStat('zuul.event.gerrit.comment-added', value='1|c')
         # Per-driver per-connection
         self.assertReportedStat('zuul.event.gerrit.gerrit.comment-added',
                                 value='1|c')
-        self.assertReportedStat('zuul.pipeline.gate.current_changes',
-                                value='1|g')
-        self.assertReportedStat('zuul.pipeline.gate.job.project-merge.SUCCESS',
-                                kind='ms')
-        self.assertReportedStat('zuul.pipeline.gate.job.project-merge.SUCCESS',
-                                value='1|c')
-        self.assertReportedStat('zuul.pipeline.gate.resident_time', kind='ms')
-        self.assertReportedStat('zuul.pipeline.gate.total_changes',
-                                value='1|c')
         self.assertReportedStat(
-            'zuul.pipeline.gate.org.project.resident_time', kind='ms')
+            'zuul.tenant.tenant-one.pipeline.gate.current_changes',
+            value='1|g')
         self.assertReportedStat(
-            'zuul.pipeline.gate.org.project.total_changes', value='1|c')
+            'zuul.tenant.tenant-one.pipeline.gate.project.review_example_com.'
+            'org_project.master.job.project-merge.SUCCESS', kind='ms')
+        self.assertReportedStat(
+            'zuul.tenant.tenant-one.pipeline.gate.project.review_example_com.'
+            'org_project.master.job.project-merge.SUCCESS', value='1|c')
+        self.assertReportedStat(
+            'zuul.tenant.tenant-one.pipeline.gate.resident_time', kind='ms')
+        self.assertReportedStat(
+            'zuul.tenant.tenant-one.pipeline.gate.total_changes', value='1|c')
+        self.assertReportedStat(
+            'zuul.tenant.tenant-one.pipeline.gate.project.review_example_com.'
+            'org_project.master.resident_time', kind='ms')
+        self.assertReportedStat(
+            'zuul.tenant.tenant-one.pipeline.gate.project.review_example_com.'
+            'org_project.master.total_changes', value='1|c')
+        exec_key = 'zuul.executor.%s' % self.executor_server.hostname
+        self.assertReportedStat(exec_key + '.builds', value='1|c')
+        self.assertReportedStat('zuul.nodepool.requested', value='1|c')
+        self.assertReportedStat('zuul.nodepool.requested.label.label1',
+                                value='1|c')
+        self.assertReportedStat('zuul.nodepool.fulfilled.label.label1',
+                                value='1|c')
+        self.assertReportedStat('zuul.nodepool.requested.size.1', value='1|c')
+        self.assertReportedStat('zuul.nodepool.fulfilled.size.1', value='1|c')
+        self.assertReportedStat('zuul.nodepool.current_requests', value='1|g')
 
         for build in self.history:
             self.assertTrue(build.parameters['zuul']['voting'])
@@ -1494,6 +1513,31 @@
                 held_nodes += 1
         self.assertEqual(held_nodes, 1)
 
+    @simple_layout('layouts/autohold.yaml')
+    def test_autohold_list(self):
+        client = zuul.rpcclient.RPCClient('127.0.0.1',
+                                          self.gearman_server.port)
+        self.addCleanup(client.shutdown)
+
+        r = client.autohold('tenant-one', 'org/project', 'project-test2',
+                            "reason text", 1)
+        self.assertTrue(r)
+
+        autohold_requests = client.autohold_list()
+        self.assertNotEqual({}, autohold_requests)
+        self.assertEqual(1, len(autohold_requests.keys()))
+
+        # The single dict key should be a CSV string value
+        key = list(autohold_requests.keys())[0]
+        tenant, project, job = key.split(',')
+
+        self.assertEqual('tenant-one', tenant)
+        self.assertIn('org/project', project)
+        self.assertEqual('project-test2', job)
+
+        # Note: the value is converted from set to list by json.
+        self.assertEqual([1, "reason text"], autohold_requests[key])
+
     @simple_layout('layouts/three-projects.yaml')
     def test_dependent_behind_dequeue(self):
         # This particular test does a large amount of merges and needs a little
@@ -2143,8 +2187,7 @@
 
     def test_statsd(self):
         "Test each of the statsd methods used in the scheduler"
-        import extras
-        statsd = extras.try_import('statsd.statsd')
+        statsd = self.sched.statsd
         statsd.incr('test-incr')
         statsd.timing('test-timing', 3)
         statsd.gauge('test-gauge', 12)
@@ -2269,6 +2312,58 @@
         self.assertEqual(set(['project-test-nomatch-starts-empty',
                               'project-test-nomatch-starts-full']), run_jobs)
 
+    @simple_layout('layouts/job-vars.yaml')
+    def test_inherited_job_variables(self):
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name='parentjob', result='SUCCESS'),
+            dict(name='child1', result='SUCCESS'),
+            dict(name='child2', result='SUCCESS'),
+            dict(name='child3', result='SUCCESS'),
+        ], ordered=False)
+        j = self.getJobFromHistory('parentjob')
+        rp = set([p['name'] for p in j.parameters['projects']])
+        self.assertEqual(j.parameters['vars']['override'], 0)
+        self.assertEqual(j.parameters['vars']['child1override'], 0)
+        self.assertEqual(j.parameters['vars']['parent'], 0)
+        self.assertFalse('child1' in j.parameters['vars'])
+        self.assertFalse('child2' in j.parameters['vars'])
+        self.assertFalse('child3' in j.parameters['vars'])
+        self.assertEqual(rp, set(['org/project', 'org/project0',
+                                  'org/project0']))
+        j = self.getJobFromHistory('child1')
+        rp = set([p['name'] for p in j.parameters['projects']])
+        self.assertEqual(j.parameters['vars']['override'], 1)
+        self.assertEqual(j.parameters['vars']['child1override'], 1)
+        self.assertEqual(j.parameters['vars']['parent'], 0)
+        self.assertEqual(j.parameters['vars']['child1'], 1)
+        self.assertFalse('child2' in j.parameters['vars'])
+        self.assertFalse('child3' in j.parameters['vars'])
+        self.assertEqual(rp, set(['org/project', 'org/project0',
+                                  'org/project1']))
+        j = self.getJobFromHistory('child2')
+        rp = set([p['name'] for p in j.parameters['projects']])
+        self.assertEqual(j.parameters['vars']['override'], 2)
+        self.assertEqual(j.parameters['vars']['child1override'], 0)
+        self.assertEqual(j.parameters['vars']['parent'], 0)
+        self.assertFalse('child1' in j.parameters['vars'])
+        self.assertEqual(j.parameters['vars']['child2'], 2)
+        self.assertFalse('child3' in j.parameters['vars'])
+        self.assertEqual(rp, set(['org/project', 'org/project0',
+                                  'org/project2']))
+        j = self.getJobFromHistory('child3')
+        rp = set([p['name'] for p in j.parameters['projects']])
+        self.assertEqual(j.parameters['vars']['override'], 3)
+        self.assertEqual(j.parameters['vars']['child1override'], 0)
+        self.assertEqual(j.parameters['vars']['parent'], 0)
+        self.assertFalse('child1' in j.parameters['vars'])
+        self.assertFalse('child2' in j.parameters['vars'])
+        self.assertEqual(j.parameters['vars']['child3'], 3)
+        self.assertEqual(rp, set(['org/project', 'org/project0',
+                                  'org/project3']))
+
     def test_queue_names(self):
         "Test shared change queue names"
         tenant = self.sched.abide.tenants.get('tenant-one')
@@ -2411,6 +2506,28 @@
         self.assertIn('project-merge', status_jobs[1]['dependencies'])
         self.assertIn('project-merge', status_jobs[2]['dependencies'])
 
+    def test_reconfigure_merge(self):
+        """Test that two reconfigure events are merged"""
+
+        tenant = self.sched.abide.tenants['tenant-one']
+        (trusted, project) = tenant.getProject('org/project')
+
+        self.sched.run_handler_lock.acquire()
+        self.assertEqual(self.sched.management_event_queue.qsize(), 0)
+
+        self.sched.reconfigureTenant(tenant, project)
+        self.assertEqual(self.sched.management_event_queue.qsize(), 1)
+
+        self.sched.reconfigureTenant(tenant, project)
+        # The second event should have been combined with the first
+        # so we should still only have one entry.
+        self.assertEqual(self.sched.management_event_queue.qsize(), 1)
+
+        self.sched.run_handler_lock.release()
+        self.waitUntilSettled()
+
+        self.assertEqual(self.sched.management_event_queue.qsize(), 0)
+
     def test_live_reconfiguration(self):
         "Test that live reconfiguration works"
         self.executor_server.hold_jobs_in_build = True
@@ -4544,6 +4661,50 @@
         self.assertEqual(B.data['status'], 'MERGED')
         self.assertEqual(B.reported, 2)
 
+    def test_job_aborted(self):
+        "Test that if a execute server aborts a job, it is run again"
+        self.executor_server.hold_jobs_in_build = True
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        self.executor_server.release('.*-merge')
+        self.waitUntilSettled()
+
+        self.assertEqual(len(self.builds), 2)
+
+        # first abort
+        self.builds[0].aborted = True
+        self.executor_server.release('.*-test*')
+        self.waitUntilSettled()
+        self.assertEqual(len(self.builds), 1)
+
+        # second abort
+        self.builds[0].aborted = True
+        self.executor_server.release('.*-test*')
+        self.waitUntilSettled()
+        self.assertEqual(len(self.builds), 1)
+
+        # third abort
+        self.builds[0].aborted = True
+        self.executor_server.release('.*-test*')
+        self.waitUntilSettled()
+        self.assertEqual(len(self.builds), 1)
+
+        # fourth abort
+        self.builds[0].aborted = True
+        self.executor_server.release('.*-test*')
+        self.waitUntilSettled()
+        self.assertEqual(len(self.builds), 1)
+
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+
+        self.assertEqual(len(self.history), 7)
+        self.assertEqual(self.countJobResults(self.history, 'ABORTED'), 4)
+        self.assertEqual(self.countJobResults(self.history, 'SUCCESS'), 3)
+
     def test_rerun_on_abort(self):
         "Test that if a execute server fails to run a job, it is run again"
 
@@ -4613,6 +4774,78 @@
         self.assertIn('project-test1 : SKIPPED', A.messages[1])
         self.assertIn('project-test2 : SKIPPED', A.messages[1])
 
+    def test_nodepool_priority(self):
+        "Test that nodes are requested at the correct priority"
+
+        self.fake_nodepool.paused = True
+
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getRefUpdatedEvent())
+
+        B = self.fake_gerrit.addFakeChange('org/project1', 'master', 'B')
+        self.fake_gerrit.addEvent(B.getPatchsetCreatedEvent(1))
+
+        C = self.fake_gerrit.addFakeChange('org/project', 'master', 'C')
+        C.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(C.addApproval('Approved', 1))
+
+        self.waitUntilSettled()
+
+        reqs = self.fake_nodepool.getNodeRequests()
+
+        # The requests come back sorted by oid. Since we have three requests
+        # for the three changes each with a different priority.
+        # Also they get a serial number based on order they were received
+        # so the number on the endof the oid should map to order submitted.
+
+        # * gate first - high priority - change C
+        self.assertEqual(reqs[0]['_oid'], '100-0000000002')
+        self.assertEqual(reqs[0]['node_types'], ['label1'])
+        # * check second - normal priority - change B
+        self.assertEqual(reqs[1]['_oid'], '200-0000000001')
+        self.assertEqual(reqs[1]['node_types'], ['label1'])
+        # * post third - low priority - change A
+        # additionally, the post job defined uses an ubuntu-xenial node,
+        # so we include that check just as an extra verification
+        self.assertEqual(reqs[2]['_oid'], '300-0000000000')
+        self.assertEqual(reqs[2]['node_types'], ['ubuntu-xenial'])
+
+        self.fake_nodepool.paused = False
+        self.waitUntilSettled()
+
+    @simple_layout('layouts/multiple-templates.yaml')
+    def test_multiple_project_templates(self):
+        # Test that applying multiple project templates to a project
+        # doesn't alter them when used for a second project.
+        A = self.fake_gerrit.addFakeChange('org/project2', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        build = self.getJobFromHistory('py27')
+        self.assertEqual(build.parameters['zuul']['jobtags'], [])
+
+    def test_pending_merge_in_reconfig(self):
+        # Test that if we are waiting for an outstanding merge on
+        # reconfiguration that we continue to do so.
+        self.gearman_server.hold_merge_jobs_in_queue = True
+        A = self.fake_gerrit.addFakeChange('org/project1', 'master', 'A')
+        A.setMerged()
+        self.fake_gerrit.addEvent(A.getRefUpdatedEvent())
+        self.waitUntilSettled()
+        # Reconfigure while we still have an outstanding merge job
+        self.sched.reconfigureTenant(self.sched.abide.tenants['tenant-one'],
+                                     None)
+        self.waitUntilSettled()
+        # Verify the merge job is still running and that the item is
+        # in the pipeline
+        self.assertEqual(len(self.sched.merger.jobs), 1)
+        tenant = self.sched.abide.tenants.get('tenant-one')
+        pipeline = tenant.layout.pipelines['post']
+        self.assertEqual(len(pipeline.getAllItems()), 1)
+        self.gearman_server.hold_merge_jobs_in_queue = False
+        self.gearman_server.release()
+        self.waitUntilSettled()
+
 
 class TestExecutor(ZuulTestCase):
     tenant_config_file = 'config/single-tenant/main.yaml'
@@ -4830,6 +5063,42 @@
         self.assertEqual(self.getJobFromHistory('project-test6').result,
                          'SUCCESS')
 
+    def test_unimplied_branch_matchers(self):
+        # This tests that there are no implied branch matchers added
+        # by project templates.
+        self.create_branch('org/layered-project', 'stable')
+
+        A = self.fake_gerrit.addFakeChange(
+            'org/layered-project', 'stable', 'A')
+
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        self.assertEqual(self.getJobFromHistory('project-test1').result,
+                         'SUCCESS')
+        print(self.getJobFromHistory('project-test1').
+              parameters['zuul']['_inheritance_path'])
+
+    def test_implied_branch_matchers(self):
+        # This tests that there is an implied branch matcher when a
+        # template is used on an in-repo project pipeline definition.
+        self.create_branch('untrusted-config', 'stable')
+        self.fake_gerrit.addEvent(
+            self.fake_gerrit.getFakeBranchCreatedEvent(
+                'untrusted-config', 'stable'))
+        self.waitUntilSettled()
+
+        A = self.fake_gerrit.addFakeChange(
+            'untrusted-config', 'stable', 'A')
+
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        self.assertEqual(self.getJobFromHistory('project-test1').result,
+                         'SUCCESS')
+        print(self.getJobFromHistory('project-test1').
+              parameters['zuul']['_inheritance_path'])
+
 
 class TestSchedulerSuccessURL(ZuulTestCase):
     tenant_config_file = 'config/success-url/main.yaml'
@@ -5503,7 +5772,7 @@
                 queue = queue_candidate
                 break
         queue_item = queue.queue[0]
-        item_dynamic_layout = queue_item.current_build_set.layout
+        item_dynamic_layout = queue_item.layout
         dynamic_test_semaphore = \
             item_dynamic_layout.semaphores.get('test-semaphore')
         self.assertEqual(dynamic_test_semaphore.max, 1)
@@ -5556,3 +5825,30 @@
         self.assertEqual(A.reported, 2)
         self.assertEqual(B.reported, 2)
         self.assertEqual(C.reported, 2)
+
+
+class TestSchedulerBranchMatcher(ZuulTestCase):
+
+    @simple_layout('layouts/matcher-test.yaml')
+    def test_job_branch_ignored(self):
+        '''
+        Test that branch matching logic works.
+
+        The 'ignore-branch' job has a branch matcher that is supposed to
+        match every branch except for the 'featureA' branch, so it should
+        not be run on a change to that branch.
+        '''
+        self.create_branch('org/project', 'featureA')
+        A = self.fake_gerrit.addFakeChange('org/project', 'featureA', 'A')
+        A.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+        self.printHistory()
+        self.assertEqual(self.getJobFromHistory('project-test1').result,
+                         'SUCCESS')
+        self.assertJobNotInHistory('ignore-branch')
+        self.assertEqual(A.data['status'], 'MERGED')
+        self.assertEqual(A.reported, 2,
+                         "A should report start and success")
+        self.assertIn('gate', A.messages[1],
+                      "A should transit gate")
diff --git a/tests/unit/test_stack_dump.py b/tests/unit/test_stack_dump.py
index 824e04c..13e83c6 100644
--- a/tests/unit/test_stack_dump.py
+++ b/tests/unit/test_stack_dump.py
@@ -31,4 +31,5 @@
 
         zuul.cmd.stack_dump_handler(signal.SIGUSR2, None)
         self.assertIn("Thread", self.log_fixture.output)
+        self.assertIn("MainThread", self.log_fixture.output)
         self.assertIn("test_stack_dump_logs", self.log_fixture.output)
diff --git a/tests/unit/test_v3.py b/tests/unit/test_v3.py
index 1c633ba..b30d710 100755
--- a/tests/unit/test_v3.py
+++ b/tests/unit/test_v3.py
@@ -14,9 +14,14 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import io
 import json
+import logging
 import os
 import textwrap
+import gc
+import time
+from unittest import skip
 
 import testtools
 
@@ -151,6 +156,29 @@
         self.assertIn('Unable to inherit from final job', A.messages[0])
 
 
+class TestBranchVariants(ZuulTestCase):
+    tenant_config_file = 'config/branch-variants/main.yaml'
+
+    def test_branch_variants(self):
+        # Test branch variants of jobs with inheritance
+        self.executor_server.hold_jobs_in_build = True
+        # This creates a new branch with a copy of the config in master
+        self.create_branch('puppet-integration', 'stable')
+        self.fake_gerrit.addEvent(
+            self.fake_gerrit.getFakeBranchCreatedEvent(
+                'puppet-integration', 'stable'))
+        self.waitUntilSettled()
+
+        A = self.fake_gerrit.addFakeChange('puppet-integration', 'stable', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        self.assertEqual(len(self.builds[0].parameters['pre_playbooks']), 3)
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+
+
 class TestInRepoConfig(ZuulTestCase):
     # A temporary class to hold new tests while others are disabled
 
@@ -170,6 +198,39 @@
         self.assertIn('tenant-one-gate', A.messages[1],
                       "A should transit tenant-one gate")
 
+    @skip("This test is useful, but not reliable")
+    def test_full_and_dynamic_reconfig(self):
+        self.executor_server.hold_jobs_in_build = True
+        in_repo_conf = textwrap.dedent(
+            """
+            - job:
+                name: project-test1
+
+            - project:
+                name: org/project
+                tenant-one-gate:
+                  jobs:
+                    - project-test1
+            """)
+
+        file_dict = {'.zuul.yaml': in_repo_conf}
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A',
+                                           files=file_dict)
+        A.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+        self.sched.reconfigure(self.config)
+        self.waitUntilSettled()
+
+        gc.collect()
+        pipelines = [obj for obj in gc.get_objects()
+                     if isinstance(obj, zuul.model.Pipeline)]
+        self.assertEqual(len(pipelines), 4)
+
+        self.executor_server.hold_jobs_in_build = False
+        self.executor_server.release()
+        self.waitUntilSettled()
+
     def test_dynamic_config(self):
         in_repo_conf = textwrap.dedent(
             """
@@ -696,6 +757,47 @@
         self.assertIn('the only project definition permitted', A.messages[0],
                       "A should have a syntax error reported")
 
+    def test_untrusted_depends_on_trusted(self):
+        with open(os.path.join(FIXTURE_DIR,
+                               'config/in-repo/git/',
+                               'common-config/zuul.yaml')) as f:
+            common_config = f.read()
+
+        common_config += textwrap.dedent(
+            """
+            - job:
+                name: project-test9
+            """)
+
+        file_dict = {'zuul.yaml': common_config}
+        A = self.fake_gerrit.addFakeChange('common-config', 'master', 'A',
+                                           files=file_dict)
+        in_repo_conf = textwrap.dedent(
+            """
+            - job:
+                name: project-test1
+            - project:
+                name: org/project
+                check:
+                  jobs:
+                    - project-test9
+            """)
+
+        file_dict = {'zuul.yaml': in_repo_conf}
+        B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B',
+                                           files=file_dict)
+        B.data['commitMessage'] = '%s\n\nDepends-On: %s\n' % (
+            B.subject, A.data['id'])
+        self.fake_gerrit.addEvent(B.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+
+        self.assertEqual(B.data['status'], 'NEW')
+        self.assertEqual(B.reported, 1,
+                         "B should report failure")
+        self.assertIn('depends on a change to a config project',
+                      B.messages[0],
+                      "A should have a syntax error reported")
+
     def test_duplicate_node_error(self):
         in_repo_conf = textwrap.dedent(
             """
@@ -816,6 +918,150 @@
                       A.messages[0],
                       "A should have a syntax error reported")
 
+    def test_job_list_in_project_template_not_dict_error(self):
+        in_repo_conf = textwrap.dedent(
+            """
+            - job:
+                name: project-test1
+            - project-template:
+                name: some-jobs
+                check:
+                  jobs:
+                    - project-test1:
+                        - required-projects:
+                            org/project2
+            """)
+
+        file_dict = {'.zuul.yaml': in_repo_conf}
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A',
+                                           files=file_dict)
+        A.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+
+        self.assertEqual(A.data['status'], 'NEW')
+        self.assertEqual(A.reported, 1,
+                         "A should report failure")
+        self.assertIn('expected str for dictionary value',
+                      A.messages[0], "A should have a syntax error reported")
+
+    def test_job_list_in_project_not_dict_error(self):
+        in_repo_conf = textwrap.dedent(
+            """
+            - job:
+                name: project-test1
+            - project:
+                name: org/project1
+                check:
+                  jobs:
+                    - project-test1:
+                        - required-projects:
+                            org/project2
+            """)
+
+        file_dict = {'.zuul.yaml': in_repo_conf}
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A',
+                                           files=file_dict)
+        A.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+
+        self.assertEqual(A.data['status'], 'NEW')
+        self.assertEqual(A.reported, 1,
+                         "A should report failure")
+        self.assertIn('expected str for dictionary value',
+                      A.messages[0], "A should have a syntax error reported")
+
+    def test_project_template(self):
+        # Tests that a project template is not modified when used, and
+        # can therefore be used in subsequent reconfigurations.
+        in_repo_conf = textwrap.dedent(
+            """
+            - job:
+                name: project-test1
+            - project-template:
+                name: some-jobs
+                tenant-one-gate:
+                  jobs:
+                    - project-test1:
+                        required-projects:
+                          - org/project1
+            - project:
+                name: org/project
+                templates:
+                  - some-jobs
+            """)
+
+        file_dict = {'.zuul.yaml': in_repo_conf}
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A',
+                                           files=file_dict)
+        A.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
+        self.waitUntilSettled()
+        self.assertEqual(A.data['status'], 'MERGED')
+        self.fake_gerrit.addEvent(A.getChangeMergedEvent())
+        self.waitUntilSettled()
+        in_repo_conf = textwrap.dedent(
+            """
+            - project:
+                name: org/project1
+                templates:
+                  - some-jobs
+            """)
+        file_dict = {'.zuul.yaml': in_repo_conf}
+        B = self.fake_gerrit.addFakeChange('org/project1', 'master', 'B',
+                                           files=file_dict)
+        B.addApproval('Code-Review', 2)
+        self.fake_gerrit.addEvent(B.addApproval('Approved', 1))
+        self.waitUntilSettled()
+        self.assertEqual(B.data['status'], 'MERGED')
+
+    def test_job_remove_add(self):
+        # Tests that a job can be removed from one repo and added in another.
+        # First, remove the current config for project1 since it
+        # references the job we want to remove.
+        file_dict = {'.zuul.yaml': None}
+        A = self.fake_gerrit.addFakeChange('org/project1', 'master', 'A',
+                                           files=file_dict)
+        A.setMerged()
+        self.fake_gerrit.addEvent(A.getChangeMergedEvent())
+        self.waitUntilSettled()
+        # Then propose a change to delete the job from one repo...
+        file_dict = {'.zuul.yaml': None}
+        B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B',
+                                           files=file_dict)
+        self.fake_gerrit.addEvent(B.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+        # ...and a second that depends on it that adds it to another repo.
+        in_repo_conf = textwrap.dedent(
+            """
+            - job:
+                name: project-test1
+
+            - project:
+                name: org/project1
+                check:
+                  jobs:
+                    - project-test1
+            """)
+        in_repo_playbook = textwrap.dedent(
+            """
+            - hosts: all
+              tasks: []
+            """)
+        file_dict = {'.zuul.yaml': in_repo_conf,
+                     'playbooks/project-test1.yaml': in_repo_playbook}
+        C = self.fake_gerrit.addFakeChange('org/project1', 'master', 'C',
+                                           files=file_dict,
+                                           parent='refs/changes/1/1/1')
+        C.data['commitMessage'] = '%s\n\nDepends-On: %s\n' % (
+            C.subject, B.data['id'])
+        self.fake_gerrit.addEvent(C.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name='project-test1', result='SUCCESS', changes='2,1 3,1'),
+        ], ordered=False)
+
     def test_multi_repo(self):
         downstream_repo_conf = textwrap.dedent(
             """
@@ -1215,6 +1461,40 @@
                         "The file %s should exist" % post_flag_path)
 
 
+class TestPostPlaybooks(AnsibleZuulTestCase):
+    tenant_config_file = 'config/post-playbook/main.yaml'
+
+    def test_post_playbook_abort(self):
+        # Test that when we abort a job in the post playbook, that we
+        # don't send back POST_FAILURE.
+        self.executor_server.verbose = True
+        A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+
+        while not len(self.builds):
+            time.sleep(0.1)
+        build = self.builds[0]
+
+        post_start = os.path.join(self.test_root, build.uuid +
+                                  '.post_start.flag')
+        start = time.time()
+        while time.time() < start + 90:
+            if os.path.exists(post_start):
+                break
+            time.sleep(0.1)
+        # The post playbook has started, abort the job
+        self.fake_gerrit.addEvent(A.getChangeAbandonedEvent())
+        self.waitUntilSettled()
+
+        build = self.getJobFromHistory('python27')
+        self.assertEqual('ABORTED', build.result)
+
+        post_end = os.path.join(self.test_root, build.uuid +
+                                '.post_end.flag')
+        self.assertTrue(os.path.exists(post_start))
+        self.assertFalse(os.path.exists(post_end))
+
+
 class TestBrokenConfig(ZuulTestCase):
     # Test that we get an appropriate syntax error if we start with a
     # broken config.
@@ -1665,7 +1945,8 @@
             return f.read()
 
     def test_job_output(self):
-        # Verify that command standard output appears in the job output
+        # Verify that command standard output appears in the job output,
+        # and that failures in the final playbook get logged.
 
         # This currently only verifies we receive output from
         # localhost.  Notably, it does not verify we receive output
@@ -1690,3 +1971,36 @@
         self.assertIn(token,
                       self._get_file(self.history[0],
                                      'work/logs/job-output.txt'))
+
+    def test_job_output_failure_log(self):
+        logger = logging.getLogger('zuul.AnsibleJob')
+        output = io.StringIO()
+        logger.addHandler(logging.StreamHandler(output))
+
+        # Verify that a failure in the last post playbook emits the contents
+        # of the json output to the log
+        self.executor_server.keep_jobdir = True
+        A = self.fake_gerrit.addFakeChange('org/project2', 'master', 'A')
+        self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
+        self.waitUntilSettled()
+        self.assertHistory([
+            dict(name='job-output-failure',
+                 result='POST_FAILURE', changes='1,1'),
+        ], ordered=False)
+
+        token = 'Standard output test %s' % (self.history[0].jobdir.src_root)
+        j = json.loads(self._get_file(self.history[0],
+                                      'work/logs/job-output.json'))
+        self.assertEqual(token,
+                         j[0]['plays'][0]['tasks'][0]
+                         ['hosts']['localhost']['stdout'])
+
+        print(self._get_file(self.history[0],
+                             'work/logs/job-output.json'))
+        self.assertIn(token,
+                      self._get_file(self.history[0],
+                                     'work/logs/job-output.txt'))
+
+        log_output = output.getvalue()
+        self.assertIn('Final playbook failed', log_output)
+        self.assertIn('Failure test', log_output)
diff --git a/tools/run-migration.sh b/tools/run-migration.sh
index be297f4..618fc56 100755
--- a/tools/run-migration.sh
+++ b/tools/run-migration.sh
@@ -47,12 +47,12 @@
 
 BASE_DIR=$(cd $(dirname $0)/../..; pwd)
 cd $BASE_DIR/project-config
-if [[ $FINAL ]] ; then
+if [[ $FINAL = 1 ]] ; then
     git reset --hard
 fi
 python3 $BASE_DIR/zuul/zuul/cmd/migrate.py  --mapping=zuul/mapping.yaml \
     zuul/layout.yaml jenkins/jobs nodepool/nodepool.yaml . $VERBOSE
-if [[ $FINAL ]] ; then
+if [[ $FINAL = 1 ]] ; then
     find ../openstack-zuul-jobs/playbooks/legacy -maxdepth 1 -mindepth 1 \
         -type d  | xargs rm -rf
     mv zuul.d/zuul-legacy-* ../openstack-zuul-jobs/zuul.d/
diff --git a/tools/zuul-changes.py b/tools/zuul-changes.py
index 8b854c7..d258354 100755
--- a/tools/zuul-changes.py
+++ b/tools/zuul-changes.py
@@ -20,14 +20,15 @@
 
 parser = argparse.ArgumentParser()
 parser.add_argument('url', help='The URL of the running Zuul instance')
-parser.add_argument('pipeline_name', help='The name of the Zuul pipeline')
+parser.add_argument('tenant', help='The Zuul tenant')
+parser.add_argument('pipeline', help='The name of the Zuul pipeline')
 options = parser.parse_args()
 
 data = urllib2.urlopen('%s/status.json' % options.url).read()
 data = json.loads(data)
 
 for pipeline in data['pipelines']:
-    if pipeline['name'] != options.pipeline_name:
+    if pipeline['name'] != options.pipeline:
         continue
     for queue in pipeline['change_queues']:
         for head in queue['heads']:
@@ -36,9 +37,10 @@
                     continue
                 cid, cps = change['id'].split(',')
                 print(
-                    "zuul enqueue --trigger gerrit --pipeline %s "
-                    "--project %s --change %s,%s" % (
-                        options.pipeline_name,
+                    "zuul enqueue --tenant %s --trigger gerrit "
+                    "--pipeline %s --project %s --change %s,%s" % (
+                        options.tenant,
+                        options.pipeline,
                         change['project'],
                         cid, cps)
                 )
diff --git a/tox.ini b/tox.ini
index cc5ea58..7e84677 100644
--- a/tox.ini
+++ b/tox.ini
@@ -5,10 +5,7 @@
 
 [testenv]
 basepython = python3
-# Set STATSD env variables so that statsd code paths are tested.
-setenv = STATSD_HOST=127.0.0.1
-         STATSD_PORT=8125
-         VIRTUAL_ENV={envdir}
+setenv = VIRTUAL_ENV={envdir}
          OS_TEST_TIMEOUT=120
 passenv = ZUUL_TEST_ROOT OS_STDOUT_CAPTURE OS_STDERR_CAPTURE OS_LOG_CAPTURE OS_LOG_DEFAULTS
 usedevelop = True
diff --git a/zuul/ansible/callback/zuul_stream.py b/zuul/ansible/callback/zuul_stream.py
index 0a266df..8ba3b86 100644
--- a/zuul/ansible/callback/zuul_stream.py
+++ b/zuul/ansible/callback/zuul_stream.py
@@ -168,32 +168,8 @@
                 host=host.name,
                 filename=included_file._filename))
 
-    def _emit_playbook_banner(self):
-        # Get the hostvars from just one host - the vars we're looking for will
-        # be identical on all of them
-        hostvars = next(iter(self._play._variable_manager._hostvars.values()))
-        self._playbook_name = None
-
-        phase = hostvars.get('zuul_execution_phase', '')
-        playbook = hostvars.get('zuul_execution_canonical_name_and_path')
-        trusted = hostvars.get('zuul_execution_trusted')
-        trusted = 'trusted' if trusted == "True" else 'untrusted'
-        branch = hostvars.get('zuul_execution_branch')
-
-        if phase and phase != 'run':
-            phase = '{phase}-run'.format(phase=phase)
-        phase = phase.upper()
-
-        self._log("{phase} [{trusted} : {playbook}@{branch}]".format(
-            trusted=trusted, phase=phase, playbook=playbook, branch=branch))
-
     def v2_playbook_on_play_start(self, play):
         self._play = play
-
-        # We can't fill in this information until the first play
-        if self._playbook_name:
-            self._emit_playbook_banner()
-
         # Log an extra blank line to get space before each play
         self._log("")
 
diff --git a/zuul/ansible/filter/zuul_filters.py b/zuul/ansible/filter/zuul_filters.py
index 17ef2bb..fa21f6b 100644
--- a/zuul/ansible/filter/zuul_filters.py
+++ b/zuul/ansible/filter/zuul_filters.py
@@ -14,10 +14,19 @@
 
 
 def zuul_legacy_vars(zuul):
-    # omitted:
-    # ZUUL_URL
-    # ZUUL_REF
+    # intentionally omitted:
+    # BASE_LOG_PATH
+    # JOB_TAGS
+    # LOG_PATH
     # ZUUL_COMMIT
+    # ZUUL_REF
+    # ZUUL_URL
+    #
+    # newly added to all builds:
+    # ZUUL_SHORT_PROJECT_NAME
+    #
+    # existing in most builds but newly added for periodic:
+    # ZUUL_BRANCH
 
     short_name = zuul['project']['name'].split('/')[-1]
     params = dict(ZUUL_UUID=zuul['build'],
@@ -26,6 +35,8 @@
                   ZUUL_PIPELINE=zuul['pipeline'],
                   ZUUL_VOTING=zuul['voting'],
                   WORKSPACE='/home/zuul/workspace')
+    if 'timeout' in zuul and zuul['timeout'] is not None:
+        params['BUILD_TIMEOUT'] = str(int(zuul['timeout']) * 1000)
     if 'branch' in zuul:
         params['ZUUL_BRANCH'] = zuul['branch']
 
@@ -34,7 +45,7 @@
             ['%s:%s:refs/changes/%s/%s/%s' % (
                 i['project']['name'],
                 i['branch'],
-                str(i['change'])[:-2:],
+                str(i['change'])[-2:],
                 i['change'],
                 i['patchset'])
              for i in zuul['items']])
diff --git a/zuul/ansible/paths.py b/zuul/ansible/paths.py
index 04daef4..05f9fdc 100644
--- a/zuul/ansible/paths.py
+++ b/zuul/ansible/paths.py
@@ -24,7 +24,7 @@
 
 def _is_safe_path(path):
     full_path = os.path.realpath(os.path.abspath(os.path.expanduser(path)))
-    if not full_path.startswith(os.path.abspath(os.path.curdir)):
+    if not full_path.startswith(os.path.abspath(os.path.expanduser('~'))):
         return False
     return True
 
diff --git a/zuul/change_matcher.py b/zuul/change_matcher.py
index baea217..7f6673d 100644
--- a/zuul/change_matcher.py
+++ b/zuul/change_matcher.py
@@ -60,11 +60,13 @@
 class BranchMatcher(AbstractChangeMatcher):
 
     def matches(self, change):
-        return (
-            (hasattr(change, 'branch') and self.regex.match(change.branch)) or
-            (hasattr(change, 'ref') and
-             change.ref is not None and self.regex.match(change.ref))
-        )
+        if hasattr(change, 'branch'):
+            if self.regex.match(change.branch):
+                return True
+            return False
+        if self.regex.match(change.ref):
+            return True
+        return False
 
 
 class FileMatcher(AbstractChangeMatcher):
diff --git a/zuul/cmd/__init__.py b/zuul/cmd/__init__.py
index a5db9a6..86f7f12 100755
--- a/zuul/cmd/__init__.py
+++ b/zuul/cmd/__init__.py
@@ -23,8 +23,10 @@
 import signal
 import sys
 import traceback
+import threading
 
 yappi = extras.try_import('yappi')
+objgraph = extras.try_import('objgraph')
 
 from zuul.ansible import logconfig
 import zuul.lib.connections
@@ -37,23 +39,50 @@
 
 def stack_dump_handler(signum, frame):
     signal.signal(signal.SIGUSR2, signal.SIG_IGN)
-    log_str = ""
-    for thread_id, stack_frame in sys._current_frames().items():
-        log_str += "Thread: %s\n" % thread_id
-        log_str += "".join(traceback.format_stack(stack_frame))
     log = logging.getLogger("zuul.stack_dump")
-    log.debug(log_str)
-    if yappi:
-        if not yappi.is_running():
-            yappi.start()
-        else:
-            yappi.stop()
-            yappi_out = io.BytesIO()
-            yappi.get_func_stats().print_all(out=yappi_out)
-            yappi.get_thread_stats().print_all(out=yappi_out)
-            log.debug(yappi_out.getvalue())
-            yappi_out.close()
-            yappi.clear_stats()
+    log.debug("Beginning debug handler")
+    try:
+        threads = {}
+        for t in threading.enumerate():
+            threads[t.ident] = t
+        log_str = ""
+        for thread_id, stack_frame in sys._current_frames().items():
+            thread = threads.get(thread_id)
+            if thread:
+                thread_name = thread.name
+            else:
+                thread_name = thread.ident
+            log_str += "Thread: %s %s\n" % (thread_id, thread_name)
+            log_str += "".join(traceback.format_stack(stack_frame))
+        log.debug(log_str)
+    except Exception:
+        log.exception("Thread dump error:")
+    try:
+        if yappi:
+            if not yappi.is_running():
+                log.debug("Starting Yappi")
+                yappi.start()
+            else:
+                log.debug("Stopping Yappi")
+                yappi.stop()
+                yappi_out = io.StringIO()
+                yappi.get_func_stats().print_all(out=yappi_out)
+                yappi.get_thread_stats().print_all(out=yappi_out)
+                log.debug(yappi_out.getvalue())
+                yappi_out.close()
+                yappi.clear_stats()
+    except Exception:
+        log.exception("Yappi error:")
+    try:
+        if objgraph:
+            log.debug("Most common types:")
+            objgraph_out = io.StringIO()
+            objgraph.show_growth(limit=100, file=objgraph_out)
+            log.debug(objgraph_out.getvalue())
+            objgraph_out.close()
+    except Exception:
+        log.exception("Objgraph error:")
+    log.debug("End debug handler")
     signal.signal(signal.SIGUSR2, stack_dump_handler)
 
 
diff --git a/zuul/cmd/client.py b/zuul/cmd/client.py
index 177283e..c9e399a 100755
--- a/zuul/cmd/client.py
+++ b/zuul/cmd/client.py
@@ -61,6 +61,10 @@
                                   required=False, type=int, default=1)
         cmd_autohold.set_defaults(func=self.autohold)
 
+        cmd_autohold_list = subparsers.add_parser(
+            'autohold-list', help='list autohold requests')
+        cmd_autohold_list.set_defaults(func=self.autohold_list)
+
         cmd_enqueue = subparsers.add_parser('enqueue', help='enqueue a change')
         cmd_enqueue.add_argument('--tenant', help='tenant name',
                                  required=True)
@@ -162,6 +166,27 @@
                             count=self.args.count)
         return r
 
+    def autohold_list(self):
+        client = zuul.rpcclient.RPCClient(
+            self.server, self.port, self.ssl_key, self.ssl_cert, self.ssl_ca)
+        autohold_requests = client.autohold_list()
+
+        if len(autohold_requests.keys()) == 0:
+            print("No autohold requests found")
+            return True
+
+        table = prettytable.PrettyTable(
+            field_names=['Tenant', 'Project', 'Job', 'Count', 'Reason'])
+
+        for key, value in autohold_requests.items():
+            # The key comes to us as a CSV string because json doesn't like
+            # non-str keys.
+            tenant_name, project_name, job_name = key.split(',')
+            count, reason = value
+            table.add_row([tenant_name, project_name, job_name, count, reason])
+        print(table)
+        return True
+
     def enqueue(self):
         client = zuul.rpcclient.RPCClient(
             self.server, self.port, self.ssl_key, self.ssl_cert, self.ssl_ca)
diff --git a/zuul/cmd/executor.py b/zuul/cmd/executor.py
index 63c621d..70c80c5 100755
--- a/zuul/cmd/executor.py
+++ b/zuul/cmd/executor.py
@@ -22,6 +22,7 @@
 # instead it depends on lockfile-0.9.1 which uses pidfile.
 pid_file_module = extras.try_imports(['daemon.pidlockfile', 'daemon.pidfile'])
 
+import grp
 import logging
 import os
 import pwd
@@ -101,7 +102,10 @@
         if os.getuid() != 0:
             return
         pw = pwd.getpwnam(self.user)
-        os.setgroups([])
+        # get a list of supplementary groups for the target user, and make sure
+        # we set them when dropping privileges.
+        groups = [g.gr_gid for g in grp.getgrall() if self.user in g.gr_mem]
+        os.setgroups(groups)
         os.setgid(pw.pw_gid)
         os.setuid(pw.pw_uid)
         os.umask(0o022)
diff --git a/zuul/cmd/migrate.py b/zuul/cmd/migrate.py
index 1f0a602..efb2796 100644
--- a/zuul/cmd/migrate.py
+++ b/zuul/cmd/migrate.py
@@ -42,6 +42,9 @@
 import jenkins_jobs.parser
 import yaml
 
+JOB_MATCHERS = {}  # type: Dict[str, Dict[str, Dict]]
+TEMPLATES_TO_EXPAND = {}  # type: Dict[str, List]
+JOBS_FOR_EXPAND = collections.defaultdict(dict)  # type: ignore
 JOBS_BY_ORIG_TEMPLATE = {}  # type: ignore
 SUFFIXES = []  # type: ignore
 ENVIRONMENT = '{{ zuul | zuul_legacy_vars }}'
@@ -186,6 +189,37 @@
     return
 
 
+def normalize_project_expansions():
+    remove_from_job_matchers = []
+    template = None
+    # First find the matchers that are the same for all jobs
+    for job_name, project in copy.deepcopy(JOBS_FOR_EXPAND).items():
+        JOB_MATCHERS[job_name] = None
+        for project_name, expansion in project.items():
+            template = expansion['template']
+            if not JOB_MATCHERS[job_name]:
+                JOB_MATCHERS[job_name] = copy.deepcopy(expansion['info'])
+            else:
+                if JOB_MATCHERS[job_name] != expansion['info']:
+                    # We have different expansions for this job, it can't be
+                    # done at the job level
+                    remove_from_job_matchers.append(job_name)
+
+    for job_name in remove_from_job_matchers:
+        JOB_MATCHERS.pop(job_name, None)
+
+    # Second, find out which projects need to expand a given template
+    for job_name, project in copy.deepcopy(JOBS_FOR_EXPAND).items():
+        # There is a job-level expansion for this one
+        if job_name in JOB_MATCHERS.keys():
+            continue
+        for project_name, expansion in project.items():
+            TEMPLATES_TO_EXPAND[project_name] = []
+            if expansion['info']:
+                # There is an expansion for this project
+                TEMPLATES_TO_EXPAND[project_name].append(expansion['template'])
+
+
 # from :
 # http://stackoverflow.com/questions/8640959/how-can-i-control-what-scalar-form-pyyaml-uses-for-my-data  flake8: noqa
 def should_use_block(value):
@@ -885,7 +919,7 @@
         elif 'openstack/puppet-openstack-integration' in expanded_projects:
             output['parent'] = 'legacy-puppet-openstack-integration'
         elif has_artifacts:
-            output['parent'] = 'publish-openstack-artifacts'
+            output['parent'] = 'legacy-publish-openstack-artifacts'
         elif has_draft:
             output['success-url'] = 'html/'
         output['run'] = os.path.join(self.job_path, 'run')
@@ -898,8 +932,6 @@
         timeout = self.getTimeout()
         if timeout:
             output['timeout'] = timeout
-            output.setdefault('vars', {})
-            output['vars']['BUILD_TIMEOUT'] = str(timeout * 1000)
 
         if self.nodes:
             if len(self.nodes) == 1:
@@ -910,6 +942,14 @@
         if expanded_projects:
             output['required-projects'] = sorted(list(set(expanded_projects)))
 
+        if self.name in JOB_MATCHERS:
+            for k, v in JOB_MATCHERS[self.name].items():
+                if k in output:
+                    self.log.error(
+                        'Job %s has attributes directly and from matchers',
+                        self.name)
+                output[k] = v
+
         return output
 
     def toPipelineDict(self):
@@ -1345,7 +1385,7 @@
         for pipeline, value in template.items():
             if pipeline == 'name':
                 continue
-            if pipeline not in project:
+            if pipeline not in project or 'jobs' not in project[pipeline]:
                 project[pipeline] = dict(jobs=[])
             project[pipeline]['jobs'].extend(value['jobs'])
 
@@ -1355,7 +1395,7 @@
                 return job.orig
         return None
 
-    def applyProjectMatchers(self, matchers, project):
+    def applyProjectMatchers(self, matchers, project, final=False):
         '''
         Apply per-project job matchers to the given project.
 
@@ -1373,7 +1413,8 @@
                         self.log.debug(
                             "Applied irrelevant-files to job %s in project %s",
                             job, project['name'])
-                        job = {job: {'irrelevant-files': list(set(files))}}
+                        job = {job: {'irrelevant-files':
+                                     sorted(list(set(files)))}}
                 elif isinstance(job, dict):
                     job = job.copy()
                     job_name = get_single_key(job)
@@ -1387,8 +1428,8 @@
                         if 'irrelevant-files' not in extras:
                             extras['irrelevant-files'] = []
                         extras['irrelevant-files'].extend(files)
-                        extras['irrelevant-files'] = list(
-                            set(extras['irrelevant-files']))
+                        extras['irrelevant-files'] = sorted(list(
+                            set(extras['irrelevant-files'])))
                     job[job_name] = extras
                 new_jobs.append(job)
             return new_jobs
@@ -1398,17 +1439,61 @@
                 if k in ('templates', 'name'):
                     continue
                 project[k]['jobs'] = processPipeline(
-                    project[k]['jobs'], job_name_regex, files)
+                    project[k].get('jobs', []), job_name_regex, files)
 
-        for matcher in matchers:
-            # find the project-specific section
-            for skipper in matcher.get('skip-if', []):
-                if skipper.get('project'):
-                    if re.search(skipper['project'], project['name']):
-                        if 'all-files-match-any' in skipper:
-                            applyIrrelevantFiles(
-                                matcher['name'],
-                                skipper['all-files-match-any'])
+        if matchers:
+            for matcher in matchers:
+                # find the project-specific section
+                for skipper in matcher.get('skip-if', []):
+                    if skipper.get('project'):
+                        if re.search(skipper['project'], project['name']):
+                            if 'all-files-match-any' in skipper:
+                                applyIrrelevantFiles(
+                                    matcher['name'],
+                                    skipper['all-files-match-any'])
+
+        if not final:
+            return
+
+        for k, v in project.items():
+            if k in ('templates', 'name'):
+                continue
+            jobs = []
+            for job in project[k].get('jobs', []):
+                if isinstance(job, dict):
+                    job_name = get_single_key(job)
+                else:
+                    job_name = job
+                if job_name in JOB_MATCHERS:
+                    jobs.append(job)
+                    continue
+                orig_name = self.getOldJobName(job_name)
+                if not orig_name:
+                    jobs.append(job)
+                    continue
+                orig_name = orig_name.format(
+                    name=project['name'].split('/')[1])
+                info = {}
+                for layout_job in self.mapping.layout.get('jobs', []):
+                    if 'parameter-function' in layout_job:
+                        continue
+                    if 'skip-if' in layout_job:
+                        continue
+                    if re.search(layout_job['name'], orig_name):
+                        if not layout_job.get('voting', True):
+                            info['voting'] = False
+                        if layout_job.get('branch'):
+                            info['branches'] = layout_job['branch']
+                        if layout_job.get('files'):
+                            info['files'] = layout_job['files']
+                        if not isinstance(job, dict):
+                            job = {job: info}
+                        else:
+                            job[job_name].update(info)
+
+                jobs.append(job)
+            if jobs:
+                project[k]['jobs'] = jobs
 
     def writeProject(self, project):
         '''
@@ -1423,12 +1508,7 @@
         if 'name' in project:
             new_project['name'] = project['name']
 
-        job_matchers = self.scanForProjectMatchers(project['name'])
-        if job_matchers:
-            exp_template_names = self.findReferencedTemplateNames(
-                job_matchers, project['name'])
-        else:
-            exp_template_names = []
+        exp_template_names = TEMPLATES_TO_EXPAND.get(project['name'], [])
 
         templates_to_expand = []
         if 'template' in project:
@@ -1447,6 +1527,51 @@
                 new_project[key] = collections.OrderedDict()
                 if key == 'gate':
                     for queue in self.change_queues:
+                        if (project['name'] not in queue.getProjects() or
+                                len(queue.getProjects()) == 1):
+                            continue
+                        new_project[key]['queue'] = queue.name
+                tmp = [job for job in self.makeNewJobs(value)]
+                # Don't insert into self.job_objects - that was done
+                # in the speculative pass
+                jobs = [job.toPipelineDict() for job in tmp]
+                if jobs:
+                    new_project[key]['jobs'] = jobs
+                if not new_project[key]:
+                    del new_project[key]
+
+        for name in templates_to_expand:
+            self.expandTemplateIntoProject(name, new_project)
+
+        job_matchers = self.scanForProjectMatchers(project['name'])
+
+        # Need a deep copy after expansion, else our templates end up
+        # also getting this change.
+        new_project = copy.deepcopy(new_project)
+        self.applyProjectMatchers(job_matchers, new_project, final=True)
+
+        return new_project
+
+    def checkSpeculativeProject(self, project):
+        '''
+        Create a new v3 project definition expanding all templates.
+        '''
+        new_project = collections.OrderedDict()
+        if 'name' in project:
+            new_project['name'] = project['name']
+
+        templates_to_expand = []
+        for template in project.get('template', []):
+            templates_to_expand.append(template['name'])
+
+        # We have to do this section to expand self.job_objects
+        for key, value in project.items():
+            if key in ('name', 'template'):
+                continue
+            else:
+                new_project[key] = collections.OrderedDict()
+                if key == 'gate':
+                    for queue in self.change_queues:
                         if project['name'] not in queue.getProjects():
                             continue
                         if len(queue.getProjects()) == 1:
@@ -1454,18 +1579,60 @@
                         new_project[key]['queue'] = queue.name
                 tmp = [job for job in self.makeNewJobs(value)]
                 self.job_objects.extend(tmp)
-                jobs = [job.toPipelineDict() for job in tmp]
-                new_project[key]['jobs'] = jobs
 
         for name in templates_to_expand:
-            self.expandTemplateIntoProject(name, new_project)
 
-        # Need a deep copy after expansion, else our templates end up
-        # also getting this change.
-        new_project = copy.deepcopy(new_project)
-        self.applyProjectMatchers(job_matchers, new_project)
+            expand_project = copy.deepcopy(new_project)
+            self.expandTemplateIntoProject(name, expand_project)
 
-        return new_project
+            # Need a deep copy after expansion, else our templates end up
+            # also getting this change.
+            expand_project = copy.deepcopy(expand_project)
+            job_matchers = self.scanForProjectMatchers(project['name'])
+            self.applyProjectMatchers(job_matchers, expand_project)
+
+            # We should now have a project-pipeline with only the
+            # jobs expanded from this one template
+            for project_part in expand_project.values():
+                # The pipelines are dicts - we only want pipelines
+                if isinstance(project_part, dict):
+                    if 'jobs' not in project_part:
+                        continue
+                    self.processProjectTemplateExpansion(
+                        project_part, project, name)
+
+    def processProjectTemplateExpansion(self, project_part, project, template):
+        # project_part should be {'jobs': []}
+        job_list = project_part['jobs']
+        for new_job in job_list:
+            if isinstance(new_job, dict):
+                new_job_name = get_single_key(new_job)
+                info = new_job[new_job_name]
+            else:
+                new_job_name = new_job
+                info = None
+            orig_name = self.getOldJobName(new_job_name)
+            if not orig_name:
+                self.log.error("Job without old name: %s", new_job_name)
+                continue
+            orig_name = orig_name.format(name=project['name'].split('/')[1])
+
+            for layout_job in self.mapping.layout.get('jobs', []):
+                if 'parameter-function' in layout_job:
+                    continue
+                if re.search(layout_job['name'], orig_name):
+                    if not info:
+                        info = {}
+                    if not layout_job.get('voting', True):
+                        info['voting'] = False
+                    if layout_job.get('branch'):
+                        info['branches'] = layout_job['branch']
+                    if layout_job.get('files'):
+                        info['files'] = layout_job['files']
+
+            if info:
+                expansion = dict(info=info, template=template)
+                JOBS_FOR_EXPAND[new_job_name][project['name']] = expansion
 
     def writeJobs(self):
         output_dir = self.setupDir()
@@ -1487,6 +1654,10 @@
             template_config,
             key=lambda template: template['project-template']['name'])
 
+        for project in self.layout.get('projects', []):
+            self.checkSpeculativeProject(project)
+        normalize_project_expansions()
+
         project_names = []
         for project in self.layout.get('projects', []):
             project_names.append(project['name'])
diff --git a/zuul/cmd/scheduler.py b/zuul/cmd/scheduler.py
index a9923c6..d920d8e 100755
--- a/zuul/cmd/scheduler.py
+++ b/zuul/cmd/scheduler.py
@@ -29,6 +29,7 @@
 
 import zuul.cmd
 from zuul.lib.config import get_default
+from zuul.lib.statsd import get_statsd_config
 
 # No zuul imports here because they pull in paramiko which must not be
 # imported until after the daemonization.
@@ -97,8 +98,14 @@
             os.close(pipe_write)
             self.setup_logging('gearman_server', 'log_config')
             import zuul.lib.gearserver
-            statsd_host = os.environ.get('STATSD_HOST')
-            statsd_port = int(os.environ.get('STATSD_PORT', 8125))
+
+            (statsd_host, statsd_port, statsd_prefix) = get_statsd_config(
+                self.config)
+            if statsd_prefix:
+                statsd_prefix += '.zuul.geard'
+            else:
+                statsd_prefix = 'zuul.geard'
+
             host = get_default(self.config, 'gearman_server', 'listen_address')
             port = int(get_default(self.config, 'gearman_server', 'port',
                                    4730))
@@ -112,7 +119,7 @@
                                            host=host,
                                            statsd_host=statsd_host,
                                            statsd_port=statsd_port,
-                                           statsd_prefix='zuul.geard')
+                                           statsd_prefix=statsd_prefix)
 
             # Keep running until the parent dies:
             pipe_read = os.fdopen(pipe_read)
@@ -154,8 +161,10 @@
         zookeeper = zuul.zk.ZooKeeper()
         zookeeper_hosts = get_default(self.config, 'zookeeper',
                                       'hosts', '127.0.0.1:2181')
+        zookeeper_timeout = float(get_default(self.config, 'zookeeper',
+                                              'session_timeout', 10.0))
 
-        zookeeper.connect(zookeeper_hosts)
+        zookeeper.connect(zookeeper_hosts, timeout=zookeeper_timeout)
 
         cache_expiry = get_default(self.config, 'webapp', 'status_expiry', 1)
         listen_address = get_default(self.config, 'webapp', 'listen_address',
diff --git a/zuul/configloader.py b/zuul/configloader.py
index afdf329..426842b 100644
--- a/zuul/configloader.py
+++ b/zuul/configloader.py
@@ -11,6 +11,7 @@
 # under the License.
 
 import base64
+import collections
 from contextlib import contextmanager
 import copy
 import os
@@ -383,57 +384,58 @@
 class JobParser(object):
     ANSIBLE_ROLE_RE = re.compile(r'^(ansible[-_.+]*)*(role[-_.+]*)*')
 
-    @staticmethod
-    def getSchema():
-        zuul_role = {vs.Required('zuul'): str,
-                     'name': str}
+    zuul_role = {vs.Required('zuul'): str,
+                 'name': str}
 
-        galaxy_role = {vs.Required('galaxy'): str,
-                       'name': str}
+    galaxy_role = {vs.Required('galaxy'): str,
+                   'name': str}
 
-        role = vs.Any(zuul_role, galaxy_role)
+    role = vs.Any(zuul_role, galaxy_role)
 
-        job_project = {vs.Required('name'): str,
-                       'override-branch': str}
+    job_project = {vs.Required('name'): str,
+                   'override-branch': str}
 
-        secret = {vs.Required('name'): str,
-                  vs.Required('secret'): str}
+    secret = {vs.Required('name'): str,
+              vs.Required('secret'): str}
 
-        job = {vs.Required('name'): str,
-               'parent': vs.Any(str, None),
-               'final': bool,
-               'failure-message': str,
-               'success-message': str,
-               'failure-url': str,
-               'success-url': str,
-               'hold-following-changes': bool,
-               'voting': bool,
-               'semaphore': str,
-               'tags': to_list(str),
-               'branches': to_list(str),
-               'files': to_list(str),
-               'secrets': to_list(vs.Any(secret, str)),
-               'irrelevant-files': to_list(str),
-               # validation happens in NodeSetParser
-               'nodeset': vs.Any(dict, str),
-               'timeout': int,
-               'attempts': int,
-               'pre-run': to_list(str),
-               'post-run': to_list(str),
-               'run': str,
-               '_source_context': model.SourceContext,
-               '_start_mark': ZuulMark,
-               'roles': to_list(role),
-               'required-projects': to_list(vs.Any(job_project, str)),
-               'vars': dict,
-               'dependencies': to_list(str),
-               'allowed-projects': to_list(str),
-               'override-branch': str,
-               'description': str,
-               'post-review': bool
-               }
+    # Attributes of a job that can also be used in Project and ProjectTemplate
+    job_attributes = {'parent': vs.Any(str, None),
+                      'final': bool,
+                      'failure-message': str,
+                      'success-message': str,
+                      'failure-url': str,
+                      'success-url': str,
+                      'hold-following-changes': bool,
+                      'voting': bool,
+                      'semaphore': str,
+                      'tags': to_list(str),
+                      'branches': to_list(str),
+                      'files': to_list(str),
+                      'secrets': to_list(vs.Any(secret, str)),
+                      'irrelevant-files': to_list(str),
+                      # validation happens in NodeSetParser
+                      'nodeset': vs.Any(dict, str),
+                      'timeout': int,
+                      'attempts': int,
+                      'pre-run': to_list(str),
+                      'post-run': to_list(str),
+                      'run': str,
+                      '_source_context': model.SourceContext,
+                      '_start_mark': ZuulMark,
+                      'roles': to_list(role),
+                      'required-projects': to_list(vs.Any(job_project, str)),
+                      'vars': dict,
+                      'dependencies': to_list(str),
+                      'allowed-projects': to_list(str),
+                      'override-branch': str,
+                      'description': str,
+                      'post-review': bool}
 
-        return vs.Schema(job)
+    job_name = {vs.Required('name'): str}
+
+    job = dict(collections.ChainMap(job_name, job_attributes))
+
+    schema = vs.Schema(job)
 
     simple_attributes = [
         'final',
@@ -456,58 +458,69 @@
         # the reference definition of this job, and this is a project
         # repo, add an implicit branch matcher for this branch
         # (assuming there are no explicit branch matchers).  But only
-        # for top-level job definitions and variants.
-        # Project-pipeline job variants should more closely attach to
-        # their branch if they appear in a project-repo.
+        # for top-level job definitions and variants.  Never for
+        # project-templates.  They, and in-project project-pipeline
+        # job variants, should more closely attach to their branch if
+        # they appear in a project-repo.  That's handled in the
+        # ProjectParser.
         if (reference and
             reference.source_context and
             reference.source_context.branch != job.source_context.branch):
-            same_context = False
+            same_branch = False
         else:
-            same_context = True
+            same_branch = True
 
         if (job.source_context and
             (not job.source_context.trusted) and
-            ((not same_context) or project_pipeline)):
+            (not project_pipeline) and
+            (not same_branch)):
             return [job.source_context.branch]
         return None
 
     @staticmethod
-    def fromYaml(tenant, layout, conf, project_pipeline=False):
-        with configuration_exceptions('job', conf):
-            JobParser.getSchema()(conf)
+    def fromYaml(tenant, layout, conf, project_pipeline=False,
+                 name=None, validate=True):
+        if validate:
+            with configuration_exceptions('job', conf):
+                JobParser.schema(conf)
+
+        if name is None:
+            name = conf['name']
 
         # NB: The default detection system in the Job class requires
         # that we always assign values directly rather than modifying
         # them (e.g., "job.run = ..." rather than
         # "job.run.append(...)").
 
-        reference = layout.jobs.get(conf['name'], [None])[0]
+        reference = layout.jobs.get(name, [None])[0]
 
-        job = model.Job(conf['name'])
+        job = model.Job(name)
         job.source_context = conf.get('_source_context')
+        job.source_line = conf.get('_start_mark').line + 1
 
-        is_variant = layout.hasJob(conf['name'])
-        if 'parent' in conf:
-            if conf['parent'] is not None:
-                # Parent job is explicitly specified, so inherit from it.
-                parent = layout.getJob(conf['parent'])
-                job.inheritFrom(parent)
+        is_variant = layout.hasJob(name)
+        if not is_variant:
+            if 'parent' in conf:
+                if conf['parent'] is not None:
+                    # Parent job is explicitly specified, so inherit from it.
+                    parent = layout.getJob(conf['parent'])
+                    job.inheritFrom(parent)
+                else:
+                    # Parent is explicitly set as None, so user intends
+                    # this to be a base job.  That's only okay if we're in
+                    # a config project.
+                    if not conf['_source_context'].trusted:
+                        raise Exception(
+                            "Base jobs must be defined in config projects")
             else:
-                # Parent is explicitly set as None, so user intends
-                # this to be a base job.  That's only okay if we're in
-                # a config project.
-                if not conf['_source_context'].trusted:
-                    raise Exception(
-                        "Base jobs must be defined in config projects")
-        else:
-            # Parent is not explicitly set, so inherit from the
-            # default -- but only if this is the primary definition
-            # for the job (ie, not a variant -- variants don't need to
-            # have a parent as long as the main job does).
-            if not is_variant:
                 parent = layout.getJob(tenant.default_base_job)
                 job.inheritFrom(parent)
+        else:
+            if 'parent' in conf:
+                # TODO(jeblair): warn the user that we're ignoring the
+                # parent setting on this variant job definition.
+                pass
+
         # Secrets are part of the playbook context so we must establish
         # them earlier than playbooks.
         secrets = []
@@ -668,10 +681,7 @@
         if (not branches) and ('branches' in conf):
             branches = as_list(conf['branches'])
         if branches:
-            matchers = []
-            for branch in branches:
-                matchers.append(change_matcher.BranchMatcher(branch))
-            job.branch_matcher = change_matcher.MatchAny(matchers)
+            job.setBranchMatcher(branches)
         if 'files' in conf:
             matchers = []
             for fn in as_list(conf['files']):
@@ -709,10 +719,13 @@
 
 
 class ProjectTemplateParser(object):
-    log = logging.getLogger("zuul.ProjectTemplateParser")
+    def __init__(self, tenant, layout):
+        self.log = logging.getLogger("zuul.ProjectTemplateParser")
+        self.tenant = tenant
+        self.layout = layout
+        self.schema = self.getSchema()
 
-    @staticmethod
-    def getSchema(layout):
+    def getSchema(self):
         project_template = {
             vs.Required('name'): str,
             'description': str,
@@ -723,47 +736,41 @@
             '_start_mark': ZuulMark,
         }
 
-        for p in layout.pipelines.values():
-            project_template[p.name] = {'queue': str,
-                                        'jobs': [vs.Any(str, dict)]}
+        job = {str: vs.Any(str, JobParser.job_attributes)}
+        job_list = [vs.Any(str, job)]
+        pipeline_contents = {'queue': str, 'jobs': job_list}
+
+        for p in self.layout.pipelines.values():
+            project_template[p.name] = pipeline_contents
         return vs.Schema(project_template)
 
-    @staticmethod
-    def fromYaml(tenant, layout, conf):
-        with configuration_exceptions('project or project-template', conf):
-            ProjectTemplateParser.getSchema(layout)(conf)
-        # Make a copy since we modify this later via pop
-        conf = copy.deepcopy(conf)
+    def fromYaml(self, conf, validate=True):
+        if validate:
+            with configuration_exceptions('project-template', conf):
+                self.schema(conf)
         project_template = model.ProjectConfig(conf['name'])
         source_context = conf['_source_context']
         start_mark = conf['_start_mark']
-        for pipeline in layout.pipelines.values():
+        for pipeline in self.layout.pipelines.values():
             conf_pipeline = conf.get(pipeline.name)
             if not conf_pipeline:
                 continue
             project_pipeline = model.ProjectPipelineConfig()
             project_template.pipelines[pipeline.name] = project_pipeline
             project_pipeline.queue_name = conf_pipeline.get('queue')
-            ProjectTemplateParser._parseJobList(
-                tenant, layout, conf_pipeline.get('jobs', []),
+            self.parseJobList(
+                conf_pipeline.get('jobs', []),
                 source_context, start_mark, project_pipeline.job_list)
         return project_template
 
-    @staticmethod
-    def _parseJobList(tenant, layout, conf, source_context,
-                      start_mark, job_list):
+    def parseJobList(self, conf, source_context, start_mark, job_list):
         for conf_job in conf:
             if isinstance(conf_job, str):
-                attrs = dict(name=conf_job)
+                jobname = conf_job
+                attrs = {}
             elif isinstance(conf_job, dict):
                 # A dictionary in a job tree may override params
                 jobname, attrs = list(conf_job.items())[0]
-                if attrs:
-                    # We are overriding params, so make a new job def
-                    attrs['name'] = jobname
-                else:
-                    # Not overriding, so add a blank job
-                    attrs = dict(name=jobname)
             else:
                 raise Exception("Job must be a string or dictionary")
             attrs['_source_context'] = source_context
@@ -772,17 +779,22 @@
             # validate that the job is existing
             with configuration_exceptions('project or project-template',
                                           attrs):
-                layout.getJob(attrs['name'])
+                self.layout.getJob(jobname)
 
-            job_list.addJob(JobParser.fromYaml(tenant, layout, attrs,
-                                               project_pipeline=True))
+            job_list.addJob(JobParser.fromYaml(self.tenant, self.layout,
+                                               attrs, project_pipeline=True,
+                                               name=jobname, validate=False))
 
 
 class ProjectParser(object):
-    log = logging.getLogger("zuul.ProjectParser")
+    def __init__(self, tenant, layout, project_template_parser):
+        self.log = logging.getLogger("zuul.ProjectParser")
+        self.tenant = tenant
+        self.layout = layout
+        self.project_template_parser = project_template_parser
+        self.schema = self.getSchema()
 
-    @staticmethod
-    def getSchema(layout):
+    def getSchema(self):
         project = {
             vs.Required('name'): str,
             'description': str,
@@ -794,46 +806,52 @@
             '_start_mark': ZuulMark,
         }
 
-        for p in layout.pipelines.values():
-            project[p.name] = {'queue': str,
-                               'jobs': [vs.Any(str, dict)]}
+        job = {str: vs.Any(str, JobParser.job_attributes)}
+        job_list = [vs.Any(str, job)]
+        pipeline_contents = {'queue': str, 'jobs': job_list}
+
+        for p in self.layout.pipelines.values():
+            project[p.name] = pipeline_contents
         return vs.Schema(project)
 
-    @staticmethod
-    def fromYaml(tenant, layout, conf_list):
+    def fromYaml(self, conf_list):
         for conf in conf_list:
             with configuration_exceptions('project', conf):
-                ProjectParser.getSchema(layout)(conf)
+                self.schema(conf)
 
         with configuration_exceptions('project', conf_list[0]):
             project_name = conf_list[0]['name']
-            (trusted, project) = tenant.getProject(project_name)
+            (trusted, project) = self.tenant.getProject(project_name)
             if project is None:
                 raise ProjectNotFoundError(project_name)
             project_config = model.ProjectConfig(project.canonical_name)
 
         configs = []
         for conf in conf_list:
+            implied_branch = None
             with configuration_exceptions('project', conf):
                 if not conf['_source_context'].trusted:
                     if project != conf['_source_context'].project:
                         raise ProjectNotPermittedError()
 
-                # Make a copy since we modify this later via pop
-                conf = copy.deepcopy(conf)
-                conf_templates = conf.pop('templates', [])
+                conf_templates = conf.get('templates', [])
                 # The way we construct a project definition is by
                 # parsing the definition as a template, then applying
                 # all of the templates, including the newly parsed
                 # one, in order.
-                project_template = ProjectTemplateParser.fromYaml(
-                    tenant, layout, conf)
+                project_template = self.project_template_parser.fromYaml(
+                    conf, validate=False)
+                # If this project definition is in a place where it
+                # should get implied branch matchers, set it.
+                if (not conf['_source_context'].trusted):
+                    implied_branch = conf['_source_context'].branch
                 for name in conf_templates:
-                    if name not in layout.project_templates:
+                    if name not in self.layout.project_templates:
                         raise TemplateNotFoundError(name)
-                configs.extend([layout.project_templates[name]
+                configs.extend([(self.layout.project_templates[name],
+                                 implied_branch)
                                 for name in conf_templates])
-                configs.append(project_template)
+                configs.append((project_template, implied_branch))
                 # Set the following values to the first one that we
                 # find and ignore subsequent settings.
                 mode = conf.get('merge-mode')
@@ -848,21 +866,19 @@
             project_config.merge_mode = model.MERGER_MAP['merge-resolve']
         if project_config.default_branch is None:
             project_config.default_branch = 'master'
-        for pipeline in layout.pipelines.values():
+        for pipeline in self.layout.pipelines.values():
             project_pipeline = model.ProjectPipelineConfig()
             queue_name = None
             # For every template, iterate over the job tree and replace or
             # create the jobs in the final definition as needed.
             pipeline_defined = False
-            for template in configs:
+            for (template, implied_branch) in configs:
                 if pipeline.name in template.pipelines:
-                    ProjectParser.log.debug(
-                        "Applying template %s to pipeline %s" %
-                        (template.name, pipeline.name))
                     pipeline_defined = True
                     template_pipeline = template.pipelines[pipeline.name]
                     project_pipeline.job_list.inheritFrom(
-                        template_pipeline.job_list)
+                        template_pipeline.job_list,
+                        implied_branch)
                     if template_pipeline.queue_name:
                         queue_name = template_pipeline.queue_name
             if queue_name:
@@ -1358,7 +1374,12 @@
             # branch.  Remember the branch and then implicitly add a
             # branch selector to each job there.  This makes the
             # in-repo configuration apply only to that branch.
-            for branch in project.source.getProjectBranches(project, tenant):
+            branches = sorted(project.source.getProjectBranches(
+                project, tenant))
+            if 'master' in branches:
+                branches.remove('master')
+                branches = ['master'] + branches
+            for branch in branches:
                 new_project_unparsed_branch_config[project][branch] = \
                     model.UnparsedTenantConfig()
                 job = merger.getFiles(
@@ -1406,19 +1427,20 @@
                             (job.source_context,))
                         continue
                     loaded = conf_root
-                    job.source_context.path = fn
+                    source_context = job.source_context.copy()
+                    source_context.path = fn
                     TenantParser.log.info(
                         "Loading configuration from %s" %
-                        (job.source_context,))
-                    project = job.source_context.project
-                    branch = job.source_context.branch
-                    if job.source_context.trusted:
+                        (source_context,))
+                    project = source_context.project
+                    branch = source_context.branch
+                    if source_context.trusted:
                         incdata = TenantParser._parseConfigProjectLayout(
-                            job.files[fn], job.source_context)
+                            job.files[fn], source_context)
                         config_projects_config.extend(incdata)
                     else:
                         incdata = TenantParser._parseUntrustedProjectLayout(
-                            job.files[fn], job.source_context)
+                            job.files[fn], source_context)
                         untrusted_projects_config.extend(incdata)
                     new_project_unparsed_config[project].extend(incdata)
                     if branch in new_project_unparsed_branch_config.get(
@@ -1516,13 +1538,15 @@
                     continue
                 layout.addSemaphore(semaphore)
 
+        project_template_parser = ProjectTemplateParser(tenant, layout)
         for config_template in data.project_templates:
             classes = TenantParser._getLoadClasses(tenant, config_template)
             if 'project-template' not in classes:
                 continue
-            layout.addProjectTemplate(ProjectTemplateParser.fromYaml(
-                tenant, layout, config_template))
+            layout.addProjectTemplate(project_template_parser.fromYaml(
+                config_template))
 
+        project_parser = ProjectParser(tenant, layout, project_template_parser)
         for config_projects in data.projects.values():
             # Unlike other config classes, we expect multiple project
             # stanzas with the same name, so that a config repo can
@@ -1540,14 +1564,15 @@
             if not filtered_projects:
                 continue
 
-            layout.addProjectConfig(ProjectParser.fromYaml(
-                tenant, layout, filtered_projects))
+            layout.addProjectConfig(project_parser.fromYaml(
+                filtered_projects))
 
     @staticmethod
     def _parseLayout(base, tenant, data, scheduler, connections):
         # Don't call this method from dynamic reconfiguration because
         # it interacts with drivers and connections.
         layout = model.Layout(tenant)
+        TenantParser.log.debug("Created layout id %s", layout.uuid)
 
         TenantParser._parseLayoutItems(layout, tenant, data,
                                        scheduler, connections)
@@ -1615,9 +1640,22 @@
         for branch in branches:
             fns1 = []
             fns2 = []
-            files_list = files.connections.get(
+            files_entry = files.connections.get(
                 project.source.connection.connection_name, {}).get(
-                    project.name, {}).get(branch, {}).keys()
+                    project.name, {}).get(branch)
+            # If there is no files entry at all for this
+            # project-branch, then use the cached config.
+            if files_entry is None:
+                if trusted:
+                    incdata = project.unparsed_config
+                else:
+                    incdata = project.unparsed_branch_config.get(branch)
+                if incdata:
+                    config.extend(incdata)
+                continue
+            # Otherwise, do not use the cached config (even if the
+            # files are empty as that likely means they were deleted).
+            files_list = files_entry.keys()
             for fn in files_list:
                 if fn.startswith("zuul.d/"):
                     fns1.append(fn)
@@ -1649,14 +1687,6 @@
 
                     config.extend(incdata)
 
-            if not loaded:
-                if trusted:
-                    incdata = project.unparsed_config
-                else:
-                    incdata = project.unparsed_branch_config.get(branch)
-                if incdata:
-                    config.extend(incdata)
-
     def createDynamicLayout(self, tenant, files,
                             include_config_projects=False,
                             scheduler=None, connections=None):
@@ -1671,6 +1701,7 @@
             self._loadDynamicProjectData(config, project, files, False, tenant)
 
         layout = model.Layout(tenant)
+        self.log.debug("Created layout id %s", layout.uuid)
         if not include_config_projects:
             # NOTE: the actual pipeline objects (complete with queues
             # and enqueued items) are copied by reference here.  This
diff --git a/zuul/connection/__init__.py b/zuul/connection/__init__.py
index 3655115..b44fa46 100644
--- a/zuul/connection/__init__.py
+++ b/zuul/connection/__init__.py
@@ -14,8 +14,6 @@
 
 import abc
 
-import extras
-
 
 class BaseConnection(object, metaclass=abc.ABCMeta):
     """Base class for connections.
@@ -42,20 +40,18 @@
         self.driver = driver
         self.connection_name = connection_name
         self.connection_config = connection_config
-        self.statsd = extras.try_import('statsd.statsd')
 
     def logEvent(self, event):
         self.log.debug(
-            'Scheduling {driver} event from {connection}: {event}'.format(
-                driver=self.driver.name,
+            'Scheduling event from {connection}: {event}'.format(
                 connection=self.connection_name,
-                event=event.type))
+                event=event))
         try:
-            if self.statsd:
-                self.statsd.incr(
+            if self.sched.statsd:
+                self.sched.statsd.incr(
                     'zuul.event.{driver}.{event}'.format(
                         driver=self.driver.name, event=event.type))
-                self.statsd.incr(
+                self.sched.statsd.incr(
                     'zuul.event.{driver}.{connection}.{event}'.format(
                         driver=self.driver.name,
                         connection=self.connection_name,
diff --git a/zuul/driver/sql/sqlreporter.py b/zuul/driver/sql/sqlreporter.py
index d358d69..6a95212 100644
--- a/zuul/driver/sql/sqlreporter.py
+++ b/zuul/driver/sql/sqlreporter.py
@@ -62,15 +62,19 @@
 
                 (result, url) = item.formatJobResult(job)
 
+                start = end = None
+                if build.start_time:
+                    start = datetime.datetime.fromtimestamp(build.start_time)
+                if build.end_time:
+                    end = datetime.datetime.fromtimestamp(build.end_time)
+
                 build_inserts.append({
                     'buildset_id': buildset_ins_result.inserted_primary_key,
                     'uuid': build.uuid,
                     'job_name': build.job.name,
                     'result': result,
-                    'start_time': datetime.datetime.fromtimestamp(
-                        build.start_time),
-                    'end_time': datetime.datetime.fromtimestamp(
-                        build.end_time),
+                    'start_time': start,
+                    'end_time': end,
                     'voting': build.job.voting,
                     'log_url': url,
                     'node_name': build.node_name,
diff --git a/zuul/executor/client.py b/zuul/executor/client.py
index f97d286..ae22c8e 100644
--- a/zuul/executor/client.py
+++ b/zuul/executor/client.py
@@ -101,7 +101,7 @@
             job = super(ZuulGearmanClient, self).handleStatusRes(packet)
         except gear.UnknownJobError:
             handle = packet.getArgument(0)
-            for build in self.__zuul_gearman.builds:
+            for build in self.__zuul_gearman.builds.values():
                 if build.__gearman_job.handle == handle:
                     self.__zuul_gearman.onUnknownJob(job)
 
@@ -168,7 +168,8 @@
                            project=project,
                            tenant=tenant.name,
                            timeout=job.timeout,
-                           jobtags=sorted(job.tags))
+                           jobtags=sorted(job.tags),
+                           _inheritance_path=list(job.inheritance_path))
         if hasattr(item.change, 'branch'):
             zuul_params['branch'] = item.change.branch
         if hasattr(item.change, 'tag'):
@@ -186,6 +187,7 @@
             and item.change.newrev != '0' * 40):
             zuul_params['newrev'] = item.change.newrev
         zuul_params['projects'] = []  # Set below
+        zuul_params['_projects'] = {}  # transitional to convert to dict
         zuul_params['items'] = []
         for i in all_items:
             d = dict()
@@ -234,9 +236,10 @@
         params['vars'] = copy.deepcopy(job.variables)
         params['zuul'] = zuul_params
         projects = set()
+        required_projects = set()
 
         def make_project_dict(project, override_branch=None):
-            project_config = item.current_build_set.layout.project_configs.get(
+            project_config = item.layout.project_configs.get(
                 project.canonical_name, None)
             if project_config:
                 project_default_branch = project_config.default_branch
@@ -260,6 +263,7 @@
                     make_project_dict(project,
                                       job_project.override_branch))
                 projects.add(project)
+                required_projects.add(project)
         for i in all_items:
             if i.change.project not in projects:
                 project = i.change.project
@@ -267,18 +271,30 @@
                 projects.add(project)
 
         for p in projects:
-            zuul_params['projects'].append(dict(
+            zuul_params['_projects'][p.canonical_name] = (dict(
                 name=p.name,
                 short_name=p.name.split('/')[-1],
-                canonical_hostname=p.canonical_hostname,
+                # Duplicate this into the dict too, so that iterating
+                # project.values() is easier for callers
                 canonical_name=p.canonical_name,
+                canonical_hostname=p.canonical_hostname,
                 src_dir=os.path.join('src', p.canonical_name),
+                required=(p in required_projects),
             ))
+        # We are transitioning "projects" from a list to a dict
+        # indexed by canonical name, as it is much easier to access
+        # values in ansible.  Existing callers are converted to
+        # "_projects", then once "projects" is unused we switch it,
+        # then convert callers back.  Finally when "_projects" is
+        # unused it will be removed.
+        for cn, p in zuul_params['_projects'].items():
+            zuul_params['projects'].append(p)
 
         build = Build(job, uuid)
         build.parameters = params
 
         if job.name == 'noop':
+            self.sched.onBuildStarted(build)
             self.sched.onBuildCompleted(build, 'SUCCESS', {})
             return build
 
@@ -368,9 +384,17 @@
                     result = 'RETRY_LIMIT'
                 else:
                     build.retry = True
+            if result in ('DISCONNECT', 'ABORTED'):
+                # Always retry if the executor just went away
+                build.retry = True
             result_data = data.get('data', {})
             self.log.info("Build %s complete, result %s" %
                           (job, result))
+            # If the build should be retried, don't supply the result
+            # so that elsewhere we don't have to deal with keeping
+            # track of which results are non-final.
+            if build.retry:
+                result = None
             self.sched.onBuildCompleted(build, result, result_data)
             # The test suite expects the build to be removed from the
             # internal dict after it's added to the report queue.
@@ -399,7 +423,7 @@
 
     def onDisconnect(self, job):
         self.log.info("Gearman job %s lost due to disconnect" % job)
-        self.onBuildCompleted(job)
+        self.onBuildCompleted(job, 'DISCONNECT')
 
     def onUnknownJob(self, job):
         self.log.info("Gearman job %s lost due to unknown handle" % job)
@@ -440,7 +464,9 @@
 
     def lookForLostBuilds(self):
         self.log.debug("Looking for lost builds")
-        for build in self.builds.values():
+        # Construct a list from the values iterator to protect from it changing
+        # out from underneath us.
+        for build in list(self.builds.values()):
             if build.result:
                 # The build has finished, it will be removed
                 continue
diff --git a/zuul/executor/server.py b/zuul/executor/server.py
index af2e7e9..e7a6dbc 100644
--- a/zuul/executor/server.py
+++ b/zuul/executor/server.py
@@ -16,6 +16,7 @@
 import datetime
 import json
 import logging
+import multiprocessing
 import os
 import shutil
 import signal
@@ -28,6 +29,7 @@
 import traceback
 from zuul.lib.yamlutil import yaml
 from zuul.lib.config import get_default
+from zuul.lib.statsd import get_statsd
 
 try:
     import ara.plugins.callbacks as ara_callbacks
@@ -45,6 +47,11 @@
 DEFAULT_FINGER_PORT = 79
 
 
+class StopException(Exception):
+    """An exception raised when an inner loop is asked to stop."""
+    pass
+
+
 class ExecutorError(Exception):
     """A non-transient run-time executor error
 
@@ -88,7 +95,7 @@
         if cache_dir == jobs_base:
             raise Exception("Cache dir and jobs dir cannot be the same")
         self.thread = threading.Thread(target=self._run,
-                                       name='executor-diskaccountant')
+                                       name='diskaccountant')
         self.thread.daemon = True
         self._running = False
         self.jobs_base = jobs_base
@@ -147,7 +154,7 @@
         self.function = function
         self.args = args
         self.thread = threading.Thread(target=self._run,
-                                       name='executor-watchdog')
+                                       name='watchdog')
         self.thread.daemon = True
         self.timed_out = None
 
@@ -284,6 +291,7 @@
         #     inventory.yaml
         #   .ansible (mounted in bwrap read-write)
         #     fact-cache/localhost
+        #     cp
         #   playbook_0 (mounted in bwrap for each playbook read-only)
         #     secrets.yaml
         #     project -> ../trusted/project_0/...
@@ -324,6 +332,8 @@
         self.ansible_cache_root = os.path.join(self.root, '.ansible')
         self.fact_cache = os.path.join(self.ansible_cache_root, 'fact-cache')
         os.makedirs(self.fact_cache)
+        self.control_path = os.path.join(self.ansible_cache_root, 'cp')
+        os.makedirs(self.control_path)
         localhost_facts = os.path.join(self.fact_cache, 'localhost')
         # NOTE(pabelanger): We do not want to leak zuul-executor facts to other
         # playbooks now that smart fact gathering is enabled by default.  We
@@ -499,353 +509,6 @@
     return inventory
 
 
-class ExecutorMergeWorker(gear.TextWorker):
-    def __init__(self, executor_server, *args, **kw):
-        self.zuul_executor_server = executor_server
-        super(ExecutorMergeWorker, self).__init__(*args, **kw)
-
-    def handleNoop(self, packet):
-        # Wait until the update queue is empty before responding
-        while self.zuul_executor_server.update_queue.qsize():
-            time.sleep(1)
-
-        with self.zuul_executor_server.merger_lock:
-            super(ExecutorMergeWorker, self).handleNoop(packet)
-
-
-class ExecutorServer(object):
-    log = logging.getLogger("zuul.ExecutorServer")
-
-    def __init__(self, config, connections={}, jobdir_root=None,
-                 keep_jobdir=False, log_streaming_port=DEFAULT_FINGER_PORT):
-        self.config = config
-        self.keep_jobdir = keep_jobdir
-        self.jobdir_root = jobdir_root
-        # TODOv3(mordred): make the executor name more unique --
-        # perhaps hostname+pid.
-        self.hostname = socket.gethostname()
-        self.log_streaming_port = log_streaming_port
-        self.merger_lock = threading.Lock()
-        self.verbose = False
-        self.command_map = dict(
-            stop=self.stop,
-            pause=self.pause,
-            unpause=self.unpause,
-            graceful=self.graceful,
-            verbose=self.verboseOn,
-            unverbose=self.verboseOff,
-            keep=self.keep,
-            nokeep=self.nokeep,
-        )
-
-        self.merge_root = get_default(self.config, 'executor', 'git_dir',
-                                      '/var/lib/zuul/executor-git')
-        self.default_username = get_default(self.config, 'executor',
-                                            'default_username', 'zuul')
-        self.disk_limit_per_job = int(get_default(self.config, 'executor',
-                                                  'disk_limit_per_job', 250))
-        self.merge_email = get_default(self.config, 'merger', 'git_user_email')
-        self.merge_name = get_default(self.config, 'merger', 'git_user_name')
-        execution_wrapper_name = get_default(self.config, 'executor',
-                                             'execution_wrapper', 'bubblewrap')
-        self.execution_wrapper = connections.drivers[execution_wrapper_name]
-
-        self.connections = connections
-        # This merger and its git repos are used to maintain
-        # up-to-date copies of all the repos that are used by jobs, as
-        # well as to support the merger:cat functon to supply
-        # configuration information to Zuul when it starts.
-        self.merger = self._getMerger(self.merge_root)
-        self.update_queue = DeduplicateQueue()
-
-        state_dir = get_default(self.config, 'executor', 'state_dir',
-                                '/var/lib/zuul', expand_user=True)
-        path = os.path.join(state_dir, 'executor.socket')
-        self.command_socket = commandsocket.CommandSocket(path)
-        ansible_dir = os.path.join(state_dir, 'ansible')
-        self.ansible_dir = ansible_dir
-        if os.path.exists(ansible_dir):
-            shutil.rmtree(ansible_dir)
-
-        zuul_dir = os.path.join(ansible_dir, 'zuul')
-        plugin_dir = os.path.join(zuul_dir, 'ansible')
-
-        os.makedirs(plugin_dir, mode=0o0755)
-
-        self.library_dir = os.path.join(plugin_dir, 'library')
-        self.action_dir = os.path.join(plugin_dir, 'action')
-        self.callback_dir = os.path.join(plugin_dir, 'callback')
-        self.lookup_dir = os.path.join(plugin_dir, 'lookup')
-        self.filter_dir = os.path.join(plugin_dir, 'filter')
-
-        _copy_ansible_files(zuul.ansible, plugin_dir)
-
-        # We're copying zuul.ansible.* into a directory we are going
-        # to add to pythonpath, so our plugins can "import
-        # zuul.ansible".  But we're not installing all of zuul, so
-        # create a __init__.py file for the stub "zuul" module.
-        with open(os.path.join(zuul_dir, '__init__.py'), 'w'):
-            pass
-
-        self.job_workers = {}
-        self.disk_accountant = DiskAccountant(self.jobdir_root,
-                                              self.disk_limit_per_job,
-                                              self.stopJobByJobdir,
-                                              self.merge_root)
-
-    def _getMerger(self, root, logger=None):
-        if root != self.merge_root:
-            cache_root = self.merge_root
-        else:
-            cache_root = None
-        return zuul.merger.merger.Merger(root, self.connections,
-                                         self.merge_email, self.merge_name,
-                                         cache_root, logger)
-
-    def start(self):
-        self._running = True
-        self._command_running = True
-        server = self.config.get('gearman', 'server')
-        port = get_default(self.config, 'gearman', 'port', 4730)
-        ssl_key = get_default(self.config, 'gearman', 'ssl_key')
-        ssl_cert = get_default(self.config, 'gearman', 'ssl_cert')
-        ssl_ca = get_default(self.config, 'gearman', 'ssl_ca')
-        self.merger_worker = ExecutorMergeWorker(self, 'Zuul Executor Merger')
-        self.merger_worker.addServer(server, port, ssl_key, ssl_cert, ssl_ca)
-        self.executor_worker = gear.TextWorker('Zuul Executor Server')
-        self.executor_worker.addServer(server, port, ssl_key, ssl_cert, ssl_ca)
-        self.log.debug("Waiting for server")
-        self.merger_worker.waitForServer()
-        self.executor_worker.waitForServer()
-        self.log.debug("Registering")
-        self.register()
-
-        self.log.debug("Starting command processor")
-        self.command_socket.start()
-        self.command_thread = threading.Thread(target=self.runCommand)
-        self.command_thread.daemon = True
-        self.command_thread.start()
-
-        self.log.debug("Starting worker")
-        self.update_thread = threading.Thread(target=self._updateLoop)
-        self.update_thread.daemon = True
-        self.update_thread.start()
-        self.merger_thread = threading.Thread(target=self.run_merger)
-        self.merger_thread.daemon = True
-        self.merger_thread.start()
-        self.executor_thread = threading.Thread(target=self.run_executor)
-        self.executor_thread.daemon = True
-        self.executor_thread.start()
-        self.disk_accountant.start()
-
-    def register(self):
-        self.executor_worker.registerFunction("executor:execute")
-        self.executor_worker.registerFunction("executor:stop:%s" %
-                                              self.hostname)
-        self.merger_worker.registerFunction("merger:merge")
-        self.merger_worker.registerFunction("merger:cat")
-        self.merger_worker.registerFunction("merger:refstate")
-
-    def stop(self):
-        self.log.debug("Stopping")
-        self.disk_accountant.stop()
-        self._running = False
-        self._command_running = False
-        self.command_socket.stop()
-        self.update_queue.put(None)
-
-        for job_worker in list(self.job_workers.values()):
-            try:
-                job_worker.stop()
-            except Exception:
-                self.log.exception("Exception sending stop command "
-                                   "to worker:")
-        self.merger_worker.shutdown()
-        self.executor_worker.shutdown()
-        self.log.debug("Stopped")
-
-    def pause(self):
-        # TODOv3: implement
-        pass
-
-    def unpause(self):
-        # TODOv3: implement
-        pass
-
-    def graceful(self):
-        # TODOv3: implement
-        pass
-
-    def verboseOn(self):
-        self.verbose = True
-
-    def verboseOff(self):
-        self.verbose = False
-
-    def keep(self):
-        self.keep_jobdir = True
-
-    def nokeep(self):
-        self.keep_jobdir = False
-
-    def join(self):
-        self.update_thread.join()
-        self.merger_thread.join()
-        self.executor_thread.join()
-
-    def runCommand(self):
-        while self._command_running:
-            try:
-                command = self.command_socket.get().decode('utf8')
-                if command != '_stop':
-                    self.command_map[command]()
-            except Exception:
-                self.log.exception("Exception while processing command")
-
-    def _updateLoop(self):
-        while self._running:
-            try:
-                self._innerUpdateLoop()
-            except:
-                self.log.exception("Exception in update thread:")
-
-    def _innerUpdateLoop(self):
-        # Inside of a loop that keeps the main repositories up to date
-        task = self.update_queue.get()
-        if task is None:
-            # We are asked to stop
-            return
-        with self.merger_lock:
-            self.log.info("Updating repo %s/%s" % (
-                task.connection_name, task.project_name))
-            self.merger.updateRepo(task.connection_name, task.project_name)
-            self.log.debug("Finished updating repo %s/%s" %
-                           (task.connection_name, task.project_name))
-        task.setComplete()
-
-    def update(self, connection_name, project_name):
-        # Update a repository in the main merger
-        task = UpdateTask(connection_name, project_name)
-        task = self.update_queue.put(task)
-        return task
-
-    def run_merger(self):
-        self.log.debug("Starting merger listener")
-        while self._running:
-            try:
-                job = self.merger_worker.getJob()
-                try:
-                    if job.name == 'merger:cat':
-                        self.log.debug("Got cat job: %s" % job.unique)
-                        self.cat(job)
-                    elif job.name == 'merger:merge':
-                        self.log.debug("Got merge job: %s" % job.unique)
-                        self.merge(job)
-                    elif job.name == 'merger:refstate':
-                        self.log.debug("Got refstate job: %s" % job.unique)
-                        self.refstate(job)
-                    else:
-                        self.log.error("Unable to handle job %s" % job.name)
-                        job.sendWorkFail()
-                except Exception:
-                    self.log.exception("Exception while running job")
-                    job.sendWorkException(
-                        traceback.format_exc().encode('utf8'))
-            except gear.InterruptedError:
-                pass
-            except Exception:
-                self.log.exception("Exception while getting job")
-
-    def run_executor(self):
-        self.log.debug("Starting executor listener")
-        while self._running:
-            try:
-                job = self.executor_worker.getJob()
-                try:
-                    if job.name == 'executor:execute':
-                        self.log.debug("Got execute job: %s" % job.unique)
-                        self.executeJob(job)
-                    elif job.name.startswith('executor:stop'):
-                        self.log.debug("Got stop job: %s" % job.unique)
-                        self.stopJob(job)
-                    else:
-                        self.log.error("Unable to handle job %s" % job.name)
-                        job.sendWorkFail()
-                except Exception:
-                    self.log.exception("Exception while running job")
-                    job.sendWorkException(
-                        traceback.format_exc().encode('utf8'))
-            except gear.InterruptedError:
-                pass
-            except Exception:
-                self.log.exception("Exception while getting job")
-
-    def executeJob(self, job):
-        self.job_workers[job.unique] = AnsibleJob(self, job)
-        self.job_workers[job.unique].run()
-
-    def finishJob(self, unique):
-        del(self.job_workers[unique])
-
-    def stopJobByJobdir(self, jobdir):
-        unique = os.path.basename(jobdir)
-        self.stopJobByUnique(unique)
-
-    def stopJob(self, job):
-        try:
-            args = json.loads(job.arguments)
-            self.log.debug("Stop job with arguments: %s" % (args,))
-            unique = args['uuid']
-            self.stopJobByUnique(unique)
-        finally:
-            job.sendWorkComplete()
-
-    def stopJobByUnique(self, unique):
-        job_worker = self.job_workers.get(unique)
-        if not job_worker:
-            self.log.debug("Unable to find worker for job %s" % (unique,))
-            return
-        try:
-            job_worker.stop()
-        except Exception:
-            self.log.exception("Exception sending stop command "
-                               "to worker:")
-
-    def cat(self, job):
-        args = json.loads(job.arguments)
-        task = self.update(args['connection'], args['project'])
-        task.wait()
-        with self.merger_lock:
-            files = self.merger.getFiles(args['connection'], args['project'],
-                                         args['branch'], args['files'],
-                                         args.get('dirs', []))
-        result = dict(updated=True,
-                      files=files)
-        job.sendWorkComplete(json.dumps(result))
-
-    def refstate(self, job):
-        args = json.loads(job.arguments)
-        with self.merger_lock:
-            success, repo_state = self.merger.getRepoState(args['items'])
-        result = dict(updated=success,
-                      repo_state=repo_state)
-        job.sendWorkComplete(json.dumps(result))
-
-    def merge(self, job):
-        args = json.loads(job.arguments)
-        with self.merger_lock:
-            ret = self.merger.mergeChanges(args['items'], args.get('files'),
-                                           args.get('dirs', []),
-                                           args.get('repo_state'))
-        result = dict(merged=(ret is not None))
-        if ret is None:
-            result['commit'] = result['files'] = result['repo_state'] = None
-        else:
-            (result['commit'], result['files'], result['repo_state'],
-             recent) = ret
-        job.sendWorkComplete(json.dumps(result))
-
-
 class AnsibleJobLogAdapter(logging.LoggerAdapter):
     def process(self, msg, kwargs):
         msg, kwargs = super(AnsibleJobLogAdapter, self).process(msg, kwargs)
@@ -858,12 +521,14 @@
     RESULT_TIMED_OUT = 2
     RESULT_UNREACHABLE = 3
     RESULT_ABORTED = 4
+    RESULT_DISK_FULL = 5
 
     RESULT_MAP = {
         RESULT_NORMAL: 'RESULT_NORMAL',
         RESULT_TIMED_OUT: 'RESULT_TIMED_OUT',
         RESULT_UNREACHABLE: 'RESULT_UNREACHABLE',
         RESULT_ABORTED: 'RESULT_ABORTED',
+        RESULT_DISK_FULL: 'RESULT_DISK_FULL',
     }
 
     def __init__(self, executor_server, job):
@@ -876,6 +541,7 @@
         self.proc_lock = threading.Lock()
         self.running = False
         self.aborted = False
+        self.aborted_reason = None
         self.thread = None
         self.private_key_file = get_default(self.executor_server.config,
                                             'executor', 'private_key_file',
@@ -890,12 +556,16 @@
 
     def run(self):
         self.running = True
-        self.thread = threading.Thread(target=self.execute)
+        self.thread = threading.Thread(target=self.execute,
+                                       name='build-%s' % self.job.unique)
         self.thread.start()
 
-    def stop(self):
+    def stop(self, reason=None):
         self.aborted = True
+        self.aborted_reason = reason
         self.abortRunningProc()
+
+    def wait(self):
         if self.thread:
             self.thread.join()
 
@@ -1042,6 +712,13 @@
         self.job.sendWorkStatus(0, 100)
 
         result = self.runPlaybooks(args)
+
+        # Stop the persistent SSH connections.
+        setup_status, setup_code = self.runAnsibleCleanup(
+            self.jobdir.setup_playbook)
+
+        if self.aborted_reason == self.RESULT_DISK_FULL:
+            result = 'DISK_FULL'
         data = self.getResultData()
         result_data = json.dumps(dict(result=result,
                                       data=data))
@@ -1158,13 +835,44 @@
             # TODOv3(pabelanger): Implement post-run timeout setting.
             post_status, post_code = self.runAnsiblePlaybook(
                 playbook, args['timeout'], success, phase='post', index=index)
+            if post_status == self.RESULT_ABORTED:
+                return 'ABORTED'
             if post_status != self.RESULT_NORMAL or post_code != 0:
+                success = False
                 # If we encountered a pre-failure, that takes
                 # precedence over the post result.
                 if not pre_failed:
                     result = 'POST_FAILURE'
+                if (index + 1) == len(self.jobdir.post_playbooks):
+                    self._logFinalPlaybookError()
+
         return result
 
+    def _logFinalPlaybookError(self):
+        # Failures in the final post playbook can include failures
+        # uploading logs, which makes diagnosing issues difficult.
+        # Grab the output from the last playbook from the json
+        # file and log it.
+        json_output = self.jobdir.job_output_file.replace('txt', 'json')
+        self.log.debug("Final playbook failed")
+        if not os.path.exists(json_output):
+            self.log.debug("JSON logfile {logfile} is missing".format(
+                logfile=json_output))
+            return
+        try:
+            output = json.load(open(json_output, 'r'))
+            last_playbook = output[-1]
+            # Transform json to yaml - because it's easier to read and given
+            # the size of the data it'll be extra-hard to read this as an
+            # all on one line stringified nested dict.
+            yaml_out = yaml.safe_dump(last_playbook, default_flow_style=False)
+            for line in yaml_out.split('\n'):
+                self.log.debug(line)
+        except Exception:
+            self.log.exception(
+                "Could not decode json from {logfile}".format(
+                    logfile=json_output))
+
     def getHostList(self, args):
         hosts = []
         for node in args['nodes']:
@@ -1491,6 +1199,13 @@
             config.write('display_args_to_stdout = %s\n' %
                          str(not jobdir_playbook.secrets_content))
 
+            # Increase the internal poll interval of ansible.
+            # The default interval of 0.001s is optimized for interactive
+            # ui at the expense of CPU load. As we have a non-interactive
+            # automation use case a longer poll interval is more suitable
+            # and reduces CPU load of the ansible process.
+            config.write('internal_poll_interval = 0.01\n')
+
             config.write('[ssh_connection]\n')
             # NB: when setting pipelining = True, keep_remote_files
             # must be False (the default).  Otherwise it apparently
@@ -1501,6 +1216,7 @@
             # command which expects interactive input on a tty (such
             # as sudo) it does not hang.
             config.write('pipelining = True\n')
+            config.write('control_path_dir = %s\n' % self.jobdir.control_path)
             ssh_args = "-o ControlMaster=auto -o ControlPersist=60s " \
                 "-o UserKnownHostsFile=%s" % self.jobdir.known_hosts
             config.write('ssh_args = %s\n' % ssh_args)
@@ -1522,7 +1238,7 @@
             except Exception:
                 self.log.exception("Exception while killing ansible process:")
 
-    def runAnsible(self, cmd, timeout, playbook):
+    def runAnsible(self, cmd, timeout, playbook, wrapped=True):
         config_file = playbook.ansible_config
         env_copy = os.environ.copy()
         env_copy.update(self.ssh_agent.env)
@@ -1563,8 +1279,12 @@
         if playbook.secrets_content:
             secrets[playbook.secrets] = playbook.secrets_content
 
-        context = self.executor_server.execution_wrapper.getExecutionContext(
-            ro_paths, rw_paths, secrets)
+        if wrapped:
+            wrapper = self.executor_server.execution_wrapper
+        else:
+            wrapper = self.executor_server.connections.drivers['nullwrap']
+
+        context = wrapper.getExecutionContext(ro_paths, rw_paths, secrets)
 
         popen = context.getPopen(
             work_dir=self.jobdir.work_root,
@@ -1583,6 +1303,7 @@
             self.proc = popen(
                 cmd,
                 cwd=self.jobdir.work_root,
+                stdin=subprocess.DEVNULL,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.STDOUT,
                 preexec_fn=os.setsid,
@@ -1630,6 +1351,13 @@
         elif ret == -9:
             # Received abort request.
             return (self.RESULT_ABORTED, None)
+        elif ret == 1:
+            if syntax_buffer[0].startswith(b'ERROR!'):
+                with open(self.jobdir.job_output_file, 'a') as job_output:
+                    for line in syntax_buffer:
+                        job_output.write("{now} | {line}\n".format(
+                            now=datetime.datetime.now(),
+                            line=line.decode('utf-8').rstrip()))
         elif ret == 4:
             # Ansible could not parse the yaml.
             self.log.debug("Ansible parse error")
@@ -1652,7 +1380,7 @@
                     now=datetime.datetime.now()))
                 found_marker = False
                 for line in syntax_buffer:
-                    if line.startswith('ERROR! Unexpected Exception'):
+                    if line.startswith(b'ERROR! Unexpected Exception'):
                         found_marker = True
                     if not found_marker:
                         continue
@@ -1672,11 +1400,63 @@
                '-a', 'gather_subset=!all']
 
         result, code = self.runAnsible(
-            cmd=cmd, timeout=60, playbook=playbook)
+            cmd=cmd, timeout=60, playbook=playbook,
+            wrapped=False)
         self.log.debug("Ansible complete, result %s code %s" % (
             self.RESULT_MAP[result], code))
         return result, code
 
+    def runAnsibleCleanup(self, playbook):
+        # TODO(jeblair): This requires a bugfix in Ansible 2.4
+        # Once this is used, increase the controlpersist timeout.
+        return (self.RESULT_NORMAL, 0)
+
+        if self.executor_server.verbose:
+            verbose = '-vvv'
+        else:
+            verbose = '-v'
+
+        cmd = ['ansible', '*', verbose, '-m', 'meta',
+               '-a', 'reset_connection']
+
+        result, code = self.runAnsible(
+            cmd=cmd, timeout=60, playbook=playbook,
+            wrapped=False)
+        self.log.debug("Ansible complete, result %s code %s" % (
+            self.RESULT_MAP[result], code))
+        return result, code
+
+    def emitPlaybookBanner(self, playbook, step, phase, result=None):
+        # This is used to print a header and a footer, respectively at the
+        # beginning and the end of each playbook execution.
+        # We are doing it from the executor rather than from a callback because
+        # the parameters are not made available to the callback until it's too
+        # late.
+        phase = phase or ''
+        trusted = playbook.trusted
+        trusted = 'trusted' if trusted else 'untrusted'
+        branch = playbook.branch
+        playbook = playbook.canonical_name_and_path
+
+        if phase and phase != 'run':
+            phase = '{phase}-run'.format(phase=phase)
+        phase = phase.upper()
+
+        if result is not None:
+            result = self.RESULT_MAP[result]
+            msg = "{phase} {step} {result}: [{trusted} : {playbook}@{branch}]"
+            msg = msg.format(phase=phase, step=step, result=result,
+                             trusted=trusted, playbook=playbook, branch=branch)
+        else:
+            msg = "{phase} {step}: [{trusted} : {playbook}@{branch}]"
+            msg = msg.format(phase=phase, step=step, trusted=trusted,
+                             playbook=playbook, branch=branch)
+
+        with open(self.jobdir.job_output_file, 'a') as job_output:
+            job_output.write("{now} | {msg}\n".format(
+                now=datetime.datetime.now(),
+                msg=msg))
+
     def runAnsiblePlaybook(self, playbook, timeout, success=None,
                            phase=None, index=None):
         if self.executor_server.verbose:
@@ -1707,8 +1487,482 @@
         if self.executor_variables_file is not None:
             cmd.extend(['-e@%s' % self.executor_variables_file])
 
+        self.emitPlaybookBanner(playbook, 'START', phase)
+
         result, code = self.runAnsible(
             cmd=cmd, timeout=timeout, playbook=playbook)
         self.log.debug("Ansible complete, result %s code %s" % (
             self.RESULT_MAP[result], code))
+
+        self.emitPlaybookBanner(playbook, 'END', phase, result=result)
         return result, code
+
+
+class ExecutorMergeWorker(gear.TextWorker):
+    def __init__(self, executor_server, *args, **kw):
+        self.zuul_executor_server = executor_server
+        super(ExecutorMergeWorker, self).__init__(*args, **kw)
+
+    def handleNoop(self, packet):
+        # Wait until the update queue is empty before responding
+        while self.zuul_executor_server.update_queue.qsize():
+            time.sleep(1)
+
+        with self.zuul_executor_server.merger_lock:
+            super(ExecutorMergeWorker, self).handleNoop(packet)
+
+
+class ExecutorExecuteWorker(gear.TextWorker):
+    def __init__(self, executor_server, *args, **kw):
+        self.zuul_executor_server = executor_server
+        super(ExecutorExecuteWorker, self).__init__(*args, **kw)
+
+    def handleNoop(self, packet):
+        # Delay our response to running a new job based on the number
+        # of jobs we're currently running, in an attempt to spread
+        # load evenly among executors.
+        workers = len(self.zuul_executor_server.job_workers)
+        delay = (workers ** 2) / 1000.0
+        time.sleep(delay)
+        return super(ExecutorExecuteWorker, self).handleNoop(packet)
+
+
+class ExecutorServer(object):
+    log = logging.getLogger("zuul.ExecutorServer")
+    _job_class = AnsibleJob
+
+    def __init__(self, config, connections={}, jobdir_root=None,
+                 keep_jobdir=False, log_streaming_port=DEFAULT_FINGER_PORT):
+        self.config = config
+        self.keep_jobdir = keep_jobdir
+        self.jobdir_root = jobdir_root
+        # TODOv3(mordred): make the executor name more unique --
+        # perhaps hostname+pid.
+        self.hostname = get_default(self.config, 'executor', 'hostname',
+                                    socket.gethostname())
+        self.log_streaming_port = log_streaming_port
+        self.merger_lock = threading.Lock()
+        self.run_lock = threading.Lock()
+        self.verbose = False
+        self.command_map = dict(
+            stop=self.stop,
+            pause=self.pause,
+            unpause=self.unpause,
+            graceful=self.graceful,
+            verbose=self.verboseOn,
+            unverbose=self.verboseOff,
+            keep=self.keep,
+            nokeep=self.nokeep,
+        )
+
+        self.statsd = get_statsd(config)
+        self.merge_root = get_default(self.config, 'executor', 'git_dir',
+                                      '/var/lib/zuul/executor-git')
+        self.default_username = get_default(self.config, 'executor',
+                                            'default_username', 'zuul')
+        self.disk_limit_per_job = int(get_default(self.config, 'executor',
+                                                  'disk_limit_per_job', 250))
+        self.merge_email = get_default(self.config, 'merger', 'git_user_email')
+        self.merge_name = get_default(self.config, 'merger', 'git_user_name')
+        self.merge_speed_limit = get_default(
+            config, 'merger', 'git_http_low_speed_limit', '1000')
+        self.merge_speed_time = get_default(
+            config, 'merger', 'git_http_low_speed_time', '30')
+        execution_wrapper_name = get_default(self.config, 'executor',
+                                             'execution_wrapper', 'bubblewrap')
+        load_multiplier = float(get_default(self.config, 'executor',
+                                            'load_multiplier', '2.5'))
+        self.max_load_avg = multiprocessing.cpu_count() * load_multiplier
+        self.accepting_work = False
+        self.execution_wrapper = connections.drivers[execution_wrapper_name]
+
+        self.connections = connections
+        # This merger and its git repos are used to maintain
+        # up-to-date copies of all the repos that are used by jobs, as
+        # well as to support the merger:cat functon to supply
+        # configuration information to Zuul when it starts.
+        self.merger = self._getMerger(self.merge_root)
+        self.update_queue = DeduplicateQueue()
+
+        state_dir = get_default(self.config, 'executor', 'state_dir',
+                                '/var/lib/zuul', expand_user=True)
+        path = os.path.join(state_dir, 'executor.socket')
+        self.command_socket = commandsocket.CommandSocket(path)
+        ansible_dir = os.path.join(state_dir, 'ansible')
+        self.ansible_dir = ansible_dir
+        if os.path.exists(ansible_dir):
+            shutil.rmtree(ansible_dir)
+
+        zuul_dir = os.path.join(ansible_dir, 'zuul')
+        plugin_dir = os.path.join(zuul_dir, 'ansible')
+
+        os.makedirs(plugin_dir, mode=0o0755)
+
+        self.library_dir = os.path.join(plugin_dir, 'library')
+        self.action_dir = os.path.join(plugin_dir, 'action')
+        self.callback_dir = os.path.join(plugin_dir, 'callback')
+        self.lookup_dir = os.path.join(plugin_dir, 'lookup')
+        self.filter_dir = os.path.join(plugin_dir, 'filter')
+
+        _copy_ansible_files(zuul.ansible, plugin_dir)
+
+        # We're copying zuul.ansible.* into a directory we are going
+        # to add to pythonpath, so our plugins can "import
+        # zuul.ansible".  But we're not installing all of zuul, so
+        # create a __init__.py file for the stub "zuul" module.
+        with open(os.path.join(zuul_dir, '__init__.py'), 'w'):
+            pass
+
+        self.job_workers = {}
+        self.disk_accountant = DiskAccountant(self.jobdir_root,
+                                              self.disk_limit_per_job,
+                                              self.stopJobDiskFull,
+                                              self.merge_root)
+
+    def _getMerger(self, root, logger=None):
+        if root != self.merge_root:
+            cache_root = self.merge_root
+        else:
+            cache_root = None
+        return zuul.merger.merger.Merger(
+            root, self.connections, self.merge_email, self.merge_name,
+            self.merge_speed_limit, self.merge_speed_time, cache_root, logger)
+
+    def start(self):
+        self._running = True
+        self._command_running = True
+        server = self.config.get('gearman', 'server')
+        port = get_default(self.config, 'gearman', 'port', 4730)
+        ssl_key = get_default(self.config, 'gearman', 'ssl_key')
+        ssl_cert = get_default(self.config, 'gearman', 'ssl_cert')
+        ssl_ca = get_default(self.config, 'gearman', 'ssl_ca')
+        self.merger_worker = ExecutorMergeWorker(self, 'Zuul Executor Merger')
+        self.merger_worker.addServer(server, port, ssl_key, ssl_cert, ssl_ca)
+        self.executor_worker = ExecutorExecuteWorker(
+            self, 'Zuul Executor Server')
+        self.executor_worker.addServer(server, port, ssl_key, ssl_cert, ssl_ca)
+        self.log.debug("Waiting for server")
+        self.merger_worker.waitForServer()
+        self.executor_worker.waitForServer()
+        self.log.debug("Registering")
+        self.register()
+
+        self.log.debug("Starting command processor")
+        self.command_socket.start()
+        self.command_thread = threading.Thread(target=self.runCommand,
+                                               name='command')
+        self.command_thread.daemon = True
+        self.command_thread.start()
+
+        self.log.debug("Starting worker")
+        self.update_thread = threading.Thread(target=self._updateLoop,
+                                              name='update')
+        self.update_thread.daemon = True
+        self.update_thread.start()
+        self.merger_thread = threading.Thread(target=self.run_merger,
+                                              name='merger')
+        self.merger_thread.daemon = True
+        self.merger_thread.start()
+        self.executor_thread = threading.Thread(target=self.run_executor,
+                                                name='executor')
+        self.executor_thread.daemon = True
+        self.executor_thread.start()
+        self.governor_stop_event = threading.Event()
+        self.governor_thread = threading.Thread(target=self.run_governor,
+                                                name='governor')
+        self.governor_thread.daemon = True
+        self.governor_thread.start()
+        self.disk_accountant.start()
+
+    def register(self):
+        self.register_work()
+        self.executor_worker.registerFunction("executor:stop:%s" %
+                                              self.hostname)
+        self.merger_worker.registerFunction("merger:merge")
+        self.merger_worker.registerFunction("merger:cat")
+        self.merger_worker.registerFunction("merger:refstate")
+
+    def register_work(self):
+        if self._running:
+            self.accepting_work = True
+            self.executor_worker.registerFunction("executor:execute")
+
+    def unregister_work(self):
+        self.accepting_work = False
+        self.executor_worker.unregisterFunction("executor:execute")
+
+    def stop(self):
+        self.log.debug("Stopping")
+        self.disk_accountant.stop()
+        # The governor can change function registration, so make sure
+        # it has stopped.
+        self.governor_stop_event.set()
+        self.governor_thread.join()
+        # Stop accepting new jobs
+        self.merger_worker.setFunctions([])
+        self.executor_worker.setFunctions([])
+        # Tell the executor worker to abort any jobs it just accepted,
+        # and grab the list of currently running job workers.
+        with self.run_lock:
+            self._running = False
+            self._command_running = False
+            workers = list(self.job_workers.values())
+        self.command_socket.stop()
+
+        for job_worker in workers:
+            try:
+                job_worker.stop()
+            except Exception:
+                self.log.exception("Exception sending stop command "
+                                   "to worker:")
+        for job_worker in workers:
+            try:
+                job_worker.wait()
+            except Exception:
+                self.log.exception("Exception waiting for worker "
+                                   "to stop:")
+
+        # Now that we aren't accepting any new jobs, and all of the
+        # running jobs have stopped, tell the update processor to
+        # stop.
+        self.update_queue.put(None)
+
+        # All job results should have been sent by now, shutdown the
+        # gearman workers.
+        self.merger_worker.shutdown()
+        self.executor_worker.shutdown()
+
+        if self.statsd:
+            base_key = 'zuul.executor.%s' % self.hostname
+            self.statsd.gauge(base_key + '.load_average', 0)
+            self.statsd.gauge(base_key + '.running_builds', 0)
+
+        self.log.debug("Stopped")
+
+    def join(self):
+        self.governor_thread.join()
+        self.update_thread.join()
+        self.merger_thread.join()
+        self.executor_thread.join()
+
+    def pause(self):
+        # TODOv3: implement
+        pass
+
+    def unpause(self):
+        # TODOv3: implement
+        pass
+
+    def graceful(self):
+        # TODOv3: implement
+        pass
+
+    def verboseOn(self):
+        self.verbose = True
+
+    def verboseOff(self):
+        self.verbose = False
+
+    def keep(self):
+        self.keep_jobdir = True
+
+    def nokeep(self):
+        self.keep_jobdir = False
+
+    def runCommand(self):
+        while self._command_running:
+            try:
+                command = self.command_socket.get().decode('utf8')
+                if command != '_stop':
+                    self.command_map[command]()
+            except Exception:
+                self.log.exception("Exception while processing command")
+
+    def _updateLoop(self):
+        while True:
+            try:
+                self._innerUpdateLoop()
+            except StopException:
+                return
+            except Exception:
+                self.log.exception("Exception in update thread:")
+
+    def _innerUpdateLoop(self):
+        # Inside of a loop that keeps the main repositories up to date
+        task = self.update_queue.get()
+        if task is None:
+            # We are asked to stop
+            raise StopException()
+        with self.merger_lock:
+            self.log.info("Updating repo %s/%s" % (
+                task.connection_name, task.project_name))
+            self.merger.updateRepo(task.connection_name, task.project_name)
+            self.log.debug("Finished updating repo %s/%s" %
+                           (task.connection_name, task.project_name))
+        task.setComplete()
+
+    def update(self, connection_name, project_name):
+        # Update a repository in the main merger
+        task = UpdateTask(connection_name, project_name)
+        task = self.update_queue.put(task)
+        return task
+
+    def run_merger(self):
+        self.log.debug("Starting merger listener")
+        while self._running:
+            try:
+                job = self.merger_worker.getJob()
+                try:
+                    self.mergerJobDispatch(job)
+                except Exception:
+                    self.log.exception("Exception while running job")
+                    job.sendWorkException(
+                        traceback.format_exc().encode('utf8'))
+            except gear.InterruptedError:
+                pass
+            except Exception:
+                self.log.exception("Exception while getting job")
+
+    def mergerJobDispatch(self, job):
+        with self.run_lock:
+            if job.name == 'merger:cat':
+                self.log.debug("Got cat job: %s" % job.unique)
+                self.cat(job)
+            elif job.name == 'merger:merge':
+                self.log.debug("Got merge job: %s" % job.unique)
+                self.merge(job)
+            elif job.name == 'merger:refstate':
+                self.log.debug("Got refstate job: %s" % job.unique)
+                self.refstate(job)
+            else:
+                self.log.error("Unable to handle job %s" % job.name)
+                job.sendWorkFail()
+
+    def run_executor(self):
+        self.log.debug("Starting executor listener")
+        while self._running:
+            try:
+                job = self.executor_worker.getJob()
+                try:
+                    self.executorJobDispatch(job)
+                except Exception:
+                    self.log.exception("Exception while running job")
+                    job.sendWorkException(
+                        traceback.format_exc().encode('utf8'))
+            except gear.InterruptedError:
+                pass
+            except Exception:
+                self.log.exception("Exception while getting job")
+
+    def executorJobDispatch(self, job):
+        with self.run_lock:
+            if not self._running:
+                job.sendWorkFail()
+                return
+            if job.name == 'executor:execute':
+                self.log.debug("Got execute job: %s" % job.unique)
+                self.executeJob(job)
+            elif job.name.startswith('executor:stop'):
+                self.log.debug("Got stop job: %s" % job.unique)
+                self.stopJob(job)
+            else:
+                self.log.error("Unable to handle job %s" % job.name)
+                job.sendWorkFail()
+
+    def executeJob(self, job):
+        if self.statsd:
+            base_key = 'zuul.executor.%s' % self.hostname
+            self.statsd.incr(base_key + '.builds')
+        self.job_workers[job.unique] = self._job_class(self, job)
+        self.job_workers[job.unique].run()
+
+    def run_governor(self):
+        while not self.governor_stop_event.wait(30):
+            try:
+                self.manageLoad()
+            except Exception:
+                self.log.exception("Exception in governor thread:")
+
+    def manageLoad(self):
+        ''' Apply some heuristics to decide whether or not we should
+            be askign for more jobs '''
+        load_avg = os.getloadavg()[0]
+        if self.accepting_work:
+            # Don't unregister if we don't have any active jobs.
+            if load_avg > self.max_load_avg and self.job_workers:
+                self.log.info(
+                    "Unregistering due to high system load {} > {}".format(
+                        load_avg, self.max_load_avg))
+                self.unregister_work()
+        elif load_avg <= self.max_load_avg:
+            self.log.info(
+                "Re-registering as load is within limits {} <= {}".format(
+                    load_avg, self.max_load_avg))
+            self.register_work()
+        if self.statsd:
+            base_key = 'zuul.executor.%s' % self.hostname
+            self.statsd.gauge(base_key + '.load_average',
+                              int(load_avg * 100))
+            self.statsd.gauge(base_key + '.running_builds',
+                              len(self.job_workers))
+
+    def finishJob(self, unique):
+        del(self.job_workers[unique])
+
+    def stopJobDiskFull(self, jobdir):
+        unique = os.path.basename(jobdir)
+        self.stopJobByUnique(unique, reason=AnsibleJob.RESULT_DISK_FULL)
+
+    def stopJob(self, job):
+        try:
+            args = json.loads(job.arguments)
+            self.log.debug("Stop job with arguments: %s" % (args,))
+            unique = args['uuid']
+            self.stopJobByUnique(unique)
+        finally:
+            job.sendWorkComplete()
+
+    def stopJobByUnique(self, unique, reason=None):
+        job_worker = self.job_workers.get(unique)
+        if not job_worker:
+            self.log.debug("Unable to find worker for job %s" % (unique,))
+            return
+        try:
+            job_worker.stop(reason)
+        except Exception:
+            self.log.exception("Exception sending stop command "
+                               "to worker:")
+
+    def cat(self, job):
+        args = json.loads(job.arguments)
+        task = self.update(args['connection'], args['project'])
+        task.wait()
+        with self.merger_lock:
+            files = self.merger.getFiles(args['connection'], args['project'],
+                                         args['branch'], args['files'],
+                                         args.get('dirs', []))
+        result = dict(updated=True,
+                      files=files)
+        job.sendWorkComplete(json.dumps(result))
+
+    def refstate(self, job):
+        args = json.loads(job.arguments)
+        with self.merger_lock:
+            success, repo_state = self.merger.getRepoState(args['items'])
+        result = dict(updated=success,
+                      repo_state=repo_state)
+        job.sendWorkComplete(json.dumps(result))
+
+    def merge(self, job):
+        args = json.loads(job.arguments)
+        with self.merger_lock:
+            ret = self.merger.mergeChanges(args['items'], args.get('files'),
+                                           args.get('dirs', []),
+                                           args.get('repo_state'))
+        result = dict(merged=(ret is not None))
+        if ret is None:
+            result['commit'] = result['files'] = result['repo_state'] = None
+        else:
+            (result['commit'], result['files'], result['repo_state'],
+             recent) = ret
+        job.sendWorkComplete(json.dumps(result))
diff --git a/zuul/lib/queue.py b/zuul/lib/queue.py
new file mode 100644
index 0000000..db8af47
--- /dev/null
+++ b/zuul/lib/queue.py
@@ -0,0 +1,78 @@
+# Copyright 2014 OpenStack Foundation
+# Copyright 2017 Red Hat, Inc.
+#
+# 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 collections
+import threading
+
+
+class MergedQueue(object):
+    def __init__(self):
+        self.queue = collections.deque()
+        self.lock = threading.RLock()
+        self.condition = threading.Condition(self.lock)
+        self.join_condition = threading.Condition(self.lock)
+        self.tasks = 0
+
+    def qsize(self):
+        return len(self.queue)
+
+    def empty(self):
+        return self.qsize() == 0
+
+    def put(self, item):
+        # Returns the original item if added, or an updated equivalent
+        # item if already enqueued.
+        self.condition.acquire()
+        ret = None
+        try:
+            for x in self.queue:
+                if item == x:
+                    ret = x
+                    if hasattr(ret, 'merge'):
+                        ret.merge(item)
+            if ret is None:
+                ret = item
+                self.queue.append(item)
+                self.condition.notify()
+        finally:
+            self.condition.release()
+        return ret
+
+    def get(self):
+        self.condition.acquire()
+        try:
+            while True:
+                try:
+                    ret = self.queue.popleft()
+                    self.join_condition.acquire()
+                    self.tasks += 1
+                    self.join_condition.release()
+                    return ret
+                except IndexError:
+                    self.condition.wait()
+        finally:
+            self.condition.release()
+
+    def task_done(self):
+        self.join_condition.acquire()
+        self.tasks -= 1
+        self.join_condition.notify()
+        self.join_condition.release()
+
+    def join(self):
+        self.join_condition.acquire()
+        while self.tasks:
+            self.join_condition.wait()
+        self.join_condition.release()
diff --git a/zuul/lib/statsd.py b/zuul/lib/statsd.py
new file mode 100644
index 0000000..0ccacf9
--- /dev/null
+++ b/zuul/lib/statsd.py
@@ -0,0 +1,30 @@
+# Copyright 2017 Red Hat, Inc.
+#
+# 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.
+
+from zuul.lib.config import get_default
+
+
+def get_statsd_config(config):
+    statsd_host = get_default(config, 'statsd', 'server')
+    statsd_port = int(get_default(config, 'statsd', 'port', 8125))
+    statsd_prefix = get_default(config, 'statsd', 'prefix')
+    return (statsd_host, statsd_port, statsd_prefix)
+
+
+def get_statsd(config):
+    (statsd_host, statsd_port, statsd_prefix) = get_statsd_config(config)
+    if statsd_host is None:
+        return None
+    import statsd
+    return statsd.StatsClient(statsd_host, statsd_port, statsd_prefix)
diff --git a/zuul/manager/__init__.py b/zuul/manager/__init__.py
index b94b8a5..edea69c 100644
--- a/zuul/manager/__init__.py
+++ b/zuul/manager/__init__.py
@@ -11,6 +11,7 @@
 # under the License.
 
 import logging
+import textwrap
 
 from zuul import exceptions
 from zuul import model
@@ -42,9 +43,10 @@
 class PipelineManager(object):
     """Abstract Base Class for enqueing and processing Changes in a Pipeline"""
 
-    log = logging.getLogger("zuul.PipelineManager")
-
     def __init__(self, sched, pipeline):
+        self.log = logging.getLogger("zuul.Pipeline.%s.%s" %
+                                     (pipeline.layout.tenant.name,
+                                      pipeline.name,))
         self.sched = sched
         self.pipeline = pipeline
         self.event_filters = []
@@ -228,20 +230,17 @@
                                (item.change, change_queue))
                 change_queue.enqueueItem(item)
 
-                # Get an updated copy of the layout if necessary.
-                # This will return one of the following:
-                # 1) An existing layout from the item ahead or pipeline.
-                # 2) A newly created layout from the cached pipeline
-                #    layout config plus the previously returned
-                #    in-repo files stored in the buildset.
-                # 3) None in the case that a fetch of the files from
-                #    the merger is still pending.
-                item.current_build_set.layout = self.getLayout(item)
-
-                # Rebuild the frozen job tree from the new layout, if
-                # we have one.  If not, it will be built later.
-                if item.current_build_set.layout:
-                    item.freezeJobGraph()
+                # Get an updated copy of the layout and update the job
+                # graph if necessary.  This resumes the buildset merge
+                # state machine.  If we have an up-to-date layout, it
+                # will go ahead and refresh the job graph if needed;
+                # or it will send a new merge job if necessary, or it
+                # will do nothing if we're waiting on a merge job.
+                item.job_graph = None
+                item.layout = None
+                if item.active:
+                    if self.prepareItem(item):
+                        self.prepareJobs(item)
 
                 # Re-set build results in case any new jobs have been
                 # added to the tree.
@@ -373,7 +372,7 @@
     def executeJobs(self, item):
         # TODO(jeblair): This should return a value indicating a job
         # was executed.  Appears to be a longstanding bug.
-        if not item.current_build_set.layout:
+        if not item.layout:
             return False
 
         jobs = item.findJobsToRun(
@@ -407,13 +406,8 @@
                     old_build_set.item, build.job)
 
             if not was_running:
-                try:
-                    nodeset = build.build_set.getJobNodeSet(build.job.name)
-                    self.sched.nodepool.returnNodeSet(nodeset)
-                except Exception:
-                    self.log.exception("Unable to return nodeset %s for "
-                                       "canceled build request %s" %
-                                       (nodeset, build))
+                nodeset = build.build_set.getJobNodeSet(build.job.name)
+                self.sched.nodepool.returnNodeSet(nodeset)
             build.result = 'CANCELED'
             canceled = True
             canceled_jobs.add(build.job.name)
@@ -434,30 +428,60 @@
         import zuul.configloader
         loader = zuul.configloader.ConfigLoader()
 
-        build_set = item.current_build_set
         self.log.debug("Loading dynamic layout")
+        (trusted_updates, untrusted_updates) = item.includesConfigUpdates()
+        build_set = item.current_build_set
+        trusted_layout_verified = False
         try:
             # First parse the config as it will land with the
             # full set of config and project repos.  This lets us
             # catch syntax errors in config repos even though we won't
             # actually run with that config.
-            loader.createDynamicLayout(
-                item.pipeline.layout.tenant,
-                build_set.files,
-                include_config_projects=True,
-                scheduler=self.sched,
-                connections=self.sched.connections)
+            if trusted_updates:
+                self.log.debug("Loading dynamic layout (phase 1)")
+                loader.createDynamicLayout(
+                    item.pipeline.layout.tenant,
+                    build_set.files,
+                    include_config_projects=True,
+                    scheduler=self.sched,
+                    connections=self.sched.connections)
+                trusted_layout_verified = True
 
             # Then create the config a second time but without changes
             # to config repos so that we actually use this config.
-            layout = loader.createDynamicLayout(
-                item.pipeline.layout.tenant,
-                build_set.files,
-                include_config_projects=False)
+            if untrusted_updates:
+                self.log.debug("Loading dynamic layout (phase 2)")
+                layout = loader.createDynamicLayout(
+                    item.pipeline.layout.tenant,
+                    build_set.files,
+                    include_config_projects=False)
+            else:
+                # We're a change to a config repo (with no untrusted
+                # items ahead), so just use the most recently
+                # generated layout.
+                if item.item_ahead:
+                    return item.item_ahead.layout
+                else:
+                    return item.queue.pipeline.layout
+            self.log.debug("Loading dynamic layout complete")
         except zuul.configloader.ConfigurationSyntaxError as e:
-            self.log.info("Configuration syntax error "
-                          "in dynamic layout")
-            item.setConfigError(str(e))
+            self.log.info("Configuration syntax error in dynamic layout")
+            if trusted_layout_verified:
+                # The config is good if we include config-projects,
+                # but is currently invalid if we omit them.  Instead
+                # of returning the whole error message, just leave a
+                # note that the config will work once the dependent
+                # changes land.
+                msg = "This change depends on a change "\
+                      "to a config project.\n\n"
+                msg += textwrap.fill(textwrap.dedent("""\
+                The syntax of the configuration in this change has
+                been verified to be correct once the config project
+                change upon which it depends is merged, but it can not
+                be used until that occurs."""))
+                item.setConfigError(msg)
+            else:
+                item.setConfigError(str(e))
             return None
         except Exception:
             self.log.exception("Error in dynamic layout")
@@ -468,7 +492,7 @@
     def getLayout(self, item):
         if not item.change.updatesConfig():
             if item.item_ahead:
-                return item.item_ahead.current_build_set.layout
+                return item.item_ahead.layout
             else:
                 return item.queue.pipeline.layout
         # This item updates the config, ask the merger for the result.
@@ -519,10 +543,9 @@
     def prepareJobs(self, item):
         # This only runs once the item is in the pipeline's action window
         # Returns True if the item is ready, false otherwise
-        build_set = item.current_build_set
-        if not build_set.layout:
-            build_set.layout = self.getLayout(item)
-        if not build_set.layout:
+        if not item.layout:
+            item.layout = self.getLayout(item)
+        if not item.layout:
             return False
 
         if not item.job_graph:
@@ -748,8 +771,7 @@
         # pipeline, use the dynamic layout if available, otherwise,
         # fall back to the current static layout as a best
         # approximation.
-        layout = (item.current_build_set.layout or
-                  self.pipeline.layout)
+        layout = (item.layout or self.pipeline.layout)
 
         project_in_pipeline = True
         if not layout.getProjectPipelineConfig(item.change.project,
@@ -815,19 +837,28 @@
                 dt = None
             items = len(self.pipeline.getAllItems())
 
-            # stats.timers.zuul.pipeline.NAME.resident_time
-            # stats_counts.zuul.pipeline.NAME.total_changes
-            # stats.gauges.zuul.pipeline.NAME.current_changes
-            key = 'zuul.pipeline.%s' % self.pipeline.name
+            tenant = self.pipeline.layout.tenant
+            basekey = 'zuul.tenant.%s' % tenant.name
+            key = '%s.pipeline.%s' % (basekey, self.pipeline.name)
+            # stats.timers.zuul.tenant.<tenant>.pipeline.<pipeline>.resident_time
+            # stats_counts.zuul.tenant.<tenant>.pipeline.<pipeline>.total_changes
+            # stats.gauges.zuul.tenant.<tenant>.pipeline.<pipeline>.current_changes
             self.sched.statsd.gauge(key + '.current_changes', items)
             if dt:
                 self.sched.statsd.timing(key + '.resident_time', dt)
                 self.sched.statsd.incr(key + '.total_changes')
 
-            # stats.timers.zuul.pipeline.NAME.ORG.PROJECT.resident_time
-            # stats_counts.zuul.pipeline.NAME.ORG.PROJECT.total_changes
-            project_name = item.change.project.name.replace('/', '.')
-            key += '.%s' % project_name
+            hostname = (item.change.project.canonical_hostname.
+                        replace('.', '_'))
+            projectname = (item.change.project.name.
+                           replace('.', '_').replace('/', '.'))
+            projectname = projectname.replace('.', '_').replace('/', '.')
+            branchname = item.change.branch.replace('.', '_').replace('/', '.')
+            # stats.timers.zuul.tenant.<tenant>.pipeline.<pipeline>.
+            #   project.<host>.<project>.<branch>.resident_time
+            # stats_counts.zuul.tenant.<tenant>.pipeline.<pipeline>.
+            #   project.<host>.<project>.<branch>.total_changes
+            key += '.project.%s.%s.%s' % (hostname, projectname, branchname)
             if dt:
                 self.sched.statsd.timing(key + '.resident_time', dt)
                 self.sched.statsd.incr(key + '.total_changes')
diff --git a/zuul/manager/dependent.py b/zuul/manager/dependent.py
index 411894e..5aef453 100644
--- a/zuul/manager/dependent.py
+++ b/zuul/manager/dependent.py
@@ -10,8 +10,6 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-import logging
-
 from zuul import model
 from zuul.manager import PipelineManager, StaticChangeQueueContextManager
 from zuul.manager import DynamicChangeQueueContextManager
@@ -25,7 +23,6 @@
     using the Optmistic Branch Prediction logic with Nearest Non-Failing Item
     reparenting algorithm for handling errors.
     """
-    log = logging.getLogger("zuul.DependentPipelineManager")
     changes_merge = True
 
     def __init__(self, *args, **kwargs):
diff --git a/zuul/manager/independent.py b/zuul/manager/independent.py
index 7b0a9f5..65f5ca0 100644
--- a/zuul/manager/independent.py
+++ b/zuul/manager/independent.py
@@ -10,8 +10,6 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-import logging
-
 from zuul import model
 from zuul.manager import PipelineManager, DynamicChangeQueueContextManager
 
@@ -19,7 +17,6 @@
 class IndependentPipelineManager(PipelineManager):
     """PipelineManager that puts every Change into its own ChangeQueue."""
 
-    log = logging.getLogger("zuul.IndependentPipelineManager")
     changes_merge = False
 
     def _postConfig(self, layout):
diff --git a/zuul/merger/merger.py b/zuul/merger/merger.py
index 8b98bfb..035d1d0 100644
--- a/zuul/merger/merger.py
+++ b/zuul/merger/merger.py
@@ -13,10 +13,13 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from contextlib import contextmanager
+import logging
+import os
+import shutil
+
 import git
 import gitdb
-import os
-import logging
 
 import zuul.model
 
@@ -38,22 +41,37 @@
             raise
 
 
+@contextmanager
+def timeout_handler(path):
+    try:
+        yield
+    except git.exc.GitCommandError as e:
+        if e.status == -9:
+            # Timeout.  The repo could be in a bad state, so delete it.
+            shutil.rmtree(path)
+        raise
+
+
 class ZuulReference(git.Reference):
     _common_path_default = "refs/zuul"
     _points_to_commits_only = True
 
 
 class Repo(object):
-    def __init__(self, remote, local, email, username, sshkey=None,
-                 cache_path=None, logger=None):
+    def __init__(self, remote, local, email, username, speed_limit, speed_time,
+                 sshkey=None, cache_path=None, logger=None, git_timeout=300):
         if logger is None:
             self.log = logging.getLogger("zuul.Repo")
         else:
             self.log = logger
+        self.env = {
+            'GIT_HTTP_LOW_SPEED_LIMIT': speed_limit,
+            'GIT_HTTP_LOW_SPEED_TIME': speed_time,
+        }
+        self.git_timeout = git_timeout
         if sshkey:
-            self.env = {'GIT_SSH_COMMAND': 'ssh -i %s' % (sshkey,)}
-        else:
-            self.env = {}
+            self.env['GIT_SSH_COMMAND'] = 'ssh -i %s' % (sshkey,)
+
         self.remote_url = remote
         self.local_path = local
         self.email = email
@@ -62,7 +80,7 @@
         self._initialized = False
         try:
             self._ensure_cloned()
-        except:
+        except Exception:
             self.log.exception("Unable to initialize repo for %s" % remote)
 
     def _ensure_cloned(self):
@@ -75,12 +93,10 @@
             self.log.debug("Cloning from %s to %s" % (self.remote_url,
                                                       self.local_path))
             if self.cache_path:
-                git.Repo.clone_from(self.cache_path, self.local_path,
-                                    env=self.env)
+                self._git_clone(self.cache_path)
                 rewrite_url = True
             else:
-                git.Repo.clone_from(self.remote_url, self.local_path,
-                                    env=self.env)
+                self._git_clone(self.remote_url)
         repo = git.Repo(self.local_path)
         repo.git.update_environment(**self.env)
         # Create local branches corresponding to all the remote branches
@@ -104,6 +120,18 @@
     def isInitialized(self):
         return self._initialized
 
+    def _git_clone(self, url):
+        mygit = git.cmd.Git(os.getcwd())
+        mygit.update_environment(**self.env)
+        with timeout_handler(self.local_path):
+            mygit.clone(git.cmd.Git.polish_url(url), self.local_path,
+                        kill_after_timeout=self.git_timeout)
+
+    def _git_fetch(self, repo, remote, ref=None, **kwargs):
+        with timeout_handler(self.local_path):
+            repo.git.fetch(remote, ref, kill_after_timeout=self.git_timeout,
+                           **kwargs)
+
     def createRepoObject(self):
         self._ensure_cloned()
         repo = git.Repo(self.local_path)
@@ -225,19 +253,18 @@
 
     def fetch(self, ref):
         repo = self.createRepoObject()
-        # The git.remote.fetch method may read in git progress info and
-        # interpret it improperly causing an AssertionError. Because the
-        # data was fetched properly subsequent fetches don't seem to fail.
-        # So try again if an AssertionError is caught.
-        origin = repo.remotes.origin
-        try:
-            origin.fetch(ref)
-        except AssertionError:
-            origin.fetch(ref)
+        # NOTE: The following is currently not applicable, but if we
+        # switch back to fetch methods from GitPython, we need to
+        # consider it:
+        #   The git.remote.fetch method may read in git progress info and
+        #   interpret it improperly causing an AssertionError. Because the
+        #   data was fetched properly subsequent fetches don't seem to fail.
+        #   So try again if an AssertionError is caught.
+        self._git_fetch(repo, 'origin', ref)
 
     def fetchFrom(self, repository, ref):
         repo = self.createRepoObject()
-        repo.git.fetch(repository, ref)
+        self._git_fetch(repo, repository, ref)
 
     def createZuulRef(self, ref, commit='HEAD'):
         repo = self.createRepoObject()
@@ -254,15 +281,14 @@
     def update(self):
         repo = self.createRepoObject()
         self.log.debug("Updating repository %s" % self.local_path)
-        origin = repo.remotes.origin
         if repo.git.version_info[:2] < (1, 9):
             # Before 1.9, 'git fetch --tags' did not include the
             # behavior covered by 'git --fetch', so we run both
             # commands in that case.  Starting with 1.9, 'git fetch
             # --tags' is all that is necessary.  See
             # https://github.com/git/git/blob/master/Documentation/RelNotes/1.9.0.txt#L18-L20
-            origin.fetch()
-        origin.fetch(tags=True)
+            self._git_fetch(repo, 'origin')
+        self._git_fetch(repo, 'origin', tags=True)
 
     def getFiles(self, files, dirs=[], branch=None, commit=None):
         ret = {}
@@ -293,7 +319,7 @@
 
 class Merger(object):
     def __init__(self, working_root, connections, email, username,
-                 cache_root=None, logger=None):
+                 speed_limit, speed_time, cache_root=None, logger=None):
         self.logger = logger
         if logger is None:
             self.log = logging.getLogger("zuul.Merger")
@@ -306,6 +332,8 @@
         self.connections = connections
         self.email = email
         self.username = username
+        self.speed_limit = speed_limit
+        self.speed_time = speed_time
         self.cache_root = cache_root
 
     def _addProject(self, hostname, project_name, url, sshkey):
@@ -318,8 +346,9 @@
                                           project_name)
             else:
                 cache_path = None
-            repo = Repo(url, path, self.email, self.username,
-                        sshkey, cache_path, self.logger)
+            repo = Repo(
+                url, path, self.email, self.username, self.speed_limit,
+                self.speed_time, sshkey, cache_path, self.logger)
 
             self.repos[key] = repo
         except Exception:
diff --git a/zuul/merger/server.py b/zuul/merger/server.py
index 881209d..765d9e0 100644
--- a/zuul/merger/server.py
+++ b/zuul/merger/server.py
@@ -33,9 +33,13 @@
                                  '/var/lib/zuul/merger-git')
         merge_email = get_default(self.config, 'merger', 'git_user_email')
         merge_name = get_default(self.config, 'merger', 'git_user_name')
-
-        self.merger = merger.Merger(merge_root, connections, merge_email,
-                                    merge_name)
+        speed_limit = get_default(
+            config, 'merger', 'git_http_low_speed_limit', '1000')
+        speed_time = get_default(
+            config, 'merger', 'git_http_low_speed_time', '30')
+        self.merger = merger.Merger(
+            merge_root, connections, merge_email, merge_name, speed_limit,
+            speed_time)
 
     def start(self):
         self._running = True
diff --git a/zuul/model.py b/zuul/model.py
index 4c5a51f..7d6e80c 100644
--- a/zuul/model.py
+++ b/zuul/model.py
@@ -23,6 +23,8 @@
 import urllib.parse
 import textwrap
 
+from zuul import change_matcher
+
 MERGER_MERGE = 1          # "git merge"
 MERGER_MERGE_RESOLVE = 2  # "git merge -s resolve"
 MERGER_CHERRY_PICK = 3    # "git cherry-pick"
@@ -44,6 +46,12 @@
     'high': PRECEDENCE_HIGH,
 }
 
+PRIORITY_MAP = {
+    PRECEDENCE_NORMAL: 200,
+    PRECEDENCE_LOW: 300,
+    PRECEDENCE_HIGH: 100,
+}
+
 # Request states
 STATE_REQUESTED = 'requested'
 STATE_PENDING = 'pending'
@@ -515,6 +523,7 @@
         self.job = job
         self.nodeset = nodeset
         self._state = STATE_REQUESTED
+        self.requested_time = time.time()
         self.state_time = time.time()
         self.stat = None
         self.uid = uuid4().hex
@@ -525,6 +534,14 @@
         self.canceled = False
 
     @property
+    def priority(self):
+        if self.build_set:
+            precedence = self.build_set.item.pipeline.precedence
+        else:
+            precedence = PRECEDENCE_NORMAL
+        return PRIORITY_MAP[precedence]
+
+    @property
     def fulfilled(self):
         return (self._state == STATE_FULFILLED) and not self.failed
 
@@ -817,6 +834,7 @@
         self.other_attributes = dict(
             name=None,
             source_context=None,
+            source_line=None,
             inheritance_path=(),
         )
 
@@ -851,14 +869,16 @@
         return self.name
 
     def __repr__(self):
-        return '<Job %s branches: %s source: %s>' % (self.name,
-                                                     self.branch_matcher,
-                                                     self.source_context)
+        return '<Job %s branches: %s source: %s#%s>' % (
+            self.name,
+            self.branch_matcher,
+            self.source_context,
+            self.source_line)
 
     def __getattr__(self, name):
         v = self.__dict__.get(name)
         if v is None:
-            return copy.deepcopy(self.attributes[name])
+            return self.attributes[name]
         return v
 
     def _get(self, name):
@@ -868,6 +888,8 @@
         return Attributes(name=self.name)
 
     def setRun(self):
+        msg = 'self %s' % (repr(self),)
+        self.inheritance_path = self.inheritance_path + (msg,)
         if not self.run:
             self.run = self.implied_run
 
@@ -898,14 +920,21 @@
         if changed:
             self.roles = tuple(newroles)
 
+    def setBranchMatcher(self, branches):
+        # Set the branch matcher to match any of the supplied branches
+        matchers = []
+        for branch in branches:
+            matchers.append(change_matcher.BranchMatcher(branch))
+        self.branch_matcher = change_matcher.MatchAny(matchers)
+
     def updateVariables(self, other_vars):
-        v = self.variables
+        v = copy.deepcopy(self.variables)
         Job._deepUpdate(v, other_vars)
         self.variables = v
 
     def updateProjects(self, other_projects):
-        required_projects = self.required_projects
-        Job._deepUpdate(required_projects, other_projects)
+        required_projects = self.required_projects.copy()
+        required_projects.update(other_projects)
         self.required_projects = required_projects
 
     @staticmethod
@@ -932,7 +961,7 @@
         # copy all attributes
         for k in self.inheritable_attributes:
             if (other._get(k) is not None):
-                setattr(self, k, copy.deepcopy(getattr(other, k)))
+                setattr(self, k, getattr(other, k))
 
         msg = 'inherit from %s' % (repr(other),)
         self.inheritance_path = other.inheritance_path + (msg,)
@@ -1025,12 +1054,14 @@
         else:
             self.jobs[job.name] = [job]
 
-    def inheritFrom(self, other):
+    def inheritFrom(self, other, implied_branch):
         for jobname, jobs in other.jobs.items():
-            if jobname in self.jobs:
-                self.jobs[jobname].extend(jobs)
-            else:
-                self.jobs[jobname] = jobs
+            joblist = self.jobs.setdefault(jobname, [])
+            for job in jobs:
+                if not job.branch_matcher and implied_branch:
+                    job = job.copy()
+                    job.setBranchMatcher([implied_branch])
+                joblist.append(job)
 
 
 class JobGraph(object):
@@ -1129,7 +1160,6 @@
         self.start_time = None
         self.end_time = None
         self.estimated_time = None
-        self.pipeline = None
         self.canceled = False
         self.retry = False
         self.parameters = {}
@@ -1141,6 +1171,10 @@
         return ('<Build %s of %s on %s>' %
                 (self.uuid, self.job.name, self.worker))
 
+    @property
+    def pipeline(self):
+        return self.build_set.item.pipeline
+
     def getSafeAttributes(self):
         return Attributes(uuid=self.uuid,
                           result=self.result,
@@ -1229,8 +1263,6 @@
         self.item = item
         self.builds = {}
         self.result = None
-        self.next_build_set = None
-        self.previous_build_set = None
         self.uuid = None
         self.commit = None
         self.dependent_items = None
@@ -1243,7 +1275,6 @@
         self.node_requests = {}  # job -> reqs
         self.files = RepoFiles()
         self.repo_state = {}
-        self.layout = None
         self.tries = {}
 
     @property
@@ -1338,7 +1369,7 @@
         item = self.item
         layout = None
         while item:
-            layout = item.current_build_set.layout
+            layout = item.layout
             if layout:
                 break
             item = item.item_ahead
@@ -1369,10 +1400,8 @@
         self.pipeline = queue.pipeline
         self.queue = queue
         self.change = change  # a ref
-        self.build_sets = []
         self.dequeued_needing_change = False
         self.current_build_set = BuildSet(self)
-        self.build_sets.append(self.current_build_set)
         self.item_ahead = None
         self.items_behind = []
         self.enqueue_time = None
@@ -1382,7 +1411,7 @@
         self.quiet = False
         self.active = False  # Whether an item is within an active window
         self.live = True  # Whether an item is intended to be processed at all
-        # TODO(jeblair): move job_graph to buildset
+        self.layout = None
         self.job_graph = None
 
     def __repr__(self):
@@ -1394,17 +1423,12 @@
             id(self), self.change, pipeline)
 
     def resetAllBuilds(self):
-        old = self.current_build_set
-        self.current_build_set.result = 'CANCELED'
         self.current_build_set = BuildSet(self)
-        old.next_build_set = self.current_build_set
-        self.current_build_set.previous_build_set = old
-        self.build_sets.append(self.current_build_set)
+        self.layout = None
         self.job_graph = None
 
     def addBuild(self, build):
         self.current_build_set.addBuild(build)
-        build.pipeline = self.pipeline
 
     def removeBuild(self, build):
         self.current_build_set.removeBuild(build)
@@ -1415,8 +1439,7 @@
     def freezeJobGraph(self):
         """Find or create actual matching jobs for this item's change and
         store the resulting job tree."""
-        layout = self.current_build_set.layout
-        job_graph = layout.createJobGraph(self)
+        job_graph = self.layout.createJobGraph(self)
         for job in job_graph.getJobs():
             # Ensure that each jobs's dependencies are fully
             # accessible.  This will raise an exception if not.
@@ -1491,6 +1514,25 @@
     def wasDequeuedNeedingChange(self):
         return self.dequeued_needing_change
 
+    def includesConfigUpdates(self):
+        includes_trusted = False
+        includes_untrusted = False
+        tenant = self.pipeline.layout.tenant
+        item = self
+        while item:
+            if item.change.updatesConfig():
+                (trusted, project) = tenant.getProject(
+                    item.change.project.canonical_name)
+                if trusted:
+                    includes_trusted = True
+                else:
+                    includes_untrusted = True
+            if includes_trusted and includes_untrusted:
+                # We're done early
+                return (includes_trusted, includes_untrusted)
+            item = item.item_ahead
+        return (includes_trusted, includes_untrusted)
+
     def isHoldingFollowingChanges(self):
         if not self.live:
             return False
@@ -2098,6 +2140,7 @@
     def __init__(self, name):
         self.name = name
         self.merge_mode = None
+        # The default branch for the project (usually master).
         self.default_branch = None
         self.pipelines = {}
         self.private_key_file = None
@@ -2277,6 +2320,7 @@
     """Holds all of the Pipelines."""
 
     def __init__(self, tenant):
+        self.uuid = uuid4().hex
         self.tenant = tenant
         self.project_configs = {}
         self.project_templates = {}
@@ -2498,14 +2542,14 @@
 
     @staticmethod
     def _max_count(item, semaphore_name):
-        if not item.current_build_set.layout:
+        if not item.layout:
             # This should not occur as the layout of the item must already be
             # built when acquiring or releasing a semaphore for a job.
             raise Exception("Item {} has no layout".format(item))
 
         # find the right semaphore
         default_semaphore = Semaphore(semaphore_name, 1)
-        semaphores = item.current_build_set.layout.semaphores
+        semaphores = item.layout.semaphores
         return semaphores.get(semaphore_name, default_semaphore).max
 
 
diff --git a/zuul/nodepool.py b/zuul/nodepool.py
index dc855cd..7dafca0 100644
--- a/zuul/nodepool.py
+++ b/zuul/nodepool.py
@@ -22,6 +22,35 @@
         self.requests = {}
         self.sched = scheduler
 
+    def emitStats(self, request):
+        if not self.sched.statsd:
+            return
+        statsd = self.sched.statsd
+        # counter zuul.nodepool.requested
+        # counter zuul.nodepool.requested.label.<label>
+        # counter zuul.nodepool.requested.size.<size>
+        # gauge zuul.nodepool.current_requests
+        state = request.state
+        if request.canceled:
+            state = 'canceled'
+            dt = None
+        elif request.state in (model.STATE_FULFILLED, model.STATE_FAILED):
+            dt = int((request.state_time - request.requested_time) * 1000)
+        else:
+            dt = None
+        key = 'zuul.nodepool.%s' % state
+        statsd.incr(key)
+        if dt:
+            statsd.timing(key, dt)
+        for node in request.nodeset.getNodes():
+            statsd.incr(key + '.label.%s' % node.label)
+            if dt:
+                statsd.timing(key + '.label.%s' % node.label, dt)
+        statsd.incr(key + '.size.%s' % len(request.nodeset.nodes))
+        if dt:
+            statsd.timing(key + '.size.%s' % len(request.nodeset.nodes), dt)
+        statsd.gauge('zuul.nodepool.current_requests', len(self.requests))
+
     def requestNodes(self, build_set, job):
         # Create a copy of the nodeset to represent the actual nodes
         # returned by nodepool.
@@ -33,6 +62,7 @@
             self.sched.zk.submitNodeRequest(req, self._updateNodeRequest)
             # Logged after submission so that we have the request id
             self.log.info("Submited node request %s" % (req,))
+            self.emitStats(req)
         else:
             self.log.info("Fulfilling empty node request %s" % (req,))
             req.state = model.STATE_FULFILLED
@@ -90,10 +120,15 @@
         self.log.info("Returning nodeset %s" % (nodeset,))
         for node in nodeset.getNodes():
             if node.lock is None:
-                raise Exception("Node %s is not locked" % (node,))
-            if node.state == model.STATE_IN_USE:
-                node.state = model.STATE_USED
-                self.sched.zk.storeNode(node)
+                self.log.error("Node %s is not locked" % (node,))
+            else:
+                try:
+                    if node.state == model.STATE_IN_USE:
+                        node.state = model.STATE_USED
+                        self.sched.zk.storeNode(node)
+                except Exception:
+                    self.log.exception("Exception storing node %s "
+                                       "while unlocking:" % (node,))
         self._unlockNodes(nodeset.getNodes())
 
     def unlockNodeSet(self, nodeset):
@@ -106,18 +141,21 @@
             except Exception:
                 self.log.exception("Error unlocking node:")
 
-    def lockNodeSet(self, nodeset):
-        self._lockNodes(nodeset.getNodes())
+    def lockNodeSet(self, nodeset, request_id):
+        self._lockNodes(nodeset.getNodes(), request_id)
 
-    def _lockNodes(self, nodes):
+    def _lockNodes(self, nodes, request_id):
         # Try to lock all of the supplied nodes.  If any lock fails,
         # try to unlock any which have already been locked before
         # re-raising the error.
         locked_nodes = []
         try:
             for node in nodes:
+                if node.allocated_to != request_id:
+                    raise Exception("Node %s allocated to %s, not %s" %
+                                    (node.id, node.allocated_to, request_id))
                 self.log.debug("Locking node %s" % (node,))
-                self.sched.zk.lockNode(node)
+                self.sched.zk.lockNode(node, timeout=30)
                 locked_nodes.append(node)
         except Exception:
             self.log.exception("Error locking nodes:")
@@ -134,29 +172,48 @@
 
         if request.canceled:
             del self.requests[request.uid]
+            self.emitStats(request)
             return False
 
-        if request.state in (model.STATE_FULFILLED, model.STATE_FAILED):
+        # TODOv3(jeblair): handle allocation failure
+        if deleted:
+            self.log.debug("Resubmitting lost node request %s" % (request,))
+            request.id = None
+            self.sched.zk.submitNodeRequest(request, self._updateNodeRequest)
+        elif request.state in (model.STATE_FULFILLED, model.STATE_FAILED):
             self.log.info("Node request %s %s" % (request, request.state))
 
             # Give our results to the scheduler.
             self.sched.onNodesProvisioned(request)
             del self.requests[request.uid]
 
+            self.emitStats(request)
+
             # Stop watching this request node.
             return False
-        # TODOv3(jeblair): handle allocation failure
-        elif deleted:
-            self.log.debug("Resubmitting lost node request %s" % (request,))
-            self.sched.zk.submitNodeRequest(request, self._updateNodeRequest)
+
         return True
 
-    def acceptNodes(self, request):
+    def acceptNodes(self, request, request_id):
         # Called by the scheduler when it wants to accept and lock
         # nodes for (potential) use.
 
         self.log.info("Accepting node request %s" % (request,))
 
+        if request_id != request.id:
+            self.log.info("Skipping node accept for %s (resubmitted as %s)",
+                          request_id, request.id)
+            return
+
+        # Make sure the request still exists. It's possible it could have
+        # disappeared if we lost the ZK session between when the fulfillment
+        # response was added to our queue, and when we actually get around to
+        # processing it. Nodepool will automatically reallocate the assigned
+        # nodes in that situation.
+        if not self.sched.zk.nodeRequestExists(request):
+            self.log.info("Request %s no longer exists", request.id)
+            return
+
         if request.canceled:
             self.log.info("Ignoring canceled node request %s" % (request,))
             # The request was already deleted when it was canceled
@@ -166,7 +223,7 @@
         if request.fulfilled:
             # If the request suceeded, try to lock the nodes.
             try:
-                self.lockNodeSet(request.nodeset)
+                self.lockNodeSet(request.nodeset, request.id)
                 locked = True
             except Exception:
                 self.log.exception("Error locking nodes:")
diff --git a/zuul/rpcclient.py b/zuul/rpcclient.py
index 1a0a084..8f2e5dc 100644
--- a/zuul/rpcclient.py
+++ b/zuul/rpcclient.py
@@ -56,6 +56,14 @@
                 'count': count}
         return not self.submitJob('zuul:autohold', data).failure
 
+    def autohold_list(self):
+        data = {}
+        job = self.submitJob('zuul:autohold_list', data)
+        if job.failure:
+            return False
+        else:
+            return json.loads(job.data[0])
+
     def enqueue(self, tenant, pipeline, project, trigger, change):
         data = {'tenant': tenant,
                 'pipeline': pipeline,
diff --git a/zuul/rpclistener.py b/zuul/rpclistener.py
index 52a7e51..11d6684 100644
--- a/zuul/rpclistener.py
+++ b/zuul/rpclistener.py
@@ -50,6 +50,7 @@
 
     def register(self):
         self.worker.registerFunction("zuul:autohold")
+        self.worker.registerFunction("zuul:autohold_list")
         self.worker.registerFunction("zuul:enqueue")
         self.worker.registerFunction("zuul:enqueue_ref")
         self.worker.registerFunction("zuul:promote")
@@ -90,6 +91,17 @@
             except Exception:
                 self.log.exception("Exception while getting job")
 
+    def handle_autohold_list(self, job):
+        req = {}
+
+        # The json.dumps() call cannot handle dict keys that are not strings
+        # so we convert our key to a CSV string that the caller can parse.
+        for key, value in self.sched.autohold_requests.items():
+            new_key = ','.join(key)
+            req[new_key] = value
+
+        job.sendWorkComplete(json.dumps(req))
+
     def handle_autohold(self, job):
         args = json.loads(job.arguments)
         params = {}
diff --git a/zuul/scheduler.py b/zuul/scheduler.py
index 5432661..33b6723 100644
--- a/zuul/scheduler.py
+++ b/zuul/scheduler.py
@@ -15,7 +15,6 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-import extras
 import json
 import logging
 import os
@@ -31,6 +30,8 @@
 from zuul import exceptions
 from zuul import version as zuul_version
 from zuul.lib.config import get_default
+from zuul.lib.statsd import get_statsd
+import zuul.lib.queue
 
 
 class ManagementEvent(object):
@@ -71,10 +72,28 @@
     the path specified in the configuration.
 
     :arg Tenant tenant: the tenant to reconfigure
+    :arg Project project: if supplied, clear the cached configuration
+         from this project first
     """
-    def __init__(self, tenant):
+    def __init__(self, tenant, project):
         super(TenantReconfigureEvent, self).__init__()
-        self.tenant = tenant
+        self.tenant_name = tenant.name
+        self.projects = set([project])
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
+    def __eq__(self, other):
+        if not isinstance(other, TenantReconfigureEvent):
+            return False
+        # We don't check projects because they will get combined when
+        # merged.
+        return (self.tenant_name == other.tenant_name)
+
+    def merge(self, other):
+        if self.tenant_name != other.tenant_name:
+            raise Exception("Can not merge events from different tenants")
+        self.projects |= other.projects
 
 
 class PromoteEvent(ManagementEvent):
@@ -161,6 +180,7 @@
 
     def __init__(self, request):
         self.request = request
+        self.request_id = request.id
 
 
 def toList(item):
@@ -207,7 +227,7 @@
         self.executor = None
         self.merger = None
         self.connections = None
-        self.statsd = extras.try_import('statsd.statsd')
+        self.statsd = get_statsd(config)
         # TODO(jeblair): fix this
         # Despite triggers being part of the pipeline, there is one trigger set
         # per scheduler. The pipeline handles the trigger filters but since
@@ -219,7 +239,7 @@
 
         self.trigger_event_queue = queue.Queue()
         self.result_event_queue = queue.Queue()
-        self.management_event_queue = queue.Queue()
+        self.management_event_queue = zuul.lib.queue.MergedQueue()
         self.abide = model.Abide()
 
         if not testonly:
@@ -259,22 +279,16 @@
         self.zk = zk
 
     def addEvent(self, event):
-        self.log.debug("Adding trigger event: %s" % event)
         self.trigger_event_queue.put(event)
         self.wake_event.set()
-        self.log.debug("Done adding trigger event: %s" % event)
 
     def onBuildStarted(self, build):
-        self.log.debug("Adding start event for build: %s" % build)
         build.start_time = time.time()
         event = BuildStartedEvent(build)
         self.result_event_queue.put(event)
         self.wake_event.set()
-        self.log.debug("Done adding start event for build: %s" % build)
 
     def onBuildCompleted(self, build, result, result_data):
-        self.log.debug("Adding complete event for build: %s result: %s" % (
-            build, result))
         build.end_time = time.time()
         build.result_data = result_data
         # Note, as soon as the result is set, other threads may act
@@ -284,61 +298,60 @@
         build.result = result
         try:
             if self.statsd and build.pipeline:
-                jobname = build.job.name.replace('.', '_')
-                key = 'zuul.pipeline.%s.all_jobs' % build.pipeline.name
+                tenant = build.pipeline.layout.tenant
+                jobname = build.job.name.replace('.', '_').replace('/', '_')
+                hostname = (build.build_set.item.change.project.
+                            canonical_hostname.replace('.', '_'))
+                projectname = (build.build_set.item.change.project.name.
+                               replace('.', '_').replace('/', '_'))
+                branchname = (build.build_set.item.change.branch.
+                              replace('.', '_').replace('/', '_'))
+                basekey = 'zuul.tenant.%s' % tenant.name
+                pipekey = '%s.pipeline.%s' % (basekey, build.pipeline.name)
+                # zuul.tenant.<tenant>.pipeline.<pipeline>.all_jobs
+                key = '%s.all_jobs' % pipekey
                 self.statsd.incr(key)
-                for label in build.node_labels:
-                    # Jenkins includes the node name in its list of labels, so
-                    # we filter it out here, since that is not statistically
-                    # interesting.
-                    if label == build.node_name:
-                        continue
-                    dt = int((build.start_time - build.execute_time) * 1000)
-                    key = 'zuul.pipeline.%s.label.%s.wait_time' % (
-                        build.pipeline.name, label)
-                    self.statsd.timing(key, dt)
-                key = 'zuul.pipeline.%s.job.%s.%s' % (build.pipeline.name,
-                                                      jobname, build.result)
+                jobkey = '%s.project.%s.%s.%s.job.%s' % (
+                    pipekey, hostname, projectname, branchname, jobname)
+                # zuul.tenant.<tenant>.pipeline.<pipeline>.project.
+                #   <host>.<project>.<branch>.job.<job>.<result>
+                key = '%s.%s' % (jobkey, build.result)
                 if build.result in ['SUCCESS', 'FAILURE'] and build.start_time:
                     dt = int((build.end_time - build.start_time) * 1000)
                     self.statsd.timing(key, dt)
                 self.statsd.incr(key)
-
-                key = 'zuul.pipeline.%s.job.%s.wait_time' % (
-                    build.pipeline.name, jobname)
+                # zuul.tenant.<tenant>.pipeline.<pipeline>.project.
+                #  <host>.<project>.<branch>.job.<job>.wait_time
+                key = '%s.wait_time' % jobkey
                 dt = int((build.start_time - build.execute_time) * 1000)
                 self.statsd.timing(key, dt)
-        except:
+        except Exception:
             self.log.exception("Exception reporting runtime stats")
         event = BuildCompletedEvent(build)
         self.result_event_queue.put(event)
         self.wake_event.set()
-        self.log.debug("Done adding complete event for build: %s" % build)
 
     def onMergeCompleted(self, build_set, merged, updated,
                          commit, files, repo_state):
-        self.log.debug("Adding merge complete event for build set: %s" %
-                       build_set)
         event = MergeCompletedEvent(build_set, merged,
                                     updated, commit, files, repo_state)
         self.result_event_queue.put(event)
         self.wake_event.set()
 
     def onNodesProvisioned(self, req):
-        self.log.debug("Adding nodes provisioned event for build set: %s" %
-                       req.build_set)
         event = NodesProvisionedEvent(req)
         self.result_event_queue.put(event)
         self.wake_event.set()
 
-    def reconfigureTenant(self, tenant):
-        self.log.debug("Prepare to reconfigure")
-        event = TenantReconfigureEvent(tenant)
+    def reconfigureTenant(self, tenant, project):
+        self.log.debug("Submitting tenant reconfiguration event for "
+                       "%s due to project %s", tenant.name, project)
+        event = TenantReconfigureEvent(tenant, project)
         self.management_event_queue.put(event)
         self.wake_event.set()
 
     def reconfigure(self, config):
-        self.log.debug("Prepare to reconfigure")
+        self.log.debug("Submitting reconfiguration event")
         event = ReconfigureEvent(config)
         self.management_event_queue.put(event)
         self.wake_event.set()
@@ -457,7 +470,7 @@
         self.layout_lock.acquire()
         self.config = event.config
         try:
-            self.log.debug("Performing reconfiguration")
+            self.log.debug("Full reconfiguration beginning")
             loader = configloader.ConfigLoader()
             abide = loader.loadConfig(
                 self.config.get('scheduler', 'tenant_config'),
@@ -468,24 +481,31 @@
             self.abide = abide
         finally:
             self.layout_lock.release()
+        self.log.debug("Full reconfiguration complete")
 
     def _doTenantReconfigureEvent(self, event):
         # This is called in the scheduler loop after another thread submits
         # a request
         self.layout_lock.acquire()
         try:
-            self.log.debug("Performing tenant reconfiguration")
+            self.log.debug("Tenant reconfiguration beginning")
+            # If a change landed to a project, clear out the cached
+            # config before reconfiguring.
+            for project in event.projects:
+                project.unparsed_config = None
+            old_tenant = self.abide.tenants[event.tenant_name]
             loader = configloader.ConfigLoader()
             abide = loader.reloadTenant(
                 self.config.get('scheduler', 'tenant_config'),
                 self._get_project_key_dir(),
                 self, self.merger, self.connections,
-                self.abide, event.tenant)
-            tenant = abide.tenants[event.tenant.name]
+                self.abide, old_tenant)
+            tenant = abide.tenants[event.tenant_name]
             self._reconfigureTenant(tenant)
             self.abide = abide
         finally:
             self.layout_lock.release()
+        self.log.debug("Tenant reconfiguration complete")
 
     def _reenqueueGetProject(self, tenant, item):
         project = item.change.project
@@ -769,8 +789,7 @@
                     # or a branch was just created or deleted.  Clear
                     # out cached data for this project and perform a
                     # reconfiguration.
-                    change.project.unparsed_config = None
-                    self.reconfigureTenant(tenant)
+                    self.reconfigureTenant(tenant, change.project)
                 for pipeline in tenant.layout.pipelines.values():
                     if event.isPatchsetCreated():
                         pipeline.manager.removeOldVersionsOfChange(change)
@@ -853,8 +872,12 @@
             try:
                 self.nodepool.holdNodeSet(nodeset, autohold_key)
             except Exception:
-                self.log.exception("Unable to process autohold for %s",
+                self.log.exception("Unable to process autohold for %s:",
                                    autohold_key)
+                if autohold_key in self.autohold_requests:
+                    self.log.debug("Removing autohold %s due to exception",
+                                   autohold_key)
+                    del self.autohold_requests[autohold_key]
 
             self.nodepool.returnNodeSet(nodeset)
         except Exception:
@@ -891,9 +914,10 @@
 
     def _doNodesProvisionedEvent(self, event):
         request = event.request
+        request_id = event.request_id
         build_set = request.build_set
 
-        self.nodepool.acceptNodes(request)
+        self.nodepool.acceptNodes(request, request_id)
         if request.canceled:
             return
 
@@ -932,6 +956,9 @@
         data['result_event_queue'] = {}
         data['result_event_queue']['length'] = \
             self.result_event_queue.qsize()
+        data['management_event_queue'] = {}
+        data['management_event_queue']['length'] = \
+            self.management_event_queue.qsize()
 
         if self.last_reconfigured:
             data['last_reconfigured'] = self.last_reconfigured * 1000
diff --git a/zuul/zk.py b/zuul/zk.py
index 5ea4e56..ede78be 100644
--- a/zuul/zk.py
+++ b/zuul/zk.py
@@ -90,7 +90,7 @@
     def resetLostFlag(self):
         self._became_lost = False
 
-    def connect(self, hosts, read_only=False):
+    def connect(self, hosts, read_only=False, timeout=10.0):
         '''
         Establish a connection with ZooKeeper cluster.
 
@@ -100,10 +100,12 @@
         :param str hosts: Comma-separated list of hosts to connect to (e.g.
             127.0.0.1:2181,127.0.0.1:2182,[::1]:2183).
         :param bool read_only: If True, establishes a read-only connection.
-
+        :param float timeout: The ZooKeeper session timeout, in
+            seconds (default: 10.0).
         '''
         if self.client is None:
-            self.client = KazooClient(hosts=hosts, read_only=read_only)
+            self.client = KazooClient(hosts=hosts, read_only=read_only,
+                                      timeout=timeout)
             self.client.add_listener(self._connection_listener)
             self.client.start()
 
@@ -145,12 +147,10 @@
             from ZooKeeper).  The watcher should return False when
             further updates are no longer necessary.
         '''
-        priority = 100  # TODO(jeblair): integrate into nodereq
-
         data = node_request.toDict()
         data['created_time'] = time.time()
 
-        path = '%s/%s-' % (self.REQUEST_ROOT, priority)
+        path = '%s/%s-' % (self.REQUEST_ROOT, node_request.priority)
         path = self.client.create(path, self._dictToStr(data),
                                   makepath=True,
                                   sequence=True, ephemeral=True)
@@ -160,7 +160,6 @@
         def callback(data, stat):
             if data:
                 data = self._strToDict(data)
-                node_request.updateFromDict(data)
                 request_nodes = list(node_request.nodeset.getNodes())
                 for i, nodeid in enumerate(data.get('nodes', [])):
                     node_path = '%s/%s' % (self.NODE_ROOT, nodeid)
@@ -168,6 +167,7 @@
                     node_data = self._strToDict(node_data)
                     request_nodes[i].id = nodeid
                     request_nodes[i].updateFromDict(node_data)
+                node_request.updateFromDict(data)
             deleted = (data is None)  # data *are* none
             return watcher(node_request, deleted)
 
@@ -187,6 +187,19 @@
         except kze.NoNodeError:
             pass
 
+    def nodeRequestExists(self, node_request):
+        '''
+        See if a NodeRequest exists in ZooKeeper.
+
+        :param NodeRequest node_request: A NodeRequest to verify.
+
+        :returns: True if the request exists, False otherwise.
+        '''
+        path = '%s/%s' % (self.REQUEST_ROOT, node_request.id)
+        if self.client.exists(path):
+            return True
+        return False
+
     def storeNode(self, node):
         '''Store the node.
 
@@ -256,6 +269,9 @@
         for nodeid in nodes:
             node_path = '%s/%s' % (self.NODE_ROOT, nodeid)
             node_data, node_stat = self.client.get(node_path)
+            if not node_data:
+                self.log.warning("Node ID %s has no data", nodeid)
+                continue
             node_data = self._strToDict(node_data)
             if (node_data['state'] == zuul.model.STATE_HOLD and
                     node_data.get('hold_job') == identifier):