[Notes] [Git][BuildGrid/buildgrid][finn/74-operation-cancelation] Working commit

[finn <gitlab mg gitlab com>]

Title: GitLab

finn pushed to branch finn/74-operation-cancelation at BuildGrid / buildgrid

Commits:

• 96e070f8
  by Finn at 2018-10-31T17:44:04Z

  Working commit
  

9 changed files:

• buildgrid/_app/commands/cmd_bot.py
• buildgrid/bot/bot.py
• buildgrid/bot/hardware/device.py
• + buildgrid/bot/hardware/interface.py
• buildgrid/bot/hardware/worker.py
• buildgrid/bot/session.py
• + buildgrid/bot/tenant.py
• + buildgrid/bot/tenantmanager.py
• buildgrid/server/bots/instance.py

Changes:

buildgrid/_app/commands/cmd_bot.py

---

...	...	@@ -28,8 +28,7 @@ from urllib.parse import urlparse
28	28	import click
29	29	import grpc
30	30
31		-from buildgrid.bot import bot, bot_interface
32		-from buildgrid.bot.bot_session import BotSession, Device, Worker
	31	+from buildgrid.bot import bot, interface, session, hardware
33	32
34	33	from ..bots import buildbox, dummy, host
35	34	from ..cli import pass_context

buildgrid/bot/bot.py

---

...	...	@@ -17,7 +17,7 @@
17	17	Bot
18	18	====
19	19
20		-Creates a bot session.
	20	+Creates a bot session and sends updates to the server.
21	21	"""
22	22
23	23	import asyncio
...	...	@@ -45,6 +45,7 @@ class Bot:
45	45	loop.run_forever()
46	46	except KeyboardInterrupt:
47	47	pass
	48	+
48	49	finally:
49	50	task.cancel()
50	51	loop.close()

buildgrid/bot/hardware/device.py

---

...	...	@@ -13,6 +13,15 @@
13	13	# limitations under the License.
14	14
15	15
	16	+"""
	17	+Device
	18	+======
	19	+
	20	+A device.
	21	+"""
	22	+
	23	+from buildgrid._protos.google.devtools.remoteworkers.v1test2 import worker_pb2
	24	+
16	25	class Device:
17	26
18	27	def __init__(self, properties=None):
...	...	@@ -21,23 +30,17 @@ class Device:
21	30	is running a bit and responsible to actually executing commands.
22	31	All other devices are known as Attatched Devices and must be controlled
23	32	by the Primary Device.
	33	+
	34	+ properties (list(dict(string : string))) : Properties of device. Keys may
	35	+ repeated.
24	36	"""
25	37
26		- self.__properties = {}
	38	+ self._properties = {}
	39	+ self.__property_keys = ['os', 'has-docker']
27	40	self.__name = str(uuid.uuid4())
28	41
29		- if properties:
30		- for k, v in properties.items():
31		- if k == 'os':
32		- self.__properties[k] = v
33		-
34		- elif k == 'docker':
35		- if v not in ('True', 'False'):
36		- raise ValueError('Value not supported: [{}]'.format(v))
37		- self.__properties[k] = v
38		-
39		- else:
40		- raise KeyError('Key not supported: [{}]'.format(k))
	42	+ for prop in properties:
	43	+ self._add_property(prop)
41	44
42	45	@property
43	46	def name(self):
...	...	@@ -45,13 +48,25 @@ class Device:
45	48
46	49	@property
47	50	def properties(self):
48		- return self.__properties
	51	+ return self._properties
49	52
50	53	def get_pb2(self):
51	54	device = worker_pb2.Device(handle=self._name)
52		- property_message = worker_pb2.Device.Property()
53	55	for k, v in self._properties.items():
54		- property_message.key = k
55		- property_message.value = v
56		- device.properties.extend([property_message])
	56	+ for prop in v:
	57	+ property_message = worker_pb2.Device.Property()
	58	+ property_message.key = k
	59	+ property_message.value = prop
	60	+ device.properties.extend([property_message])
57	61	return device
	62	+
	63	+ def _add_property(self, key, value):
	64	+ if key in self.__property_keys:
	65	+ prop = self._properties.get(key)
	66	+ if not prop:
	67	+ self._properties[key] = [value]
	68	+ else:
	69	+ prop[key].append(value)
	70	+
	71	+ else:
	72	+ raise KeyError('Key not supported: [{}]'.format(key))

buildgrid/bot/hardware/interface.py

