Add quota support to zuul-launcher

This does the following:

* Retries node requests without penalty if they hit an unexpected
  quota error.  The new node state "TEMPFAIL" is used for this.
* Calculates total provider quota in a method similar to nodepool
  and caches it and updates it every 5 minutes.
* Keeps track of realtime zuul quota usage by way of a custom
  TreeCache that adds and removes quota as it sees events.  This
  lets us know real-time usage without needing to iterate over
  all nodes.
* Avoids starting a node state machine if it is expected that the
  node will put the provider over quota.  This is similar, but not
  quite the same as nodepool's idea of a paused provider.
* Attempts to keep the percentage quota used of all providers
  roughly equivalent while still including some randomization.
  This is still likely not a final allocation algorithm.

This has some issues that will need to be resolved in subsequent
commits:

* It is possible for a node request to be starved if multiple launchers
  are running for a provider, and there are requests of different
  sizes for that provider (the large one may never run if the small
  ones are able to fit in the last bit of quota).
* Because we record quota failures as node objects in ZK, if a node
  request continually hits quota errors, our node records in ZK will
  grow without bound.

Despite these issues, this change is useful for continued testing
of the system, and since it is mainly focused with adding quota
support and only minimally changes the node/request handling algorithms,
it stands on its own.

Change-Id: I6d65db2b103cc3f0aec7e46a6a63a393399c91eb

tests/base.py

@@ -246,7 +246,7 @@ def driver_config(driver, **kw):
         if driver_dict is None:
             driver_dict = {}
             test.__driver_config__ = driver_dict
-        driver_dict[driver] == kw
+        driver_dict[driver] = kw
         return test
     return decorator
 
@@ -3877,6 +3877,14 @@ class ZuulTestCase(BaseTestCase):
         path = os.path.join(self.test_root, "changes.data")
         self.test_config.changes.load(path)
 
+    def waitForNodeRequest(self, request, timeout=10):
+        with self.createZKContext(None) as ctx:
+            for _ in iterate_timeout(
+                    timeout, "nodeset request to be fulfilled"):
+                request.refresh(ctx)
+                if request.state == model.NodesetRequest.State.FULFILLED:
+                    return
+
     def requestNodes(self, labels, tenant="tenant-one", pipeline="check",
                      timeout=10):
         result_queue = PipelineResultEventQueue(
@@ -3899,14 +3907,17 @@ class ZuulTestCase(BaseTestCase):
                     zuul_event_id=uuid.uuid4().hex,
                     span_info=None,
                 )
-                for _ in iterate_timeout(
-                        timeout, "nodeset request to be fulfilled"):
-                    result_events = list(result_queue)
-                    if result_events:
-                        for event in result_events:
-                            # Remove event(s) from queue
-                            result_queue.ack(event)
-                        break
+                if timeout:
+                    for _ in iterate_timeout(
+                            timeout, "nodeset request to be fulfilled"):
+                        result_events = list(result_queue)
+                        if result_events:
+                            for event in result_events:
+                                # Remove event(s) from queue
+                                result_queue.ack(event)
+                            break
+                else:
+                    return request
 
             self.assertEqual(len(result_events), 1)
             for event in result_queue:

tests/fake_openstack.py

@@ -258,6 +258,23 @@ class FakeOpenstackConnection:
         floating_ip['port_id'] = 'fake'
         server['addresses']['public'].append(floating_ip._fake_toDict())
 
+    def get_compute_limits(self):
+        return FakeOpenstackObject(
+            max_total_cores=self.cloud.max_cores,
+            max_total_instances=self.cloud.max_instances,
+            max_total_ram_size=self.cloud.max_ram,
+            total_cores_used=4 * len(self.cloud.servers),
+            total_instances_used=len(self.cloud.servers),
+            total_ram_used=8192 * len(self.cloud.servers),
+        )
+
+    def get_volume_limits(self):
+        return FakeOpenstackObject(
+            absolute=dict(
+                maxTotalVolumes=self.cloud.max_volumes,
+                maxTotalVolumeGigabytes=self.cloud.max_volume_gb,
+            ))
+
 
 class FakeOpenstackProviderEndpoint(OpenstackProviderEndpoint):
     def _getClient(self):

tests/unit/test_aws_driver.py

