[Notes] [Git][BuildGrid/buildgrid][raoul/smarter-bot-calls] 3 commits: scheduler.py: allow requests to block until a job appears

[Raoul Hidalgo Charman <gitlab mg gitlab com>]

Title: GitLab

Raoul Hidalgo Charman pushed to branch raoul/smarter-bot-calls at BuildGrid / buildgrid

Commits:

• 1c027d41
  by Raoul Hidalgo Charman at 2018-11-30T11:37:28Z

  scheduler.py: allow requests to block until a job appears
  
  This uses the Queue rather than dequeue, allowing request_job_leases to block
  until a specified timeout.
  

• 5f8fcd59
  by Raoul Hidalgo Charman at 2018-11-30T11:37:51Z

  Smarter bot calls
  
  Use grpc time outs and ability to wait on jobs to have the server hang pending
  on a job coming through. This reduces the need for constant update requests
  while still responding to new jobs quickly.
  

• 9768067b
  by Raoul Hidalgo Charman at 2018-11-30T11:49:57Z

  bot: add reconnects and waiting on job status
  
  Bot will periodically attempt to reconnect if a controller goes down.
  If a job is running it will similar to the server side, wait on the job
  completing else respond periodically.
  
  This logic may need changing if/when bots can have multiple leases.
  

12 changed files:

• buildgrid/_app/commands/cmd_bot.py
• buildgrid/bot/bot.py
• buildgrid/bot/interface.py
• buildgrid/bot/session.py
• buildgrid/bot/tenantmanager.py
• buildgrid/server/bots/instance.py
• buildgrid/server/bots/service.py
• buildgrid/server/scheduler.py
• buildgrid/settings.py
• tests/integration/bot_session.py
• tests/integration/bots_service.py
• tests/utils/bots_interface.py

Changes:

buildgrid/_app/commands/cmd_bot.py

---

...	...	@@ -31,7 +31,7 @@ from buildgrid.bot import bot, interface, session
31	31	from buildgrid.bot.hardware.interface import HardwareInterface
32	32	from buildgrid.bot.hardware.device import Device
33	33	from buildgrid.bot.hardware.worker import Worker
34		-
	34	+from buildgrid.settings import INTERVAL_BUFFER