---

	1	+# Copyright (C) 2018 Bloomberg LP
	2	+#
	3	+# Licensed under the Apache License, Version 2.0 (the "License");
	4	+# you may not use this file except in compliance with the License.
	5	+# You may obtain a copy of the License at
	6	+#
	7	+# <http://www.apache.org/licenses/LICENSE-2.0>
	8	+#
	9	+# Unless required by applicable law or agreed to in writing, software
	10	+# distributed under the License is distributed on an "AS IS" BASIS,
	11	+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	12	+# See the License for the specific language governing permissions and
	13	+# limitations under the License.
	14	+
	15	+
	16	+"""
	17	+HardwareInterface
	18	+=================
	19	+
	20	+Class to configure hardware and check requirements of leases.
	21	+
	22	+In the future this could also be used to request and display
	23	+the status of hardware.
	24	+"""
	25	+
	26	+
	27	+from buildgrid._exceptions import FailedPreconditionError
	28	+
	29	+
	30	+class HardwareInterface:
	31	+
	32	+ def __init__(self, worker):
	33	+ self._worker = worker
	34	+
	35	+ def configure_hardware(self, lease):
	36	+ """ Can check if the requirements can be met and also
	37	+ in the future, potentially configure the hardware.
	38	+ """
	39	+ worker = self._worker
	40	+ worker_requirements = lease.worker
	41	+
	42	+ for config_requirement in worker_requirements.configs:
	43	+ if config_requirement.key not in worker.configs:
	44	+ raise FailedPreconditionError("Config not supported: [{}]".format(config_requirement))
	45	+
	46	+ def get_worker_pb2(self):
	47	+ return self._worker.get_pb2()

buildgrid/bot/hardware/worker.py

---

...	...	@@ -13,39 +13,34 @@
13	13	# limitations under the License.
14	14
15	15
	16	+from buildgrid._protos.google.devtools.remoteworkers.v1test2 import worker_pb2
	17	+
	18	+
16	19	class Worker:
17	20
18	21	def __init__(self, properties=None, configs=None):
19	22	self._devices = []
20		- self.__configs = {}
21		- self.__properties = {}
	23	+ self._configs = {}
	24	+ self._properties = {}
	25	+ self.__property_keys = ['pool']
	26	+ self.__config_keys = ['DockerImage']
22	27
23	28	if properties:
24	29	for k, v in properties.items():
25		- self.update_config(k, v)
	30	+ if k in self.__property_keys:
	31	+ self._add_properties(k, v)
26	32
27	33	if configs:
28	34	for k, v in configs.items():
29		- if k == 'DockerImage':
30		- self.__configs[k] = v
31		- else:
32		- raise KeyError('Key not supported: [{}]'.format(k))
	35	+ self._add_config(k, v)
33	36
34	37	@property
35	38	def configs(self):
36		- return self.__configs
	39	+ return self._configs
37	40
38	41	@property
39	42	def properties(self):
40		- return self.__properties
41		-
42		- def update_config(self, key, value):
43		- """Update the config of the device.
44		- Only certain keys are allowed."""
45		- if k == 'pool':
46		- self.__properties[k] = v
47		- else:
48		- raise KeyError('Key not supported: [{}]'.format(k))
	43	+ return self._properties
49	44
50	45	def add_device(self, device):
51	46	self._devices.append(device)
...	...	@@ -53,16 +48,41 @@ class Worker:
53	48	def get_pb2(self):
54	49	devices = [device.get_pb2() for device in self._devices]
55	50	worker = worker_pb2.Worker(devices=devices)
56		- property_message = worker_pb2.Worker.Property()
57		- for k, v in self.__properties.items():
58		- property_message.key = k
59		- property_message.value = v
60		- worker.properties.extend([property_message])
61		-
62		- config_message = worker_pb2.Worker.Config()
63		- for k, v in self.__properties.items():
64		- property_message.key = k
65		- property_message.value = v
66		- worker.configs.extend([config_message])
	51	+
	52	+ for k, v in self._properties.items():
	53	+ for prop in v:
	54	+ property_message = worker_pb2.Device.Property()
	55	+ property_message.key = k
	56	+ property_message.value = prop
	57	+ device.properties.extend([property_message])
	58	+
	59	+ for k, v in self._configs.items():
	60	+ for cfg in v:
	61	+ config_message = worker_pb2.Worker.Config()
	62	+ config.key = k
	63	+ config_message.value = cfg
	64	+ worker.configs.extend([config_message])
