Don't lock pipelines during local layout update
In our Zuul deployment we can see that the local layout update takes very long on a busy scheduler. We also observed that due to the locking in the post-config hook, we might starve pipelines for the duration of the local layout updates. Since the pipeline state is already reset after the inital reconfiguration we don't need to reset it again after a local layout update. In case the pipeline's layout UUID is already the same as our current one, we can skip that step. Because we no longer restore the pipeline state from ZK in _postConfig, we may exit the _reconfigureTenant method with an out of date copy of the queue contents. In particular, on startup our in-memory queue may be empty. Because of that, remove the statsd emission in that method and rely more heavily on the pipeline manager emitting those stats. The emission is moved to the prime method so that we still initialize the stats to 0 when we add new tenants. This change also removes the relative priority queues from the pipeline state. The relative prio queues were never properly (de-)serialized and shouldn't be stored in ZK at all. This also corrects a use of locks in the prime method. This case is probably impossible to hit with the current code. The lock which is passed to the ZKContext in prime is only used if _reconfigureTenant needs to modify objects in ZK for newly created tenants, but a newly created tenant has no ZK objects that need modifying. Because of that, a test for this code is unfeasible, but let's correct the error anyway. The lock_ctx is a generator, and the result of entering it as a context manager is the actual lock that we need to pass to the ZKContext. Change-Id: I7ad003a8fc24ab9be417a46287390675227e2db3
This commit is contained in:
Simon Westphahl
committed by
James E. Blair
James E. Blair
parent
7f5b49b3d3
commit
46e6a17692
@@ -263,6 +263,131 @@ class TestScaleOutScheduler(ZuulTestCase):
|
||||
holders = tenant.semaphore_handler.semaphoreHolders(semaphore)
|
||||
self.assertEqual(len(holders), 0)
|
||||
|
||||
@simple_layout('layouts/two-projects-integrated.yaml')
|
||||
def test_nodepool_relative_priority_check(self):
|
||||
"Test that nodes are requested at the relative priority"
|
||||
self.fake_nodepool.pause()
|
||||
|
||||
# Start a second scheduler that uses the existing layout
|
||||
app = self.createScheduler()
|
||||
app.start()
|
||||
|
||||
# Hold the lock on the first scheduler so that if any events
|
||||
# happen, they are processed by the second scheduler.
|
||||
with self.scheds.first.sched.run_handler_lock:
|
||||
A = self.fake_gerrit.addFakeChange('org/project', 'master', 'A')
|
||||
self.fake_gerrit.addEvent(A.getPatchsetCreatedEvent(1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
B = self.fake_gerrit.addFakeChange('org/project', 'master', 'B')
|
||||
self.fake_gerrit.addEvent(B.getPatchsetCreatedEvent(1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
C = self.fake_gerrit.addFakeChange('org/project1', 'master', 'C')
|
||||
self.fake_gerrit.addEvent(C.getPatchsetCreatedEvent(1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
D = self.fake_gerrit.addFakeChange('org/project2', 'master', 'D')
|
||||
self.fake_gerrit.addEvent(D.getPatchsetCreatedEvent(1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
reqs = self.fake_nodepool.getNodeRequests()
|
||||
|
||||
# The requests come back sorted by priority.
|
||||
|
||||
# Change A, first change for project, high relative priority.
|
||||
self.assertEqual(reqs[0]['_oid'], '200-0000000000')
|
||||
self.assertEqual(reqs[0]['relative_priority'], 0)
|
||||
|
||||
# Change C, first change for project1, high relative priority.
|
||||
self.assertEqual(reqs[1]['_oid'], '200-0000000002')
|
||||
self.assertEqual(reqs[1]['relative_priority'], 0)
|
||||
|
||||
# Change B, second change for project, lower relative priority.
|
||||
self.assertEqual(reqs[2]['_oid'], '200-0000000001')
|
||||
self.assertEqual(reqs[2]['relative_priority'], 1)
|
||||
|
||||
# Change D, first change for project2 shared with project1,
|
||||
# lower relative priority than project1.
|
||||
self.assertEqual(reqs[3]['_oid'], '200-0000000003')
|
||||
self.assertEqual(reqs[3]['relative_priority'], 1)
|
||||
|
||||
# Fulfill only the first request
|
||||
self.fake_nodepool.fulfillRequest(reqs[0])
|
||||
for x in iterate_timeout(30, 'fulfill request'):
|
||||
reqs = list(self.scheds.first.sched.nodepool.getNodeRequests())
|
||||
if len(reqs) < 4:
|
||||
break
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
reqs = self.fake_nodepool.getNodeRequests()
|
||||
|
||||
# Change B, now first change for project, equal priority.
|
||||
self.assertEqual(reqs[0]['_oid'], '200-0000000001')
|
||||
self.assertEqual(reqs[0]['relative_priority'], 0)
|
||||
|
||||
# Change C, now first change for project1, equal priority.
|
||||
self.assertEqual(reqs[1]['_oid'], '200-0000000002')
|
||||
self.assertEqual(reqs[1]['relative_priority'], 0)
|
||||
|
||||
# Change D, first change for project2 shared with project1,
|
||||
# still lower relative priority than project1.
|
||||
self.assertEqual(reqs[2]['_oid'], '200-0000000003')
|
||||
self.assertEqual(reqs[2]['relative_priority'], 1)
|
||||
|
||||
self.fake_nodepool.unpause()
|
||||
self.waitUntilSettled()
|
||||
|
||||
@simple_layout('layouts/two-projects-integrated.yaml')
|
||||
def test_nodepool_relative_priority_gate(self):
|
||||
"Test that nodes are requested at the relative priority"
|
||||
self.fake_nodepool.pause()
|
||||
|
||||
# Start a second scheduler that uses the existing layout
|
||||
app = self.createScheduler()
|
||||
app.start()
|
||||
|
||||
# Hold the lock on the first scheduler so that if any events
|
||||
# happen, they are processed by the second scheduler.
|
||||
with self.scheds.first.sched.run_handler_lock:
|
||||
A = self.fake_gerrit.addFakeChange('org/project1', 'master', 'A')
|
||||
A.addApproval('Code-Review', 2)
|
||||
self.fake_gerrit.addEvent(A.addApproval('Approved', 1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
B = self.fake_gerrit.addFakeChange('org/project2', 'master', 'B')
|
||||
B.addApproval('Code-Review', 2)
|
||||
self.fake_gerrit.addEvent(B.addApproval('Approved', 1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
# project does not share a queue with project1 and project2.
|
||||
C = self.fake_gerrit.addFakeChange('org/project', 'master', 'C')
|
||||
C.addApproval('Code-Review', 2)
|
||||
self.fake_gerrit.addEvent(C.addApproval('Approved', 1))
|
||||
self.waitUntilSettled(matcher=[app])
|
||||
|
||||
reqs = self.fake_nodepool.getNodeRequests()
|
||||
|
||||
# The requests come back sorted by priority.
|
||||
|
||||
# Change A, first change for shared queue, high relative
|
||||
# priority.
|
||||
self.assertEqual(reqs[0]['_oid'], '100-0000000000')
|
||||
self.assertEqual(reqs[0]['relative_priority'], 0)
|
||||
|
||||
# Change C, first change for independent project, high
|
||||
# relative priority.
|
||||
self.assertEqual(reqs[1]['_oid'], '100-0000000002')
|
||||
self.assertEqual(reqs[1]['relative_priority'], 0)
|
||||
|
||||
# Change B, second change for shared queue, lower relative
|
||||
# priority.
|
||||
self.assertEqual(reqs[2]['_oid'], '100-0000000001')
|
||||
self.assertEqual(reqs[2]['relative_priority'], 1)
|
||||
|
||||
self.fake_nodepool.unpause()
|
||||
self.waitUntilSettled()
|
||||
|
||||
|
||||
class TestSOSCircularDependencies(ZuulTestCase):
|
||||
# Those tests are testing specific interactions between multiple
|
||||
|
||||
@@ -90,25 +90,37 @@ class PipelineManager(metaclass=ABCMeta):
|
||||
self.current_context = None
|
||||
|
||||
def _postConfig(self, layout):
|
||||
# All pipelines support shared queues for setting
|
||||
# relative_priority; only the dependent pipeline uses them for
|
||||
# pipeline queing.
|
||||
# If our layout UUID already matches the UUID in ZK, we don't
|
||||
# need to make any changes in ZK. But we do still need to
|
||||
# update our local object pointers. Note that our local queue
|
||||
# state may still be out of date after this because we skip
|
||||
# the refresh.
|
||||
self.buildChangeQueues(layout)
|
||||
ctx = self.sched.createZKContext(None, self.log)
|
||||
with self.currentContext(ctx):
|
||||
if layout.uuid == PipelineState.peekLayoutUUID(self.pipeline):
|
||||
self.pipeline.state = PipelineState()
|
||||
self.pipeline.state._set(pipeline=self.pipeline)
|
||||
self.pipeline.change_list = PipelineChangeList.create(
|
||||
self.pipeline)
|
||||
return
|
||||
|
||||
with pipeline_lock(
|
||||
self.sched.zk_client, self.pipeline.tenant.name, self.pipeline.name
|
||||
) as lock:
|
||||
ctx = self.sched.createZKContext(lock, self.log)
|
||||
with self.currentContext(ctx):
|
||||
# Since the layout UUID is new, this will move queues
|
||||
# to "old_queues" and refresh the pipeline state as a
|
||||
# side effect.
|
||||
self.pipeline.state = PipelineState.resetOrCreate(
|
||||
self.pipeline, layout.uuid)
|
||||
self.pipeline.change_list = PipelineChangeList.create(
|
||||
self.pipeline)
|
||||
self.buildChangeQueues(layout)
|
||||
|
||||
def buildChangeQueues(self, layout):
|
||||
self.log.debug("Building relative_priority queues")
|
||||
# Note: change_queues is serialized to ZK, so mutate a copy
|
||||
# and then update the attribute when we finish.
|
||||
change_queues = self.pipeline.relative_priority_queues.copy()
|
||||
change_queues = self.pipeline.relative_priority_queues
|
||||
tenant = self.pipeline.tenant
|
||||
layout_project_configs = layout.project_configs
|
||||
|
||||
@@ -145,7 +157,6 @@ class PipelineManager(metaclass=ABCMeta):
|
||||
change_queue.append(project)
|
||||
self.log.debug("Added project %s to queue: %s" %
|
||||
(project, queue_name))
|
||||
self.pipeline.setRelativePriorityQueues(change_queues)
|
||||
|
||||
def getSubmitAllowNeeds(self):
|
||||
# Get a list of code review labels that are allowed to be
|
||||
|
||||
@@ -21,9 +21,6 @@ class IndependentPipelineManager(PipelineManager):
|
||||
changes_merge = False
|
||||
type = 'independent'
|
||||
|
||||
def _postConfig(self, layout):
|
||||
super(IndependentPipelineManager, self)._postConfig(layout)
|
||||
|
||||
def getChangeQueue(self, change, event, existing=None):
|
||||
log = get_annotated_logger(self.log, event)
|
||||
|
||||
|
||||
+18
-5
@@ -20,6 +20,7 @@ import json
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import zlib
|
||||
from functools import total_ordering
|
||||
|
||||
import re2
|
||||
@@ -439,6 +440,7 @@ class Pipeline(object):
|
||||
self.dequeue_on_new_patchset = True
|
||||
self.ignore_dependencies = False
|
||||
self.manager = None
|
||||
self.relative_priority_queues = {}
|
||||
self.precedence = PRECEDENCE_NORMAL
|
||||
self.supercedes = []
|
||||
self.triggers = []
|
||||
@@ -464,10 +466,6 @@ class Pipeline(object):
|
||||
def queues(self):
|
||||
return self.state.queues
|
||||
|
||||
@property
|
||||
def relative_priority_queues(self):
|
||||
return self.state.relative_priority_queues
|
||||
|
||||
@property
|
||||
def actions(self):
|
||||
return (
|
||||
@@ -595,7 +593,6 @@ class PipelineState(zkobject.ZKObject):
|
||||
state=Pipeline.STATE_NORMAL,
|
||||
queues=[],
|
||||
old_queues=[],
|
||||
relative_priority_queues={},
|
||||
consecutive_failures=0,
|
||||
disabled=False,
|
||||
pipeline=None,
|
||||
@@ -612,6 +609,22 @@ class PipelineState(zkobject.ZKObject):
|
||||
obj._load(context, path=path)
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
def peekLayoutUUID(cls, pipeline):
|
||||
ctx = pipeline.manager.current_context
|
||||
try:
|
||||
path = cls.pipelinePath(pipeline)
|
||||
compressed_data, zstat = ctx.client.get(path)
|
||||
try:
|
||||
raw = zlib.decompress(compressed_data)
|
||||
except zlib.error:
|
||||
# Fallback for old, uncompressed data
|
||||
raw = compressed_data
|
||||
data = json.loads(raw.decode("utf8"))
|
||||
return data["layout_uuid"]
|
||||
except NoNodeError:
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def resetOrCreate(cls, pipeline, layout_uuid):
|
||||
ctx = pipeline.manager.current_context
|
||||
|
||||
+18
-16
@@ -790,6 +790,20 @@ class Scheduler(threading.Thread):
|
||||
self.repl.stop()
|
||||
self.repl = None
|
||||
|
||||
def _reportInitialStats(self, tenant):
|
||||
if not self.statsd:
|
||||
return
|
||||
try:
|
||||
for pipeline in tenant.layout.pipelines.values():
|
||||
# stats.gauges.zuul.tenant.<tenant>.pipeline.
|
||||
# <pipeline>.current_changes
|
||||
key = 'zuul.tenant.%s.pipeline.%s' % (
|
||||
tenant.name, pipeline.name)
|
||||
self.statsd.gauge(key + '.current_changes', 0)
|
||||
except Exception:
|
||||
self.log.exception("Exception reporting initial "
|
||||
"pipeline stats:")
|
||||
|
||||
def prime(self, config):
|
||||
self.log.info("Priming scheduler config")
|
||||
start = time.monotonic()
|
||||
@@ -814,13 +828,13 @@ class Scheduler(threading.Thread):
|
||||
# In case we don't have a cached layout state we need to
|
||||
# acquire the write lock since we load a new tenant.
|
||||
if layout_state is None:
|
||||
tlock = tenant_write_lock(self.zk_client, tenant_name)
|
||||
lock_ctx = tenant_write_lock(self.zk_client, tenant_name)
|
||||
else:
|
||||
tlock = tenant_read_lock(self.zk_client, tenant_name)
|
||||
lock_ctx = tenant_read_lock(self.zk_client, tenant_name)
|
||||
|
||||
# Consider all caches valid (min. ltime -1)
|
||||
min_ltimes = defaultdict(lambda: defaultdict(lambda: -1))
|
||||
with tlock:
|
||||
with lock_ctx as tlock:
|
||||
# Refresh the layout state now that we are holding the lock
|
||||
# and we can be sure it won't be changed concurrently.
|
||||
layout_state = self.tenant_layout_state.get(tenant_name)
|
||||
@@ -845,6 +859,7 @@ class Scheduler(threading.Thread):
|
||||
# Reconfigure only tenants w/o an existing layout state
|
||||
ctx = self.createZKContext(tlock, self.log)
|
||||
self._reconfigureTenant(ctx, tenant)
|
||||
self._reportInitialStats(tenant)
|
||||
else:
|
||||
self.local_layout_state[tenant_name] = layout_state
|
||||
self.connections.reconfigureDrivers(tenant)
|
||||
@@ -1481,19 +1496,6 @@ class Scheduler(threading.Thread):
|
||||
self.local_layout_state[tenant.name] = layout_state
|
||||
self.tenant_layout_state[tenant.name] = layout_state
|
||||
|
||||
if self.statsd:
|
||||
try:
|
||||
for pipeline in tenant.layout.pipelines.values():
|
||||
items = len([x for x in pipeline.getAllItems() if x.live])
|
||||
# stats.gauges.zuul.tenant.<tenant>.pipeline.
|
||||
# <pipeline>.current_changes
|
||||
key = 'zuul.tenant.%s.pipeline.%s' % (
|
||||
tenant.name, pipeline.name)
|
||||
self.statsd.gauge(key + '.current_changes', items)
|
||||
except Exception:
|
||||
self.log.exception("Exception reporting initial "
|
||||
"pipeline stats:")
|
||||
|
||||
def _reconfigureDeleteTenant(self, context, tenant):
|
||||
# Called when a tenant is deleted during reconfiguration
|
||||
self.log.info("Removing tenant %s during reconfiguration" %
|
||||
|
||||
Reference in New Issue
Block a user