[Notes] [Git][BuildGrid/buildgrid][raoul/smarter-bot-calls] 3 commits: Smarter bot calls

[Raoul Hidalgo Charman <gitlab mg gitlab com>]

Title: GitLab

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

Commits:

• 1f59cbd0
  by Raoul Hidalgo Charman at 2018-12-04T18:31:07Z

  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.
  
  This is part of #118 and #126
  

• 0221bfc9
  by Raoul Hidalgo Charman at 2018-12-04T18:31:22Z

  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.
  
  This is part of #118 and #126
  

• 4a4e6b73
  by Raoul Hidalgo Charman at 2018-12-04T18:35:51Z

  scheduler.py: Add max time for jobs to block
  
  This stops the issue of grpc calls timeouts defaulting to the max int64 value.
  MAX_JOB_BLOCK_TIME is in the settings file.
  

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

---

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

---

...	...	@@ -24,6 +24,7 @@ import logging
24	24	import grpc
25	25
26	26	from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2, bots_pb2_grpc
	27	+from buildgrid.settings import NETWORK_TIMEOUT
27	28
28	29
29	30	class BotInterface:
...	...	@@ -31,28 +32,32 @@ class BotInterface:
31	32	Interface handles calls to the server.
32	33	"""
33	34
34		- def __init__(self, channel):
	35	+ def __init__(self, channel, interval):
35	36	self.__logger = logging.getLogger(__name__)
36	37
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):
	46	+ """ Creates a bot session returning a grpc StatusCode if it failed """
40	47	request = bots_pb2.CreateBotSessionRequest(parent=parent,
41	48	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
	49	+ return self._bot_call(self._stub.CreateBotSession, request)
48	50
49	51	def update_bot_session(self, bot_session, update_mask=None):
	52	+ """ Updates a bot session returning a grpc StatusCode if it failed """
50	53	request = bots_pb2.UpdateBotSessionRequest(name=bot_session.name,
51	54	bot_session=bot_session,
52	55	update_mask=update_mask)
53		- try:
54		- return self._stub.UpdateBotSession(request)
	56	+ return self._bot_call(self._stub.UpdateBotSession, request)
55	57
	58	+ def _bot_call(self, call, request):
	59	+ try:
	60	+ return call(request, timeout=self.interval + NETWORK_TIMEOUT)
56	61	except grpc.RpcError as e:
57		- self.__logger.error(e)
58		- raise
	62	+ self.__logger.error(e.code())
	63	+ return e.code()

buildgrid/bot/session.py

---

...	...	@@ -19,9 +19,12 @@ Bot Session
19	19
20	20	Allows connections
21	21	"""
	22	+import asyncio
22	23	import logging
23	24	import platform
24	25
	26	+from grpc import StatusCode
	27	+
25	28	from buildgrid._enums import BotStatus, LeaseState
26	29	from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
27	30	from buildgrid._protos.google.rpc import code_pb2
...	...	@@ -47,6 +50,8 @@ class BotSession:
47	50	self._status = BotStatus.OK.value
48	51	self._tenant_manager = TenantManager()
49	52
	53	+ self.connected = False
	54	+
50	55	self.__parent = parent
51	56	self.__bot_id = '{}.{}'.format(parent, platform.node())
52	57	self.__name = None
...	...	@@ -58,10 +63,33 @@ class BotSession:
58	63	def bot_id(self):
59	64	return self.__bot_id
60	65
	66	+ async def run(self):
	67	+ """ Run a bot session
	68	+
	69	+ This connects and reconnects via create bot session and waits on update
	70	+ bot session calls.
	71	+ """
	72	+ self.__logger.debug("Starting bot session")
	73	+ interval = self._bots_interface.interval
	74	+ while True:
	75	+ if not self.connected:
	76	+ self.create_bot_session()
	77	+ else:
	78	+ self.update_bot_session()
	79	+
	80	+ if not self.connected:
	81	+ await asyncio.sleep(interval)
	82	+ else:
	83	+ await self._tenant_manager.wait_on_tenants(interval)
	84	+
61	85	def create_bot_session(self):
62	86	self.__logger.debug("Creating bot session")
63	87
64	88	session = self._bots_interface.create_bot_session(self.__parent, self.get_pb2())
	89	+ if session in list(StatusCode):
	90	+ self.connected = False
	91	+ return
	92	+ self.connected = True
65	93	self.__name = session.name
66	94
67	95	self.__logger.info("Created bot session with name: [%s]", self.__name)
...	...	@@ -73,6 +101,13 @@ class BotSession:
73	101	self.__logger.debug("Updating bot session: [%s]", self.__bot_id)
74	102
75	103	session = self._bots_interface.update_bot_session(self.get_pb2())
	104	+ if session == StatusCode.DEADLINE_EXCEEDED:
	105	+ # try to continue to do update session if it passed the timeout
	106	+ return
	107	+ elif session in StatusCode:
	108	+ self.connected = False
	109	+ return
	110	+ self.connected = True
76	111	server_ids = []
77	112
78	113	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 NETWORK_TIMEOUT
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 - NETWORK_TIMEOUT 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

---

...	...	@@ -25,6 +25,7 @@ from queue import Queue, Empty
25	25
26	26	from buildgrid._enums import LeaseState, OperationStage
27	27	from buildgrid._exceptions import NotFoundError
	28	+from buildgrid.settings import MAX_JOB_BLOCK_TIME
28	29
29	30
30	31	class Scheduler:
...	...	@@ -134,11 +135,15 @@ class Scheduler:
134	135	worker_capabilities (dict): a set of key-value pairs decribing the
135	136	worker properties, configuration and state at the time of the
136	137	request.
137		- timeout (int): time to block waiting on job queue
	138	+ timeout (int): time to block waiting on job queue, caps if longer
	139	+ than MAX_JOB_BLOCK_TIME
138	140	"""
139	141	if not timeout and self.queue.empty():
140	142	return []
141	143
	144	+ # Cap the timeout if it's larger than MAX_JOB_BLOCK_TIME
	145	+ if timeout:
	146	+ timeout = timeout if timeout < MAX_JOB_BLOCK_TIME else MAX_JOB_BLOCK_TIME
142	147	try:
143	148	job = (self.queue.get(block=True, timeout=timeout)
144	149	if timeout

buildgrid/settings.py

---

...	...	@@ -30,3 +30,10 @@ MAX_REQUEST_SIZE = 2 * 1024 * 1024
30	30
31	31	# Maximum number of elements per gRPC request:
32	32	MAX_REQUEST_COUNT = 500
	33	+
	34	+# time in seconds to pad timeouts
	35	+NETWORK_TIMEOUT = 5
	36	+
	37	+# Hard limit for waiting on jobs, avoids grpc timeouts not being set defaulting
	38	+# the interval to the max int64 value
	39	+MAX_JOB_BLOCK_TIME = 300

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.