67	65
68	66	return worker
	67	+
	68	+ def _add_config(self, key, value):
	69	+ if key in self.__config_keys:
	70	+ cfg = self._configs.get(key)
	71	+ if not cfg:
	72	+ self._configs[key] = [value]
	73	+ else:
	74	+ cfg[key].append(value)
	75	+
	76	+ else:
	77	+ raise KeyError('Key not supported: [{}]'.format(key))
	78	+
	79	+ def _add_property(self, key, value):
	80	+ if key in self.__property_keys:
	81	+ prop = self._properties.get(key)
	82	+ if not prop:
	83	+ self._properties[key] = [value]
	84	+ else:
	85	+ prop[key].append(value)
	86	+
	87	+ else:
	88	+ raise KeyError('Key not supported: [{}]'.format(key))

buildgrid/bot/session.py

---

...	...	@@ -31,12 +31,12 @@ import grpc
31	31
32	32	from buildgrid._enums import BotStatus, LeaseState
33	33	from buildgrid._protos.google.rpc import code_pb2
34		-from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2, worker_pb2
35	34	from buildgrid._exceptions import BotError
36	35
	36	+from buildgrid._exceptions import FailedPreconditionError
37	37
38	38	class BotSession:
39		- def __init__(self, parent, interface, worker):
	39	+ def __init__(self, parent, bots_interface, hardware_interface):
40	40	""" Unique bot ID within the farm used to identify this bot
41	41	Needs to be human readable.
42	42	All prior sessions with bot_id of same ID are invalidated.
...	...	@@ -45,14 +45,14 @@ class BotSession:
45	45	"""
46	46
47	47	self.logger = logging.getLogger(__name__)
48		- self._context = None
49	48
50		- self._worker = worker
51		- self._interface = interface
52		- self._leases = {}
53		- self._parent = parent
	49	+ self._bots_interface = bots_interface
	50	+ self._hardware_interface_interface = hardware_interface
	51	+
54	52	self._status = BotStatus.OK.value
	53	+ self._tenant_manager = TenantManager()
55	54
	55	+ self.__parent = parent
56	56	self.__bot_id = '{}.{}'.format(parent, platform.node())
57	57	self.__name = None
58	58
...	...	@@ -60,10 +60,8 @@ class BotSession:
60	60	def bot_id(self):
61	61	return self.__bot_id
62	62
63		- def create_bot_session(self, work, context=None):
	63	+ def create_bot_session(self):
64	64	self.logger.debug("Creating bot session")
65		- self._work = work
66		- self._context = context
67	65
68	66	session = self._interface.create_bot_session(self._parent, self.get_pb2())
69	67	self.__name = session.name
...	...	@@ -71,62 +69,49 @@ class BotSession:
71	69	self.logger.info("Created bot session with name: [{}]".format(self.__name))
72	70
73	71	for lease in session.leases:
74		- self._update_lease_from_server(lease)
	72	+ self._register_lease(lease)
75	73
76	74	def update_bot_session(self):
77	75	self.logger.debug("Updating bot session: [{}]".format(self.__bot_id))
	76	+
78	77	session = self._interface.update_bot_session(self.get_pb2())
79		- for k, v in list(self._leases.items()):
80		- if v.state == LeaseState.COMPLETED.value:
81		- del self._leases[k]
	78	+ server_ids = []
82	79
83	80	for lease in session.leases:
84		- self._update_lease_from_server(lease)
	81	+ server_ids.append(lease.id)
85	82
86		- def get_pb2(self):
87		- leases = list(self._leases.values())
88		- if not leases:
89		- leases = None
	83	+ if lease.status.PENDING:
	84	+ self._register_lease(lease)
	85	+
	86	+ elif lease.status.CANCELLED:
	87	+ self._tenant_manager.cancel_tenancy(lease_id)
	88	+
	89	+ closed_lease_ids = [x for x in self._tenant_manager.get_lease_ids() if x not in server_ids]
	90	+
	91	+ for lease_id in closed_lease_ids:
	92	+ self._tenant_manager.remove_tenant(lease_id)
90	93
	94	+ def get_pb2(self):
91	95	return bots_pb2.BotSession(worker=self._worker.get_pb2(),
92	96	status=self._status,
93		- leases=leases,
	97	+ leases=self._tenant_manager.get_leases(),
94	98	bot_id=self.__bot_id,
95	99	name=self.__name)
96	100
97		- def lease_completed(self, lease):
98		- lease.state = LeaseState.COMPLETED.value
99		- self._leases[lease.id] = lease
100		-
101		- def _update_lease_from_server(self, lease):
102		- """
103		- State machine for any recieved updates to the leases.
104		- """
105		- # TODO: Compare with previous state of lease
106		- if lease.state == LeaseState.PENDING.value:
107		- lease.state = LeaseState.ACTIVE.value
108		- self._leases[lease.id] = lease
109		- self.update_bot_session()
110		- asyncio.ensure_future(self.create_work(lease))
111		-
112		- async def create_work(self, lease):
113		- self.logger.debug("Work created: [{}]".format(lease.id))
114		- loop = asyncio.get_event_loop()
115		-
	101	+ def _register_lease(self, lease):
	102	+ lease_id = lease.id