@@ -30,6 +30,7 @@ from tests.base import (
     iterate_timeout,
     simple_layout,
     return_data,
+    driver_config,
 )
 from tests.unit.test_launcher import ImageMocksFixture
 from tests.unit.test_cloud_driver import BaseCloudDriverTest
@@ -171,6 +172,13 @@ class TestAwsDriver(BaseCloudDriverTest):
     def test_aws_node_lifecycle(self):
         self._test_node_lifecycle('debian-normal')
 
+    @simple_layout('layouts/nodepool.yaml', enable_nodepool=True)
+    @driver_config('aws', ec2_quotas={
+        'L-1216C47A': 2,
+    })
+    def test_aws_quota(self):
+        self._test_quota('debian-normal')
+
     @simple_layout('layouts/aws/nodepool-image-snapshot.yaml',
                    enable_nodepool=True)
     @return_data(

tests/unit/test_cloud_driver.py

@@ -80,7 +80,9 @@ class BaseCloudDriverTest(ZuulTestCase):
             self.assertTrue(pnode.hasLock())
             self._assertProviderNodeAttributes(pnode)
 
-        self.assertGreater(len(list(endpoint.listInstances())), 0)
+        for _ in iterate_timeout(10, "instances to appear"):
+            if len(list(endpoint.listInstances())) > 0:
+                break
         client.useNodeset(nodeset)
         self.waitUntilSettled()
 
@@ -109,6 +111,71 @@ class BaseCloudDriverTest(ZuulTestCase):
             if len(list(endpoint.listInstances())) == 0:
                 break
 
+    def _test_quota(self, label):
+        for _ in iterate_timeout(
+                30, "scheduler and launcher to have the same layout"):
+            if (self.scheds.first.sched.local_layout_state.get("tenant-one") ==
+                self.launcher.local_layout_state.get("tenant-one")):
+                break
+        endpoint = self._getEndpoint()
+        nodeset1 = model.NodeSet()
+        nodeset1.addNode(model.Node("node", label))
+
+        nodeset2 = model.NodeSet()
+        nodeset2.addNode(model.Node("node", label))
+
+        ctx = self.createZKContext(None)
+        request1 = self.requestNodes([n.label for n in nodeset1.getNodes()])
+        request2 = self.requestNodes(
+            [n.label for n in nodeset2.getNodes()],
+            timeout=None)
+
+        client = LauncherClient(self.zk_client, None)
+        request1 = client.getRequest(request1.uuid)
+
+        self.assertEqual(request1.state, model.NodesetRequest.State.FULFILLED)
+        self.assertEqual(len(request1.nodes), 1)
+
+        client.acceptNodeset(request1, nodeset1)
+        client.useNodeset(nodeset1)
+
+        # We should still be waiting on request2.
+
+        # TODO: This is potentially racy (but only producing
+        # false-negatives) and also slow. We should find a way to
+        # determine if the launcher had paused a provider.
+        with testtools.ExpectedException(Exception):
+            self.waitForNodeRequest(request2, 10)
+        request2 = client.getRequest(request2.uuid)
+        self.assertEqual(request2.state, model.NodesetRequest.State.ACCEPTED)
+
+        client.returnNodeset(nodeset1)
+        self.waitUntilSettled()
+
+        for node in nodeset1.getNodes():
+            pnode = node._provider_node
+            for _ in iterate_timeout(60, "node to be deleted"):
+                try:
+                    pnode.refresh(ctx)
+                except NoNodeError:
+                    break
+
+        self.waitForNodeRequest(request2, 10)
+        request2 = client.getRequest(request2.uuid)
+        self.assertEqual(request2.state, model.NodesetRequest.State.FULFILLED)
+        self.assertEqual(len(request2.nodes), 1)
+
+        client.acceptNodeset(request2, nodeset2)
+        client.useNodeset(nodeset2)
+        client.returnNodeset(nodeset2)
+        self.waitUntilSettled()
+
+        # Iterate here because the aws driver (at least) performs
+        # delayed async deletes.
+        for _ in iterate_timeout(60, "instances to be deleted"):
+            if len(list(endpoint.listInstances())) == 0:
+                break
+
     def _test_diskimage(self):
         self.waitUntilSettled()
         self.assertHistory([

tests/unit/test_launcher.py

@@ -13,12 +13,15 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import math
 import textwrap
 import time
 from collections import defaultdict
 from unittest import mock
 
+from zuul import exceptions
 from zuul import model
+import zuul.driver.aws.awsendpoint
 from zuul.launcher.client import LauncherClient
 
 import responses
@@ -146,6 +149,13 @@ class LauncherBaseTestCase(ZuulTestCase):
             Bucket='zuul',
             CreateBucketConfiguration={'LocationConstraint': 'us-west-2'})
         self.addCleanup(self.mock_aws.stop)
+        self.patch(zuul.driver.aws.awsendpoint, 'CACHE_TTL', 1)
+
+        def getQuotaLimits(self):
+            return model.QuotaInformation(default=math.inf)
+        self.patch(zuul.driver.aws.awsprovider.AwsProvider,
+                   'getQuotaLimits',
+                   getQuotaLimits)
         super().setUp()
 
     def _nodes_by_label(self):
@@ -765,6 +775,22 @@ class TestLauncher(LauncherBaseTestCase):
                 except NoNodeError:
                     break
 
+    @simple_layout('layouts/nodepool.yaml', enable_nodepool=True)
+    @mock.patch(
+        'zuul.driver.aws.awsendpoint.AwsProviderEndpoint._createInstance',
+        side_effect=exceptions.QuotaException)
+    def test_quota_failure(self, mock_create):
+        # This tests an unexpected quota error.
+        # The request should never be fulfilled
+        with testtools.ExpectedException(Exception):
+            self.requestNodes(["debian-normal"])
+
+        # We should have tried to build at least one node that was
+        # marked as tempfail.
+        nodes = self.launcher.api.nodes_cache.getItems()
+        self.assertTrue(len(nodes) > 1)
+        self.assertEqual(model.ProviderNode.State.TEMPFAILED, nodes[0].state)
+
     @simple_layout('layouts/nodepool-nodescan.yaml', enable_nodepool=True)
     @okay_tracebacks('_checkNodescanRequest')
     @mock.patch('paramiko.transport.Transport')

tests/unit/test_openstack_driver.py

@@ -28,6 +28,7 @@ from tests.base import (
     ZuulTestCase,
     simple_layout,
     return_data,
+    driver_config,
 )
 from tests.unit.test_launcher import ImageMocksFixture
 from tests.unit.test_cloud_driver import BaseCloudDriverTest
@@ -68,6 +69,17 @@ class BaseOpenstackDriverTest(ZuulTestCase):
             needs_floating_ip=self.openstack_needs_floating_ip,
             auto_attach_floating_ip=self.openstack_auto_attach_floating_ip,
         )
+        self.fake_cloud.max_instances =\
+            self.test_config.driver.openstack.get('max_instances', 100)
+        self.fake_cloud.max_cores =\
+            self.test_config.driver.openstack.get('max_cores', 100)
+        self.fake_cloud.max_ram =\
+            self.test_config.driver.openstack.get('max_ram', 1000000)
+        self.fake_cloud.max_volumes =\
+            self.test_config.driver.openstack.get('max_volumes', 100)
+        self.fake_cloud.max_volume_gb =\
+            self.test_config.driver.openstack.get('max_volume_gb', 100)
+
         self.patch(OpenstackDriver, '_endpoint_class',
                    FakeOpenstackProviderEndpoint)
         self.patch(FakeOpenstackProviderEndpoint,
@@ -90,6 +102,11 @@ class TestOpenstackDriver(BaseOpenstackDriverTest, BaseCloudDriverTest):
     def test_openstack_node_lifecycle(self):
         self._test_node_lifecycle('debian-normal')
 
+    @simple_layout('layouts/openstack/nodepool.yaml', enable_nodepool=True)
+    @driver_config('openstack', max_cores=4)
+    def test_openstack_quota(self):
+        self._test_quota('debian-normal')
+
     @simple_layout('layouts/openstack/nodepool-image.yaml',
                    enable_nodepool=True)
     @return_data(

zuul/driver/aws/awsendpoint.py

@@ -45,9 +45,9 @@ from zuul.driver.aws.util import tag_dict_to_list, tag_list_to_dict
 from zuul.driver.util import (
     ImageUploader,
     LazyExecutorTTLCache,
-    QuotaInformation,
     RateLimiter,
 )
+from zuul.model import QuotaInformation
 from zuul.provider import (
     BaseProviderEndpoint,
     statemachine
@@ -148,7 +148,7 @@ class AwsCreateStateMachine(statemachine.StateMachine):
             "image_external_id", image_external_id)
 
         # Restore local objects
-        self.quota = self.endpoint.getQuotaForLabel(
+        self.node.quota = self.endpoint.getQuotaForLabel(
             self.label, self.flavor)
 
         if self.state in (
@@ -252,7 +252,7 @@ class AwsCreateStateMachine(statemachine.StateMachine):
         if self.state == self.COMPLETE:
             self.complete = True
             return AwsInstance(self.endpoint.region, self.instance,
-                               self.host, self.quota)
+                               self.host, self.node.quota)
 
 
 class EbsSnapshotUploader(ImageUploader):

zuul/driver/aws/awsprovider.py

@@ -31,7 +31,7 @@ from zuul.driver.aws.const import (
     ON_DEMAND,
     VOLUME_QUOTA_CODES,
 )
-from zuul.driver.util import QuotaInformation
+from zuul.model import QuotaInformation
 from zuul.provider import (
     BaseProvider,
     BaseProviderFlavor,
@@ -260,16 +260,16 @@ class AwsProvider(BaseProvider, subclass_id='aws'):
         volume_types = set()
         ec2_quotas = self.endpoint._listEC2Quotas()
         ebs_quotas = self.endpoint._listEBSQuotas()
-        for label in self.labels.values():
-            if label.dedicated_host:
-                host_types.add(label.instance_type)
+        for flavor in self.flavors.values():
+            if flavor.dedicated_host:
+                host_types.add(flavor.instance_type)
             else:
-                if label.instance_type not in instance_types:
-                    instance_types[label.instance_type] = set()
-                instance_types[label.instance_type].add(
-                    SPOT if label.use_spot else ON_DEMAND)
-            if label.volume_type:
-                volume_types.add(label.volume_type)
+                if flavor.instance_type not in instance_types:
+                    instance_types[flavor.instance_type] = set()
+                instance_types[flavor.instance_type].add(
+                    SPOT if flavor.market_type == 'spot' else ON_DEMAND)
+            if flavor.volume_type:
+                volume_types.add(flavor.volume_type)
         args = dict(default=math.inf)
         for instance_type in instance_types:
             for market_type_option in instance_types[instance_type]:

zuul/driver/openstack/openstackendpoint.py

@@ -34,10 +34,10 @@ from zuul.driver.openstack.openstackmodel import (
 )
 from zuul.driver.util import (
     LazyExecutorTTLCache,
-    QuotaInformation,
     RateLimiter,
     Timer,
 )
+from zuul.model import QuotaInformation
 from zuul.provider import (
     BaseProviderEndpoint,
     statemachine
@@ -261,7 +261,7 @@ class OpenstackCreateStateMachine(statemachine.StateMachine):
             flavor_name=flavor.flavor_name,
             # min_ram=self.label.min_ram,
         )
-        self.quota = quota_from_flavor(self.os_flavor, label=self.label)
+        self.node.quota = quota_from_flavor(self.os_flavor, label=self.label)
         self.node.openstack_server_id = None
 
     def _handleServerFault(self):
@@ -375,7 +375,7 @@ class OpenstackCreateStateMachine(statemachine.StateMachine):
 
         if self.state == self.COMPLETE:
             self.complete = True
-            return self.endpoint._getInstance(self.server, self.quota)
+            return self.endpoint._getInstance(self.server, self.node.quota)
 
 
 class OpenstackProviderEndpoint(BaseProviderEndpoint):

zuul/driver/openstack/openstackmodel.py

@@ -49,6 +49,10 @@ class OpenstackInstance(statemachine.Instance):
     def getQuotaInformation(self):
         return self.quota
 
+    @property
+    def external_id(self):
+        return f'server={self.openstack_server_id}'
+
 
 class OpenstackResource(statemachine.Resource):
     TYPE_HOST = 'host'

zuul/driver/openstack/openstackprovider.py

@@ -13,7 +13,6 @@
 # under the License.
 
 import logging
-import math
 
 import zuul.provider.schema as provider_schema
 from zuul.lib.voluputil import (
@@ -26,7 +25,6 @@ from zuul.driver.openstack.openstackendpoint import (
     OpenstackCreateStateMachine,
     OpenstackDeleteStateMachine,
 )
-from zuul.driver.util import QuotaInformation
 from zuul.provider import (
     BaseProvider,
     BaseProviderFlavor,
@@ -233,9 +231,7 @@ class OpenstackProvider(BaseProvider, subclass_id='openstack'):
         pass
 
     def getQuotaLimits(self):
-        # TODO: implement
-        args = dict(default=math.inf)
-        return QuotaInformation(**args)
+        return self.endpoint.getQuotaLimits()
 
     def getQuotaForLabel(self, label):
         flavor = self.flavors[label.flavor]

zuul/driver/util.py

@@ -51,54 +51,6 @@ def to_list(item):
     return [item]
 
 
-class QuotaInformation:
-    def __init__(self, default=0, **kw):
-        '''
-        Initializes the quota information with some values. None values will
-        be initialized with default which will be typically 0 or math.inf
-        indicating an infinite limit.
-
-        :param default: The default value to use for any attribute not supplied
-                        (usually 0 or math.inf).
-        '''
-        self.quota = {}
-        for k, v in kw.items():
-            self.quota[k] = v
-        self.default = default
-
-    def _get_default(self, value, default):
-        return value if value is not None else default
-
-    def _add_subtract(self, other, add=True):
-        for resource in other.quota.keys():
-            self.quota.setdefault(resource, self.default)
-        for resource in self.quota.keys():
-            other_value = other.quota.get(resource, other.default)
-            if add:
-                self.quota[resource] += other_value
-            else:
-                self.quota[resource] -= other_value
-
-    def subtract(self, other):
-        self._add_subtract(other, add=False)
-
-    def add(self, other):
-        self._add_subtract(other, True)
-
-    def nonNegative(self):
-        for resource, value in self.quota.items():
-            if value < 0:
-                return False
-        return True
-
-    def getResources(self):
-        '''Return resources value to register in ZK node'''
-        return self.quota
-
-    def __str__(self):
-        return str(self.quota)
-
-
 class RateLimitInstance:
     def __init__(self, limiter, logger, msg):
         self.limiter = limiter

zuul/exceptions.py

@@ -91,6 +91,10 @@ class RuntimeConfigurationException(Exception):
     pass
 
 
+class QuotaException(Exception):
+    pass
+
+
 # Authentication Exceptions
 class AuthTokenException(Exception):
     defaultMsg = 'Unknown Error'

zuul/launcher/server.py

@@ -29,6 +29,7 @@ import threading
 import time
 import uuid
 
+import cachetools
 import mmh3
 import paramiko
 import requests
@@ -658,6 +659,7 @@ class Launcher:
     # Max. time the main event loop is allowed to sleep
     MAX_SLEEP = 1
     DELETE_TIMEOUT = 600
+    MAX_QUOTA_AGE = 5 * 60  # How long to keep the quota information cached
 
     def __init__(self, config, connections):
         self._running = True
@@ -669,6 +671,9 @@ class Launcher:
         # launcher
         self.endpoints = {}
 
+        self._provider_quota_cache = cachetools.TTLCache(
+            maxsize=8192, ttl=self.MAX_QUOTA_AGE)
+
         self.tracing = tracing.Tracing(self.config)
         self.zk_client = ZooKeeperClient.fromConfig(self.config)
         self.zk_client.connect()
@@ -906,8 +911,11 @@ class Launcher:
 
         label_providers = []
         for i, label_name in enumerate(request.labels):
+            # Get a list of candidate providers along with their
+            # approximate quota usage.
             candidate_providers = [
-                p for p in providers
+                (p, self.getQuotaPercentage(p))
+                for p in providers
                 if p.hasLabel(label_name)
                 and provider_failures[p.canonical_name] < p.launch_attempts
             ]
@@ -916,7 +924,13 @@ class Launcher:
                     f"No provider found for label {label_name}")
 
             # TODO: make provider selection more sophisticated
-            provider = random.choice(candidate_providers)
+            # Start by randomizing them
+            random.shuffle(candidate_providers)
+            # Then sort by quota used; providers under quota will
+            # still be randomized with other providers within 10% of
+            # the same quota usage values.
+            candidate_providers.sort(key=lambda x: x[1])
+            provider = candidate_providers[0][0]
             log.debug("Selected provider %s from candidate providers: %s",
                       provider, candidate_providers)
             label = provider.labels[label_name]
@@ -959,7 +973,7 @@ class Launcher:
         requested_nodes = []
         for i, node_id in enumerate(request.nodes):
             node = self.api.getProviderNode(node_id)
-            if node.state == node.State.FAILED:
+            if node.state in (node.State.FAILED, node.State.TEMPFAILED):
                 label_providers = self._selectProviders(request, log)
                 label, provider = label_providers[i]
                 log.info("Retrying request with provider %s", provider)
@@ -1009,6 +1023,13 @@ class Launcher:
                     and node.state in node.CREATE_STATES):
                 try:
                     self._checkNode(node, log)
+                except exceptions.QuotaException as e:
+                    state = node.State.TEMPFAILED
+                    log.info("Marking node %s as %s due to %s", node, state, e)
+                    with self.createZKContext(node._lock, self.log) as ctx:
+                        with node.activeContext(ctx):
+                            node.setState(state)
+                        self.wake_event.set()
                 except Exception:
                     state = node.State.FAILED
                     log.exception("Marking node %s as %s", node, state)
@@ -1086,6 +1107,7 @@ class Launcher:
         # TODO: check timeout
         with self.createZKContext(node._lock, self.log) as ctx:
             with node.activeContext(ctx):
+                provider = None
                 if not node.create_state_machine:
                     log.debug("Building node %s", node)
                     provider = self._getProviderForNode(node)
@@ -1096,6 +1118,15 @@ class Launcher:
                         node, image_external_id, log)
 
                 old_state = node.create_state_machine.state
+                if old_state == node.create_state_machine.START:
+                    provider = provider or self._getProviderForNode(node)
+                    if not self.doesProviderHaveQuotaForNode(
+                            provider, node, log):
+                        # Consider this provider paused, don't attempt
+                        # to create the node until we think we have
+                        # quota.
+                        self.wake_event.set()
+                        return
                 instance = node.create_state_machine.advance()
                 new_state = node.create_state_machine.state
                 if old_state != new_state:
@@ -1339,8 +1370,7 @@ class Launcher:
 
         # If we did not know the resource information before
         # launching, update it now.
-        # TODO:
-        # node.resources = instance.getQuotaInformation().get_resources()
+        node.quota = instance.getQuotaInformation()
 
         # Optionally, if the node has updated values that we set from
         # the image attributes earlier, set those.
@@ -1738,3 +1768,77 @@ class Launcher:
         # Uploads are already sorted by timestamp
         image_upload = valid_uploads[-1]
         return image_upload.external_id
+
+    def getQuotaUsed(self, provider):
+        used = model.QuotaInformation()
+        for node in self.api.getProviderNodes():
+            if (provider.canonical_name == node.provider and
+                node.state in node.ALLOCATED_STATES):
+                used.add(node.quota)
+        return used
+
+    def getUnmanagedQuotaUsed(self, provider):
+        used = model.QuotaInformation()
+
+        node_ids = self.api.nodes_cache.getKeys()
+        endpoint = provider.getEndpoint()
+        system_id = self.system.system_id
+        for instance in endpoint.listInstances():
+            meta = instance.metadata
+            if (meta.get('zuul_system_id') != system_id or
+                meta.get('zuul_node_uuid') in node_ids):
+                continue
+            qi = instance.getQuotaInformation()
+            used.add(qi)
+        return used
+
+    def getProviderQuota(self, provider):
+        val = self._provider_quota_cache.get(provider.canonical_name)
+        if val:
+            return val
+
+        # This is initialized with the full tenant quota and later becomes
+        # the quota available for nodepool.
+        quota = provider.getQuotaLimits()
+        self.log.debug("Provider quota for %s: %s",
+                       provider.name, quota)
+
+        unmanaged = self.getUnmanagedQuotaUsed(provider)
+        self.log.debug("Provider unmaneged quota used for %s: %s",
+                       provider.name, quota)
+
+        # Subtract the unmanaged quota usage from nodepool_max
+        # to get the quota available for us.
+        quota.subtract(unmanaged)
+        self._provider_quota_cache[provider.canonical_name] = quota
+        return quota
+
+    def getQuotaPercentage(self, provider):
+        # This is cached and updated every 5 minutes
+        total = self.getProviderQuota(provider).copy()
+        # This is continuously updated in the background
+        used = self.api.nodes_cache.getQuota(provider)
+        pct = 0.0
+        for resource in total.quota.keys():
+            used_r = used.quota.get(resource, used.default)
+            total_r = total.quota[resource]
+            pct = max(used_r / total_r, pct)
+        if pct < 1.0:
+            # If we are below 100% usage, lose precion so that we only
+            # consider 10% gradiations.  This may help us avoid
+            # thundering heards by allowing some randomization among
+            # launchers that are within 10% of each other.
+
+            # Over 100% use the exact value so that we always assign
+            # to the launcher with the smallest backlog.
+            pct = round(pct, 1)
+        return pct
+
+    def doesProviderHaveQuotaForNode(self, provider, node, log):
+        total = self.getProviderQuota(provider).copy()
+        self.log.debug("Provider quota before Zuul: %s", total)
+        total.subtract(self.getQuotaUsed(provider))
+        self.log.debug("Provider quota including Zuul: %s", total)
+        total.subtract(node.quota)
+        self.log.debug("Node required quota: %s", node.quota)
+        return total.nonNegative()

zuul/model.py

@@ -147,6 +147,62 @@ def filter_severity(error_list, errors=True, warnings=True):
             )]
 
 
+class QuotaInformation:
+    def __init__(self, default=0, **kw):
+        '''
+        Initializes the quota information with some values. None values will
+        be initialized with default which will be typically 0 or math.inf
+        indicating an infinite limit.
+
+        :param default: The default value to use for any attribute not supplied
+                        (usually 0 or math.inf).
+        '''
+        self.quota = {}
+        for k, v in kw.items():
+            self.quota[k] = v
+        self.default = default
+
+    def __eq__(self, other):
+        return (isinstance(other, QuotaInformation) and
+                self.default == other.default and
+                self.quota == other.quota)
+
+    def _get_default(self, value, default):
+        return value if value is not None else default
+
+    def _add_subtract(self, other, add=True):
+        for resource in other.quota.keys():
+            self.quota.setdefault(resource, self.default)
+        for resource in self.quota.keys():
+            other_value = other.quota.get(resource, other.default)
+            if add:
+                self.quota[resource] += other_value
+            else:
+                self.quota[resource] -= other_value
+
+    def copy(self):
+        return QuotaInformation(self.default, **self.quota)
+
+    def subtract(self, other):
+        self._add_subtract(other, add=False)
+
+    def add(self, other):
+        self._add_subtract(other, True)
+
+    def nonNegative(self):
+        for resource, value in self.quota.items():
+            if value < 0:
+                return False
+        return True
+
+    def getResources(self):
+        '''Return resources value to register in ZK node'''
+        return self.quota
+
+    def __str__(self):
+        return str(self.quota)
+
+
 class QueryCacheEntry:
     def __init__(self, ltime, results):
         self.ltime = ltime
@@ -2455,11 +2511,22 @@ class ProviderNode(zkobject.PolymorphicZKObjectMixin,
         BUILDING = "building"
         READY = "ready"
         FAILED = "failed"
+        TEMPFAILED = "tempfailed"
         IN_USE = "in-use"
         USED = "used"
         OUTDATED = "outdated"
         HOLD = "hold"
 
+    # States where quota counts
+    ALLOCATED_STATES = (
+        State.BUILDING,
+        State.READY,
+        State.IN_USE,
+        State.USED,
+        State.OUTDATED,
+        State.HOLD,
+    )
+
     FINAL_STATES = (
         State.READY,
         State.FAILED,
@@ -2515,10 +2582,10 @@ class ProviderNode(zkobject.PolymorphicZKObjectMixin,
             region=None,
             username=None,
             hold_expiration=None,
-            resources=None,
             attributes={},
             tenant_name=None,
             host_keys=[],
+            quota=QuotaInformation(),
             # Attributes that are not serialized
             is_locked=False,
             create_state_machine=None,
@@ -2543,6 +2610,12 @@ class ProviderNode(zkobject.PolymorphicZKObjectMixin,
     def getLockPath(self):
         return f"{self.ROOT}/{self.LOCKS_PATH}/{self.uuid}"
 
+    def deserialize(self, raw, context, extra=None):
+        data = super().deserialize(raw, context)
+        resources = data.get('quota') or {}
+        data['quota'] = QuotaInformation(**resources)
+        return data
+
     def serialize(self, context):
         if self.create_state_machine:
             self._set(create_state=self.create_state_machine.toDict())
@@ -2562,6 +2635,7 @@ class ProviderNode(zkobject.PolymorphicZKObjectMixin,
             connection_name=self.connection_name,
             create_state=self.create_state,
             delete_state=self.delete_state,
+            quota=self.quota.getResources(),
             **self.getNodeData(),
             **self.getDriverData(),
         )
@@ -2584,6 +2658,8 @@ class ProviderNode(zkobject.PolymorphicZKObjectMixin,
         return dict()
 
     def getNodeData(self):
+        # This gets stored on the ProviderNode object, but also copied
+        # to the Node object for use by the executor.
         return dict(
             attributes=self.attributes,
             az=self.az,
@@ -2602,7 +2678,7 @@ class ProviderNode(zkobject.PolymorphicZKObjectMixin,
             public_ipv4=self.public_ipv4,
             public_ipv6=self.public_ipv6,
             region=self.region,
-            resources=self.resources,
+            resources=None,
             slot=self.slot,
             tenant_name=self.tenant_name,
             username=self.username,

zuul/provider/__init__.py

@@ -20,7 +20,6 @@ import urllib.parse
 
 from zuul.lib.voluputil import Required, Optional, Nullable, assemble
 from zuul import model
-from zuul.driver.util import QuotaInformation
 from zuul.zk import zkobject
 import zuul.provider.schema as provider_schema
 
@@ -470,7 +469,7 @@ class BaseProvider(zkobject.PolymorphicZKObjectMixin,
         :returns: A :py:class:`QuotaInformation` object.
 
         """
-        return QuotaInformation(default=math.inf)
+        return model.QuotaInformation(default=math.inf)
 
     def getQuotaForLabel(self, label):
         """Return information about the quota used for a label
@@ -484,7 +483,7 @@ class BaseProvider(zkobject.PolymorphicZKObjectMixin,
 
         :returns: A :py:class:`QuotaInformation` object.
         """
-        return QuotaInformation(instances=1)
+        return model.QuotaInformation(instances=1)
 
     def getAZs(self):
         """Return a list of availability zones for this provider

zuul/zk/launcher.py

@@ -12,6 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import collections
 import json
 import logging
 import threading
@@ -20,7 +21,7 @@ from operator import attrgetter
 import mmh3
 from kazoo.exceptions import NoNodeError
 
-from zuul.model import NodesetRequest, ProviderNode
+from zuul.model import NodesetRequest, ProviderNode, QuotaInformation
 from zuul.zk.cache import ZuulTreeCache
 from zuul.zk.zkobject import ZKContext
 
@@ -113,6 +114,43 @@ class LockableZKObjectCache(ZuulTreeCache):
         self.ensureReady()
         return list(self._cached_objects.values())
 
+    def getKeys(self):
+        self.ensureReady()
+        return set(self._cached_objects.keys())
+
+
+class NodeCache(LockableZKObjectCache):
+    def __init__(self, *args, **kw):
+        # Key -> quota, for each cached object
+        self._cached_quota = {}
+        # Provider canonical name -> quota, for each provider
+        self._provider_quota = collections.defaultdict(
+            lambda: QuotaInformation())
+        super().__init__(*args, **kw)
+
+    def postCacheHook(self, event, data, stat, key, obj):
+        # Have we previously added quota for this object?
+        old_quota = self._cached_quota.get(key)
+        if key in self._cached_objects:
+            new_quota = obj.quota
+        else:
+            new_quota = None
+        # Now that we've established whether we should count the quota
+        # based on object presence, take node state into account.
+        if obj is None or obj.state not in obj.ALLOCATED_STATES:
+            new_quota = None
+        if new_quota != old_quota:
+            # Add the new value first so if another thread races these
+            # two operations, it sees us go over quota and not under.
+            if new_quota is not None:
+                self._provider_quota[obj.provider].add(obj.quota)
+            if old_quota is not None:
+                self._provider_quota[obj.provider].subtract(old_quota)
+        super().postCacheHook(event, data, stat, key, obj)
+
+    def getQuota(self, provider):
+        return self._provider_quota[provider.canonical_name]
+
 
 class LauncherApi:
     log = logging.getLogger("zuul.LauncherApi")
@@ -130,7 +168,7 @@ class LauncherApi:
             items_path=NodesetRequest.REQUESTS_PATH,
             locks_path=NodesetRequest.LOCKS_PATH,
             zkobject_class=NodesetRequest)
-        self.nodes_cache = LockableZKObjectCache(
+        self.nodes_cache = NodeCache(
             self.zk_client,
             self.event_callback,
             root=ProviderNode.ROOT,