35	35
36	36	from ..bots import buildbox, dummy, host
37	37	from ..cli import pass_context
...	...	@@ -54,7 +54,7 @@ from ..cli import pass_context
54	54	help="Public CAS client certificate for TLS (PEM-encoded)")
55	55	@click.option('--cas-server-cert', type=click.Path(exists=True, dir_okay=False), default=None,
56	56	help="Public CAS server certificate for TLS (PEM-encoded)")
57		-@click.option('--update-period', type=click.FLOAT, default=0.5, show_default=True,
	57	+@click.option('--update-period', type=click.FLOAT, default=30, show_default=True,
58	58	help="Time period for bot updates to the server in seconds.")
59	59	@click.option('--parent', type=click.STRING, default='main', show_default=True,
60	60	help="Targeted farm resource.")
...	...	@@ -66,7 +66,6 @@ def cli(context, parent, update_period, remote, client_key, client_cert, server_
66	66
67	67	context.remote = '{}:{}'.format(url.hostname, url.port or 50051)
68	68	context.remote_url = remote
69		- context.update_period = update_period
70	69	context.parent = parent
71	70
72	71	if url.scheme == 'http':
...	...	@@ -124,7 +123,7 @@ def cli(context, parent, update_period, remote, client_key, client_cert, server_
124	123
125	124	click.echo("Starting for remote=[{}]".format(context.remote))
126	125
127		- bot_interface = interface.BotInterface(context.channel)
	126	+ bot_interface = interface.BotInterface(context.channel, update_period + INTERVAL_BUFFER)
128	127	worker = Worker()
129	128	worker.add_device(Device())
130	129	hardware_interface = HardwareInterface(worker)
...	...	@@ -142,7 +141,7 @@ def run_dummy(context):
142	141	try:
143	142	bot_session = session.BotSession(context.parent, context.bot_interface, context.hardware_interface,
144	143	dummy.work_dummy, context)
145		- b = bot.Bot(bot_session, context.update_period)
	144	+ b = bot.Bot(bot_session)
146	145	b.session()
147	146	except KeyboardInterrupt:
148	147	pass
...	...	@@ -158,7 +157,7 @@ def run_host_tools(context):
158	157	try:
159	158	bot_session = session.BotSession(context.parent, context.bot_interface, context.hardware_interface,
160	159	host.work_host_tools, context)
161		- b = bot.Bot(bot_session, context.update_period)
	160	+ b = bot.Bot(bot_session)
162	161	b.session()
163	162	except KeyboardInterrupt:
164	163	pass
...	...	@@ -180,7 +179,7 @@ def run_buildbox(context, local_cas, fuse_dir):
180	179	try:
181	180	bot_session = session.BotSession(context.parent, context.bot_interface, context.hardware_interface,
182	181	buildbox.work_buildbox, context)
183		- b = bot.Bot(bot_session, context.update_period)
	182	+ b = bot.Bot(bot_session)
184	183	b.session()
185	184	except KeyboardInterrupt:
186	185	pass

buildgrid/bot/bot.py

---

...	...	@@ -20,14 +20,12 @@ import logging
20	20	class Bot:
21	21	"""Creates a local BotSession."""
22	22
23		- def __init__(self, bot_session, update_period=1):
	23	+ def __init__(self, bot_session):
24	24	"""
25	25	"""
26	26	self.__logger = logging.getLogger(__name__)
27	27
28	28	self.__bot_session = bot_session
29		- self.__update_period = update_period
30		-
31	29	self.__loop = None
32	30
33	31	def session(self):
...	...	@@ -37,7 +35,7 @@ class Bot:
37	35	self.__bot_session.create_bot_session()
38	36
39	37	try:
40		- task = asyncio.ensure_future(self.__update_bot_session())
	38	+ task = asyncio.ensure_future(self.__bot_session.run())
41	39	self.__loop.run_until_complete(task)
42	40
43	41	except KeyboardInterrupt:
...	...	@@ -46,16 +44,6 @@ class Bot:
46	44	self.__kill_everyone()
47	45	self.__logger.info("Bot shutdown.")
48	46
49		- async def __update_bot_session(self):
50		- """Calls the server periodically to inform the server the client has not died."""
51		- try:
52		- while True:
53		- self.__bot_session.update_bot_session()
54		- await asyncio.sleep(self.__update_period)
55		-
56		- except asyncio.CancelledError:
57		- pass
58		-
59	47	def __kill_everyone(self):
60	48	"""Cancels and waits for them to stop."""
61	49	self.__logger.info("Cancelling remaining tasks...")

buildgrid/bot/interface.py

---

...	...	@@ -31,28 +31,31 @@ class BotInterface:
31	31	Interface handles calls to the server.
32	32	"""
33	33
34		- def __init__(self, channel):
	34	+ def __init__(self, channel, interval):
35	35	self.__logger = logging.getLogger(__name__)
36	36
	37	+ self.__logger.info(channel)
37	38	self._stub = bots_pb2_grpc.BotsStub(channel)
	39	+ self.__interval = interval
	40	+
	41	+ @property
	42	+ def interval(self):
	43	+ return self.__interval
38	44
39	45	def create_bot_session(self, parent, bot_session):
40	46	request = bots_pb2.CreateBotSessionRequest(parent=parent,
41	47	bot_session=bot_session)
42		- try:
43		- return self._stub.CreateBotSession(request)
44		-
45		- except grpc.RpcError as e:
46		- self.__logger.error(e)
47		- raise
	48	+ return self._bot_call(self._stub.CreateBotSession, request)
48	49
49	50	def update_bot_session(self, bot_session, update_mask=None):
50	51	request = bots_pb2.UpdateBotSessionRequest(name=bot_session.name,
51	52	bot_session=bot_session,
52	53	update_mask=update_mask)
53		- try:
54		- return self._stub.UpdateBotSession(request)
	54	+ return self._bot_call(self._stub.UpdateBotSession, request)
55	55
	56	+ def _bot_call(self, call, request):
	57	+ try:
	58	+ return call(request, timeout=self.interval)
56	59	except grpc.RpcError as e:
57		- self.__logger.error(e)
58		- raise
	60	+ self.__logger.error(e.code())
	61	+ return None

buildgrid/bot/session.py

---

...	...	@@ -19,6 +19,7 @@ Bot Session
19	19
20	20	Allows connections
21	21	"""
	22	+import asyncio
22	23	import logging
23	24	import platform
24	25
...	...	@@ -47,6 +48,8 @@ class BotSession:
47	48	self._status = BotStatus.OK.value
48	49	self._tenant_manager = TenantManager()
49	50
	51	+ self.connected = False
	52	+
50	53	self.__parent = parent
51	54	self.__bot_id = '{}.{}'.format(parent, platform.node())
52	55	self.__name = None
...	...	@@ -58,10 +61,33 @@ class BotSession:
58	61	def bot_id(self):
59	62	return self.__bot_id
60	63
	64	+ async def run(self):
	65	+ """ Run a bot session
	66	+
	67	+ This connects and reconnects via create bot session and waits on update
	68	+ bot session calls.
	69	+ """
	70	+ self.__logger.debug("Starting bot session")
	71	+ interval = self._bots_interface.interval
	72	+ while True:
	73	+ if not self.connected:
	74	+ self.create_bot_session()
	75	+ else:
	76	+ self.update_bot_session()
	77	+
	78	+ if not self.connected:
	79	+ await asyncio.sleep(interval)
	80	+ else:
	81	+ await self._tenant_manager.wait_on_tenants(interval)
	82	+
61	83	def create_bot_session(self):
62	84	self.__logger.debug("Creating bot session")
63	85
64	86	session = self._bots_interface.create_bot_session(self.__parent, self.get_pb2())
	87	+ if not session:
	88	+ self.connected = False
	89	+ return
	90	+ self.connected = True
65	91	self.__name = session.name
66	92
67	93	self.__logger.info("Created bot session with name: [%s]", self.__name)
...	...	@@ -73,6 +99,10 @@ class BotSession:
73	99	self.__logger.debug("Updating bot session: [%s]", self.__bot_id)
74	100
75	101	session = self._bots_interface.update_bot_session(self.get_pb2())
	102	+ if not session:
	103	+ self.connected = False
	104	+ return
	105	+ self.connected = True
76	106	server_ids = []
77	107
78	108	for lease in session.leases:

buildgrid/bot/tenantmanager.py

---

...	...	@@ -150,6 +150,13 @@ class TenantManager:
150	150	"""
151	151	return self._tenants[lease_id].tenant_completed
152	152
	153	+ async def wait_on_tenants(self, timeout):
	154	+ if self._tasks:
	155	+ tasks = self._tasks.values()
	156	+ await asyncio.wait(tasks,
	157	+ timeout=timeout,
	158	+ return_when=asyncio.FIRST_COMPLETED)
	159	+
153	160	def _update_lease_result(self, lease_id, result):
154	161	"""Updates the lease with the result."""
155	162	self._tenants[lease_id].update_lease_result(result)

buildgrid/server/bots/instance.py

---

...	...	@@ -24,6 +24,7 @@ import logging
24	24	import uuid
25	25
26	26	from buildgrid._exceptions import InvalidArgumentError
	27	+from buildgrid.settings import INTERVAL_BUFFER
27	28
28	29	from ..job import LeaseState
29	30
...	...	@@ -75,7 +76,7 @@ class BotsInterface:
75	76	self._request_leases(bot_session)
76	77	return bot_session
77	78
78		- def update_bot_session(self, name, bot_session):
	79	+ def update_bot_session(self, name, bot_session, deadline=None):
79	80	""" Client updates the server. Any changes in state to the Lease should be
80	81	registered server side. Assigns available leases with work.
81	82	"""
...	...	@@ -93,14 +94,15 @@ class BotsInterface:
93	94	pass
94	95	lease.Clear()
95	96
96		- self._request_leases(bot_session)
	97	+ self._request_leases(bot_session, deadline)
97	98	return bot_session
98	99
99		- def _request_leases(self, bot_session):
	100	+ def _request_leases(self, bot_session, deadline=None):
100	101	# TODO: Send worker capabilities to the scheduler!
101	102	# Only send one lease at a time currently.
102	103	if not bot_session.leases:
103		- leases = self._scheduler.request_job_leases({})
	104	+ leases = self._scheduler.request_job_leases(
	105	+ {}, timeout=deadline - INTERVAL_BUFFER if deadline else None)
104	106	if leases:
105	107	for lease in leases:
106	108	self._assigned_leases[bot_session.name].add(lease.id)

buildgrid/server/bots/service.py

---

...	...	@@ -138,8 +138,10 @@ class BotsService(bots_pb2_grpc.BotsServicer):
138	138	instance_name = '/'.join(names[:-1])
139	139
140	140	instance = self._get_instance(instance_name)
141		- bot_session = instance.update_bot_session(request.name,
142		- request.bot_session)
	141	+ bot_session = instance.update_bot_session(
	142	+ request.name,
	143	+ request.bot_session,
	144	+ deadline=context.time_remaining())
143	145
144	146	if self._is_instrumented:
145	147	self.__bots[bot_id].GetCurrentTime()

buildgrid/server/scheduler.py

---

...	...	@@ -19,9 +19,9 @@ Scheduler
19	19	Schedules jobs.
20	20	"""
21	21
22		-from collections import deque
23	22	from datetime import timedelta
24	23	import logging
	24	+from queue import Queue, Empty
25	25
26	26	from buildgrid._enums import LeaseState, OperationStage
27	27	from buildgrid._exceptions import NotFoundError
...	...	@@ -41,7 +41,7 @@ class Scheduler:
41	41
42	42	self._action_cache = action_cache
43	43	self.jobs = {}
44		- self.queue = deque()
	44	+ self.queue = Queue()
45	45
46	46	self._is_instrumented = monitor
47	47
...	...	@@ -93,7 +93,7 @@ class Scheduler:
93	93	action_result = self._action_cache.get_action_result(job.action_digest)
94	94	except NotFoundError:
95	95	operation_stage = OperationStage.QUEUED
96		- self.queue.append(job)
	96	+ self.queue.put(job)
97	97
98	98	else:
99	99	job.set_cached_result(action_result)
...	...	@@ -104,7 +104,7 @@ class Scheduler:
104	104
105	105	else:
106	106	operation_stage = OperationStage.QUEUED
107		- self.queue.append(job)
	107	+ self.queue.put(job)
108	108
109	109	self._update_job_operation_stage(job.name, operation_stage)
110	110
...	...	@@ -120,25 +120,31 @@ class Scheduler:
120	120	else:
121	121	operation_stage = OperationStage.QUEUED
122	122	job.update_lease_state(LeaseState.PENDING)
123		- self.queue.append(job)
	123	+ self.queue.put(job)
124	124
125	125	self._update_job_operation_stage(job_name, operation_stage)
126	126
127	127	def list_jobs(self):
128	128	return self.jobs.values()
129	129
130		- def request_job_leases(self, worker_capabilities):
	130	+ def request_job_leases(self, worker_capabilities, timeout=None):
131	131	"""Generates a list of the highest priority leases to be run.
132	132
133	133	Args:
134	134	worker_capabilities (dict): a set of key-value pairs decribing the
135	135	worker properties, configuration and state at the time of the
136	136	request.
	137	+ timeout (int): time to block waiting on job queue
137	138	"""
138		- if not self.queue:
	139	+ if not timeout and self.queue.empty():
139	140	return []
140	141
141		- job = self.queue.popleft()
	142	+ try:
	143	+ job = (self.queue.get(block=True, timeout=timeout)
	144	+ if timeout
	145	+ else self.queue.get(block=False))
	146	+ except Empty:
	147	+ return []
142	148
143	149	lease = job.lease
144	150

buildgrid/settings.py

---

...	...	@@ -24,3 +24,6 @@ HASH_LENGTH = HASH().digest_size * 2
24	24
25	25	# Period, in seconds, for the monitoring cycle:
26	26	MONITORING_PERIOD = 5.0
	27	+
	28	+# time in seconds to pad timeouts
	29	+INTERVAL_BUFFER = 5

tests/integration/bot_session.py

---

...	...	@@ -30,6 +30,7 @@ from ..utils.utils import run_in_subprocess
30	30	from ..utils.bots_interface import serve_bots_interface
31	31
32	32
	33	+TIMEOUT = 5
33	34	INSTANCES = ['', 'instance']
34	35
35	36
...	...	@@ -48,7 +49,7 @@ class ServerInterface:
48	49	bot_session = bots_pb2.BotSession()
49	50	bot_session.ParseFromString(string_bot_session)
50	51
51		- interface = BotInterface(grpc.insecure_channel(remote))
	52	+ interface = BotInterface(grpc.insecure_channel(remote), TIMEOUT)
52	53
53	54	result = interface.create_bot_session(parent, bot_session)
54	55	queue.put(result.SerializeToString())
...	...	@@ -67,7 +68,7 @@ class ServerInterface:
67	68	bot_session = bots_pb2.BotSession()
68	69	bot_session.ParseFromString(string_bot_session)
69	70
70		- interface = BotInterface(grpc.insecure_channel(remote))
	71	+ interface = BotInterface(grpc.insecure_channel(remote), TIMEOUT)
71	72
72	73	result = interface.update_bot_session(bot_session, update_mask)
73	74	queue.put(result.SerializeToString())

tests/integration/bots_service.py

---

...	...	@@ -38,7 +38,9 @@ server = mock.create_autospec(grpc.server)
38	38	# GRPC context
39	39	@pytest.fixture
40	40	def context():
41		- yield mock.MagicMock(spec=_Context)
	41	+ context_mock = mock.MagicMock(spec=_Context)
	42	+ context_mock.time_remaining.return_value = None
	43	+ yield context_mock
42	44
43	45
44	46	@pytest.fixture

tests/utils/bots_interface.py

---

...	...	@@ -123,7 +123,7 @@ class BotsInterface:
123	123	self.__bot_session_queue.put(bot_session.SerializeToString())
124	124	return bot_session
125	125
126		- def update_bot_session(self, name, bot_session):
	126	+ def update_bot_session(self, name, bot_session, deadline=None):
127	127	for lease in bot_session.leases:
128	128	state = LeaseState(lease.state)
129	129	if state == LeaseState.COMPLETED:

—
View it on GitLab.
You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.