116	103	try:
117		- lease = await loop.run_in_executor(None, self._work, self._context, lease)
	104	+ self._tenant_manager.create_tenancy(lease)
118	105
119		- except grpc.RpcError as e:
120		- self.logger.error("RPC error thrown: [{}]".format(e))
121		- lease.status.CopyFrom(e.code())
	106	+ except KeyError as e:
	107	+ self.logger.error("Cannot register lease: [{}]. Error: {}".format(lease.id, e))
122	108
123		- except BotError as e:
124		- self.logger.error("Internal bot error thrown: [{}]".format(e))
125		- lease.status.code = code_pb2.INTERNAL
	109	+ else:
	110	+ try:
	111	+ self._hardware_interface.configure_hardware(lease)
126	112
127		- except Exception as e:
128		- self.logger.error("Exception thrown: [{}]".format(e))
129		- lease.status.code = code_pb2.INTERNAL
	113	+ except FailedPreconditionError as e:
	114	+ self._tenant_manager.complete_lease(lease_id, status=code_pb2.FailedPreconditionError)
130	115
131		- self.logger.debug("Work complete: [{}]".format(lease.id))
132		- self.lease_completed(lease)
	116	+ else:
	117	+ self._tenant_manager.create_work(lease_id)

buildgrid/bot/tenant.py

---

	1	+# Copyright (C) 2018 Bloomberg LP
	2	+#
	3	+# Licensed under the Apache License, Version 2.0 (the "License");
	4	+# you may not use this file except in compliance with the License.
	5	+# You may obtain a copy of the License at
	6	+#
	7	+# <http://www.apache.org/licenses/LICENSE-2.0>
	8	+#
	9	+# Unless required by applicable law or agreed to in writing, software
	10	+# distributed under the License is distributed on an "AS IS" BASIS,
	11	+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	12	+# See the License for the specific language governing permissions and
	13	+# limitations under the License.
	14	+
	15	+"""
	16	+Tenant
	17	+======
	18	+
	19	+Handles leased and runs leased work.
	20	+"""
	21	+
	22	+
	23	+from functools import partial
	24	+
	25	+from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
	26	+
	27	+from buildgrid._enums import LeaseState
	28	+
	29	+
	30	+class Tenant:
	31	+
	32	+ def __init__(self, lease, context):
	33	+
	34	+ if lease.state != LeaseState.PENDING:
	35	+ raise ValueError("Lease state not `PENDING`: {}".format(lease.state))
	36	+
	37	+ self.logger = logging.getLogger(__name__)
	38	+ self.lease_finished = False
	39	+
	40	+ self._context = context
	41	+ self._lease = lease
	42	+
	43	+ @property
	44	+ def lease(self):
	45	+ return self._lease
	46	+
	47	+ def get_state_state(self):
	48	+ return self._lease.state
	49	+
	50	+ def update_lease_state(self, state):
	51	+ self._lease.state = state
	52	+
	53	+ def update_lease_status(self, status):
	54	+ self._lease.status = status
	55	+
	56	+ async def run_work(self, work, executor=None, context = None):
	57	+ self.logger.debug("Work created: [{}]".format(self._lease.id))
	58	+
	59	+ # Ensures if anything happens to the lease during work, we still have a copy.
	60	+ lease = bots_pb2.Lease()
	61	+ lease.CopyFrom(self._lease)
	62	+
	63	+ loop = asyncio.get_event_loop()
	64	+
	65	+ try:
	66	+ lease = await loop.run_in_executor(executor, partial(work, context, self._lease))
	67	+ self._lease.CopyFrom(lease)
	68	+
	69	+ except asyncio.CancelledError as e:
	70	+ self.logger.error("Task cancelled: [{}]".format(e))
	71	+
	72	+ except grpc.RpcError as e:
	73	+ self.logger.error("RPC error thrown: [{}]".format(e))
	74	+ lease.status.CopyFrom(e.code())
	75	+
	76	+ except BotError as e:
	77	+ self.logger.error("Internal bot error thrown: [{}]".format(e))
	78	+ lease.status.code = code_pb2.INTERNAL
	79	+
	80	+ except Exception as e:
	81	+ self.logger.error("Exception thrown: [{}]".format(e))
	82	+ lease.status.code = code_pb2.INTERNAL
	83	+
	84	+ finally:
	85	+ self.logger.debug("Work completed: [{}]".format(lease.id))

