Ansible launcher: delay node assignment under load
Gearman wakes all available workers at the same time when a job
is available. The first one to respond gets the job. To attempt
to more evenly distribute nodes (which are assigned via a gearman
job) across multiple zuul launchers, delay for a period
related exponentially to the number of nodes this launcher currently
has between the wake up and grab job packets.
Change-Id: I307938f97b730b229c1622cd2f929fc5b65ccdad
diff --git a/zuul/launcher/ansiblelaunchserver.py b/zuul/launcher/ansiblelaunchserver.py
index 00a01d5..1457665 100644
--- a/zuul/launcher/ansiblelaunchserver.py
+++ b/zuul/launcher/ansiblelaunchserver.py
@@ -51,7 +51,19 @@
return bool(x)
-class GearWorker(gear.Worker):
+class LaunchGearWorker(gear.Worker):
+ def __init__(self, *args, **kw):
+ self.__launch_server = kw.pop('launch_server')
+ super(LaunchGearWorker, self).__init__(*args, **kw)
+
+ def handleNoop(self, packet):
+ workers = len(self.__launch_server.node_workers)
+ delay = (workers ** 2) / 1000.0
+ time.sleep(delay)
+ return super(LaunchGearWorker, self).handleNoop(packet)
+
+
+class NodeGearWorker(gear.Worker):
MASS_DO = 101
def sendMassDo(self, functions):
@@ -203,7 +215,8 @@
port = self.config.get('gearman', 'port')
else:
port = 4730
- self.worker = gear.Worker('Zuul Launch Server')
+ self.worker = LaunchGearWorker('Zuul Launch Server',
+ launch_server=self)
self.worker.addServer(server, port)
self.log.debug("Waiting for server")
self.worker.waitForServer()
@@ -533,7 +546,7 @@
port = self.config.get('gearman', 'port')
else:
port = 4730
- self.worker = GearWorker(self.name)
+ self.worker = NodeGearWorker(self.name)
self.worker.addServer(server, port)
self.log.debug("Waiting for server")
self.worker.waitForServer()