buildgrid/bot/tenantmanager.py

---

	1	+# Copyright (C) 2018 Bloomberg LP
	2	+#
	3	+# Licensed under the Apache License, Version 2.0 (the "License");
	4	+# you may not use this file except in compliance with the License.
	5	+# You may obtain a copy of the License at
	6	+#
	7	+# <http://www.apache.org/licenses/LICENSE-2.0>
	8	+#
	9	+# Unless required by applicable law or agreed to in writing, software
	10	+# distributed under the License is distributed on an "AS IS" BASIS,
	11	+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	12	+# See the License for the specific language governing permissions and
	13	+# limitations under the License.
	14	+
	15	+
	16	+"""
	17	+TenantManager
	18	+=============
	19	+
	20	+Looks after leases of work.
	21	+"""
	22	+
	23	+
	24	+import asyncio
	25	+from functools import partial
	26	+
	27	+import grpc
	28	+
	29	+from buildgrid._enums import LeaseState
	30	+
	31	+from .tenant import Tenant
	32	+
	33	+class TenantManager:
	34	+
	35	+ def __init__(self):
	36	+ self.logger = logging.getLogger(__name__)
	37	+ self._tenants = {}
	38	+ self._tasks = {}
	39	+
	40	+ def create_tenancy(self, lease):
	41	+ lease_id = lease.id
	42	+
	43	+ if lease_id not in self._tenants:
	44	+ tenant = Tenant(lease, context)
	45	+ self._tenants[lease_id] = tenant
	46	+
	47	+ else:
	48	+ raise KeyError("Lease id already exists: [{}]".format(lease_id))
	49	+
	50	+ def remove_tenant(self, lease_id):
	51	+ state = self.get_lease_state(lease_id)
	52	+ if state != LeaseState.COMPLETED or state != LeaseState.CANCELLED:
	53	+ self.logger.error("Attempting to remove a lease not finished."
	54	+ "Bot will not remove lease."
	55	+ "Lease: [{}]".format(self._tenant_manager.get_lease(lease_id)))
	56	+
	57	+ else:
	58	+ del self._tenants[lease_id]
	59	+ del self._tasks[lease_id]
	60	+
	61	+ def get_leases(self):
	62	+ leases = []
	63	+ for _, tenant in self._tenants.items():
	64	+ leases.append(tenant.lease)
	65	+
	66	+ if not leases:
	67	+ return None
	68	+
	69	+ return leases
	70	+
	71	+ def get_lease_ids(self):
	72	+ lease_ids = []
	73	+ for lease_id in self._tenants.keys():
	74	+ lease_ids.append(lease_id)
	75	+
	76	+ if not lease_ids:
	77	+ return None
	78	+
	79	+ return leases
	80	+
	81	+ def get_lease_state(self, lease_id):
	82	+ return self._tenants[lease_id].get_lease_state()
	83	+
	84	+ def complete_lease(self, lease_id, status=None):
	85	+ if status is not None:
	86	+ self._update_lease_status(lease_id, status)
	87	+
	88	+ if self._tenants[lease_id].get_lease_state() != LeaseState.CANCELLED:
	89	+ self._update_lease_state(lease_id, LeaseState.COMPLETED)
	90	+
	91	+ def create_work(self, lease_id):
	92	+ self._update_lease_state(lease_id, LeaseState.ACTIVE)
	93	+ tenant = self._tenants[lease_id]
	94	+ task = asyncio.ensure_future(tenant.run_work())
	95	+
	96	+ task.add_done_callback(partial(self.complete_lease(lease_id)))
	97	+
	98	+ def cancel_tenancy(self, lease_id):
	99	+ self._update_lease_state(LeaseState.CANCELLED)
	100	+ self._tasks[lease_id].cancel()
	101	+
	102	+ def _update_lease_state(self, lease_id, state):
	103	+ self._tenants[lease_id].update_lease_state(state)
	104	+
	105	+ def _update_lease_status(self, lease_id, status):
	106	+ self._tenants[lease_id].update_lease_status(status)

buildgrid/server/bots/instance.py

---

...	...	@@ -97,7 +97,6 @@ class BotsInterface:
97	97	return bot_session
98	98
99	99	def _check_lease_state(self, lease):
100		-
101	100	# Check for cancelled lease
102	101	if self._scheduler.get_lease_cancelled(lease.id):
103	102	return None

—
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.