[Notes] [Git][BuildGrid/buildgrid][mablanch/75-requests-multiplexing] 13 commits: Implement ContentAddressableStorage.BatchReadBlobs()

[Martin Blanchard <gitlab mg gitlab com>]

Title: GitLab

Martin Blanchard pushed to branch mablanch/75-requests-multiplexing at BuildGrid / buildgrid

Commits:

• cc92a38f
  by Santiago Gil at 2019-01-18T12:52:57Z

  Implement ContentAddressableStorage.BatchReadBlobs()
  
  This will allow `bgd cas download-file` to fetch
  multiple files in one call.
  
  https://gitlab.com/BuildGrid/buildgrid/issues/76
  
  Signed-off-by: Santiago Gil <santiago gil codethink co uk>
  

• b990291b
  by Santiago Gil at 2019-01-18T12:52:57Z

  `download_file` command: accept list of files
  
  Now a batch of files can be downloaded with a single
  command.
  
  https://gitlab.com/BuildGrid/buildgrid/issues/76
  
  Signed-off-by: Santiago Gil <santiago gil codethink co uk>
  

• 5cd06add
  by Martin Blanchard at 2019-01-18T14:04:08Z

  job.py: Add an integer priority attribute
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 68298a8a
  by Martin Blanchard at 2019-01-18T14:04:08Z

  scheduler.py: Respect priority when scheduling jobs
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 521e4f9c
  by Martin Blanchard at 2019-01-18T14:04:08Z

  Keep track of peers watching jobs
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 0050edcd
  by Martin Blanchard at 2019-01-18T14:04:08Z

  Do not duplicate jobs for same action
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 4dc9684c
  by Martin Blanchard at 2019-01-18T14:04:08Z

  Stop exposing Job objects outside of Scheduler
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 2e2634af
  by Martin Blanchard at 2019-01-18T14:04:08Z

  docs: Preserve class members order from sources
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 9259eff3
  by Martin Blanchard at 2019-01-18T14:04:08Z

  job.py: Clean-up API and improve docsrings
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 662ff877
  by Martin Blanchard at 2019-01-18T14:04:08Z

  scheduler.py: Clean-up API and improve docsrings
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• af9cb986
  by Martin Blanchard at 2019-01-21T13:03:51Z

  Handle one operation per peer watching a job
  
  https://gitlab.com/BuildGrid/buildgrid/issues/75
  

• 85c796e7
  by Martin Blanchard at 2019-01-21T13:03:52Z

  execution/service.py: Send last response on cancellation
  

• 20f89c48
  by Martin Blanchard at 2019-01-21T13:03:52Z

  cmd_operation.py: Handle cancellation during wait
  

15 changed files:

• buildgrid/_app/commands/cmd_cas.py
• buildgrid/_app/commands/cmd_operation.py
• buildgrid/_exceptions.py
• buildgrid/server/bots/instance.py
• buildgrid/server/cas/instance.py
• buildgrid/server/cas/service.py
• buildgrid/server/execution/instance.py
• buildgrid/server/execution/service.py
• buildgrid/server/job.py
• buildgrid/server/operations/instance.py
• buildgrid/server/scheduler.py
• docs/source/conf.py
• tests/integration/bots_service.py
• tests/integration/execution_service.py
• tests/integration/operations_service.py

Changes:

buildgrid/_app/commands/cmd_cas.py

---

...	...	@@ -147,33 +147,41 @@ def _create_digest(digest_string):
147	147	return digest
148	148
149	149
150		-@cli.command('download-file', short_help="Download a file from the CAS server.")
151		-@click.argument('digest-string', nargs=1, type=click.STRING, required=True)
152		-@click.argument('file-path', nargs=1, type=click.Path(exists=False), required=True)
	150	+@cli.command('download-file', short_help="Download one or more files from the CAS server. "
	151	+ "(Specified as a space-separated list of DIGEST FILE_PATH)")
	152	+@click.argument('digest-path-list', nargs=-1, type=str, required=True) # 'digest path' pairs
153	153	@click.option('--verify', is_flag=True, show_default=True,
154	154	help="Check downloaded file's integrity.")
155	155	@pass_context
156		-def download_file(context, digest_string, file_path, verify):
157		- if os.path.exists(file_path):
158		- click.echo("Error: Invalid value for " +
159		- "path=[{}] already exists.".format(file_path), err=True)
160		- return
161		-
162		- digest = _create_digest(digest_string)
	156	+def download_file(context, digest_path_list, verify):
	157	+ # Downloading files:
	158	+ downloaded_files = {}
163	159	with download(context.channel, instance=context.instance_name) as downloader:
164		- downloader.download_file(digest, file_path)
165		-
166		- if verify:
167		- file_digest = create_digest(read_file(file_path))
168		- if file_digest != digest:
169		- click.echo("Error: Failed to verify path=[{}]".format(file_path), err=True)
170		- return
171		-
172		- if os.path.isfile(file_path):
173		- click.echo("Success: Pulled path=[{}] from digest=[{}/{}]"
174		- .format(file_path, digest.hash, digest.size_bytes))
175		- else:
176		- click.echo('Error: Failed pulling "{}"'.format(file_path), err=True)
	160	+ for (digest_string, file_path) in zip(digest_path_list[0::2],
	161	+ digest_path_list[1::2]):
	162	+ if os.path.exists(file_path):
	163	+ click.echo("Error: Invalid value for " +
	164	+ "path=[{}] already exists.".format(file_path), err=True)
	165	+ continue
	166	+
	167	+ digest = _create_digest(digest_string)
	168	+
	169	+ downloader.download_file(digest, file_path)
	170	+ downloaded_files[file_path] = digest
	171	+
	172	+ # Verifying:
	173	+ for (file_path, digest) in downloaded_files.items():
	174	+ if verify:
	175	+ file_digest = create_digest(read_file(file_path))
	176	+ if file_digest != digest:
	177	+ click.echo("Error: Failed to verify path=[{}]".format(file_path), err=True)
	178	+ continue
	179	+
	180	+ if os.path.isfile(file_path):
	181	+ click.echo("Success: Pulled path=[{}] from digest=[{}/{}]"
	182	+ .format(file_path, digest.hash, digest.size_bytes))
	183	+ else:
	184	+ click.echo('Error: Failed pulling "{}"'.format(file_path), err=True)
177	185
178	186
179	187	@cli.command('download-dir', short_help="Download a directory from the CAS server.")

buildgrid/_app/commands/cmd_operation.py

---

...	...	@@ -27,6 +27,7 @@ from textwrap import indent
27	27
28	28	import click
29	29	from google.protobuf import json_format
	30	+import grpc
30	31
31	32	from buildgrid.client.authentication import setup_channel
32	33	from buildgrid._enums import OperationStage
...	...	@@ -213,10 +214,17 @@ def wait(context, operation_name, json):
213	214
214	215	operation_iterator = stub.WaitExecution(request)
215	216
216		- for operation in operation_iterator:
217		- if not json and operation.done:
218		- _print_operation_status(operation, print_details=True)
219		- elif not json:
220		- _print_operation_status(operation)
221		- else:
222		- click.echo(json_format.MessageToJson(operation))
	217	+ try:
	218	+ for operation in operation_iterator:
	219	+ if not json and operation.done:
	220	+ _print_operation_status(operation, print_details=True)
	221	+ elif not json:
	222	+ _print_operation_status(operation)
	223	+ else:
	224	+ click.echo(json_format.MessageToJson(operation))
	225	+
	226	+ except grpc.RpcError as e:
	227	+ if e.code() != grpc.StatusCode.CANCELLED:
	228	+ click.echo('Error: {}'
	229	+ .format(e.details), err=True)
	230	+ sys.exit(-1)

buildgrid/_exceptions.py

---

...	...	@@ -56,6 +56,7 @@ class CancelledError(BgdError):
56	56	"""The job was cancelled and any callers should be notified"""
57	57	def __init__(self, message, detail=None, reason=None):
58	58	super().__init__(message, detail=detail, domain=ErrorDomain.SERVER, reason=reason)
	59	+ self.last_response = None
59	60
60	61
61	62	class InvalidArgumentError(BgdError):

buildgrid/server/bots/instance.py

---

...	...	@@ -126,7 +126,7 @@ class BotsInterface:
126	126	# Job does not exist, remove from bot.
127	127	return None
128	128
129		- self._scheduler.update_job_lease(lease)
	129	+ self._scheduler.update_job_lease_state(lease.id, lease)
130	130
131	131	if lease_state == LeaseState.COMPLETED:
132	132	return None
...	...	@@ -164,7 +164,7 @@ class BotsInterface:
164	164	self.__logger.error("Assigned lease id=[%s],"
165	165	" not found on bot with name=[%s] and id=[%s]."
166	166	" Retrying job", lease_id, bot_session.name, bot_session.bot_id)
167		- self._scheduler.retry_job(lease_id)
	167	+ self._scheduler.retry_job_lease(lease_id)
168	168
169	169	def _close_bot_session(self, name):
170	170	""" Before removing the session, close any leases and
...	...	@@ -177,7 +177,7 @@ class BotsInterface:
177	177
178	178	self.__logger.debug("Attempting to close [%s] with name: [%s]", bot_id, name)
179	179	for lease_id in self._assigned_leases[name]:
180		- self._scheduler.retry_job(lease_id)
	180	+ self._scheduler.retry_job_lease(lease_id)
181	181	self._assigned_leases.pop(name)
182	182
183	183	self.__logger.debug("Closing bot session: [%s]", name)

buildgrid/server/cas/instance.py

---

...	...	@@ -24,6 +24,7 @@ import logging
24	24	from buildgrid._exceptions import InvalidArgumentError, NotFoundError, OutOfRangeError
25	25	from buildgrid._protos.google.bytestream import bytestream_pb2
26	26	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2 as re_pb2
	27	+from buildgrid._protos.google.rpc import code_pb2, status_pb2
27	28	from buildgrid.settings import HASH, HASH_LENGTH, MAX_REQUEST_SIZE, MAX_REQUEST_COUNT
28	29	from buildgrid.utils import get_hash_type
29	30
...	...	@@ -70,6 +71,35 @@ class ContentAddressableStorageInstance:
70	71
71	72	return response
72	73
	74	+ def batch_read_blobs(self, digests):
	75	+ storage = self._storage
	76	+
	77	+ response = re_pb2.BatchReadBlobsResponse()
	78	+
	79	+ requested_bytes = sum((digest.size_bytes for digest in digests))
	80	+ max_batch_size = self.max_batch_total_size_bytes()
	81	+
	82	+ if requested_bytes > max_batch_size:
	83	+ raise InvalidArgumentError('Combined total size of blobs exceeds '
	84	+ 'server limit. '
	85	+ '({} > {} [byte])'.format(requested_bytes,
	86	+ max_batch_size))
	87	+
	88	+ for digest in digests:
	89	+ response_proto = response.responses.add()
	90	+ response_proto.digest.CopyFrom(digest)
	91	+
	92	+ blob = storage.get_blob(digest)
	93	+ if blob:
	94	+ response_proto.data = blob.read()
	95	+ status_code = code_pb2.OK
	96	+ else:
	97	+ status_code = code_pb2.NOT_FOUND
	98	+
	99	+ response_proto.status.CopyFrom(status_pb2.Status(code=status_code))
	100	+
	101	+ return response
	102	+
73	103	def get_tree(self, request):
74	104	storage = self._storage
75	105

buildgrid/server/cas/service.py

---

...	...	@@ -86,8 +86,15 @@ class ContentAddressableStorageService(remote_execution_pb2_grpc.ContentAddressa
86	86	def BatchReadBlobs(self, request, context):
87	87	self.__logger.debug("BatchReadBlobs request from [%s]", context.peer())
88	88
89		- context.set_code(grpc.StatusCode.UNIMPLEMENTED)
90		- context.set_details('Method not implemented!')
	89	+ try:
	90	+ instance = self._get_instance(request.instance_name)
	91	+ response = instance.batch_read_blobs(request.digests)
	92	+ return response
	93	+
	94	+ except InvalidArgumentError as e:
	95	+ self.__logger.error(e)
	96	+ context.set_details(str(e))
	97	+ context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
91	98
92	99	return remote_execution_pb2.BatchReadBlobsResponse()
93	100

buildgrid/server/execution/instance.py

---

...	...	@@ -21,11 +21,9 @@ An instance of the Remote Execution Service.
21	21
22	22	import logging
23	23
24		-from buildgrid._exceptions import FailedPreconditionError, InvalidArgumentError
	24	+from buildgrid._exceptions import FailedPreconditionError, InvalidArgumentError, NotFoundError
25	25	from buildgrid._protos.build.bazel.remote.execution.v2.remote_execution_pb2 import Action
26		-
27		-from ..job import Job
28		-from ...utils import get_hash_type
	26	+from buildgrid.utils import get_hash_type
29	27
30	28
31	29	class ExecutionInstance:
...	...	@@ -46,44 +44,56 @@ class ExecutionInstance:
46	44	def hash_type(self):
47	45	return get_hash_type()
48	46
49		- def execute(self, action_digest, skip_cache_lookup, message_queue=None):
	47	+ def execute(self, action_digest, skip_cache_lookup):
50	48	""" Sends a job for execution.
51	49	Queues an action and creates an Operation instance to be associated with
52	50	this action.
53	51	"""
54		-
55	52	action = self._storage.get_message(action_digest, Action)
56	53
57	54	if not action:
58	55	raise FailedPreconditionError("Could not get action from storage.")
59	56
60		- job = Job(action, action_digest)
61		- if message_queue is not None:
62		- job.register_client(message_queue)
	57	+ return self._scheduler.queue_job_operation(action, action_digest,
	58	+ skip_cache_lookup=skip_cache_lookup)
63	59
64		- self._scheduler.queue_job(job, skip_cache_lookup)
	60	+ def register_job_peer(self, job_name, peer, message_queue):
	61	+ try:
	62	+ return self._scheduler.register_job_peer(job_name,
	63	+ peer, message_queue)
65	64
66		- return job.operation
	65	+ except NotFoundError:
	66	+ raise InvalidArgumentError("Job name does not exist: [{}]"
	67	+ .format(job_name))
67	68
68		- def register_message_client(self, name, queue):
	69	+ def register_operation_peer(self, operation_name, peer, message_queue):
69	70	try:
70		- self._scheduler.register_client(name, queue)
	71	+ self._scheduler.register_job_operation_peer(operation_name,
	72	+ peer, message_queue)
71	73
72		- except KeyError:
73		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
	74	+ except NotFoundError:
	75	+ raise InvalidArgumentError("Operation name does not exist: [{}]"
	76	+ .format(operation_name))
74	77
75		- def unregister_message_client(self, name, queue):
	78	+ def unregister_operation_peer(self, operation_name, peer):
76	79	try:
77		- self._scheduler.unregister_client(name, queue)
	80	+ self._scheduler.unregister_job_operation_peer(operation_name, peer)
	81	+
	82	+ except NotFoundError:
	83	+ raise InvalidArgumentError("Operation name does not exist: [{}]"
	84	+ .format(operation_name))
	85	+
	86	+ def stream_operation_updates(self, message_queue):
	87	+ error, operation = message_queue.get()
	88	+ if error is not None:
	89	+ raise error
78	90
79		- except KeyError:
80		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
	91	+ while not operation.done:
	92	+ yield operation
81	93
82		- def stream_operation_updates(self, message_queue, operation_name):
83		- job = message_queue.get()
84		- while not job.operation.done:
85		- yield job.operation
86		- job = message_queue.get()
87		- job.check_operation_status()
	94	+ error, operation = message_queue.get()
	95	+ if error is not None:
	96	+ error.last_response = operation
	97	+ raise error
88	98
89		- yield job.operation
	99	+ yield operation

buildgrid/server/execution/service.py

---

...	...	@@ -98,12 +98,15 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
98	98
99	99	try:
100	100	instance = self._get_instance(instance_name)
101		- operation = instance.execute(request.action_digest,
102		- request.skip_cache_lookup,
103		- message_queue)
	101	+
	102	+ job_name = instance.execute(request.action_digest,
	103	+ request.skip_cache_lookup)
	104	+
	105	+ operation_name = instance.register_job_peer(job_name,
	106	+ peer, message_queue)
104	107
105	108	context.add_callback(partial(self._rpc_termination_callback,
106		- peer, instance_name, operation.name, message_queue))
	109	+ peer, instance_name, operation_name))
107	110
108	111	if self._is_instrumented:
109	112	if peer not in self.__peers:
...	...	@@ -112,16 +115,13 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
112	115	else:
113	116	self.__peers[peer] += 1
114	117
115		- instanced_op_name = "{}/{}".format(instance_name, operation.name)
	118	+ operation_full_name = "{}/{}".format(instance_name, operation_name)
116	119
117		- self.__logger.info("Operation name: [%s]", instanced_op_name)
	120	+ self.__logger.info("Operation name: [%s]", operation_full_name)
118	121
119		- for operation in instance.stream_operation_updates(message_queue,
120		- operation.name):
121		- op = operations_pb2.Operation()
122		- op.CopyFrom(operation)
123		- op.name = instanced_op_name
124		- yield op
	122	+ for operation in instance.stream_operation_updates(message_queue):
	123	+ operation.name = operation_full_name
	124	+ yield operation
125	125
126	126	except InvalidArgumentError as e:
127	127	self.__logger.error(e)
...	...	@@ -139,7 +139,7 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
139	139	self.__logger.error(e)
140	140	context.set_details(str(e))
141	141	context.set_code(grpc.StatusCode.CANCELLED)
142		- yield operations_pb2.Operation()
	142	+ yield e.last_response
143	143
144	144	@authorize(AuthContext)
145	145	def WaitExecution(self, request, context):
...	...	@@ -160,9 +160,11 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
160	160	try:
161	161	instance = self._get_instance(instance_name)
162	162
163		- instance.register_message_client(operation_name, message_queue)
	163	+ instance.register_operation_peer(operation_name,
	164	+ peer, message_queue)
	165	+
164	166	context.add_callback(partial(self._rpc_termination_callback,
165		- peer, instance_name, operation_name, message_queue))
	167	+ peer, instance_name, operation_name))
166	168
167	169	if self._is_instrumented:
168	170	if peer not in self.__peers:
...	...	@@ -171,12 +173,11 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
171	173	else:
172	174	self.__peers[peer] += 1
173	175
174		- for operation in instance.stream_operation_updates(message_queue,
175		- operation_name):
176		- op = operations_pb2.Operation()
177		- op.CopyFrom(operation)
178		- op.name = request.name
179		- yield op
	176	+ operation_full_name = "{}/{}".format(instance_name, operation_name)
	177	+
	178	+ for operation in instance.stream_operation_updates(message_queue):
	179	+ operation.name = operation_full_name
	180	+ yield operation
180	181
181	182	except InvalidArgumentError as e:
182	183	self.__logger.error(e)
...	...	@@ -188,7 +189,7 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
188	189	self.__logger.error(e)
189	190	context.set_details(str(e))
190	191	context.set_code(grpc.StatusCode.CANCELLED)
191		- yield operations_pb2.Operation()
	192	+ yield e.last_response
192	193
193	194	# --- Public API: Monitoring ---
194	195
...	...	@@ -211,10 +212,10 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
211	212
212	213	# --- Private API ---
213	214
214		- def _rpc_termination_callback(self, peer, instance_name, job_name, message_queue):
	215	+ def _rpc_termination_callback(self, peer, instance_name, operation_name):
215	216	instance = self._get_instance(instance_name)
216	217
217		- instance.unregister_message_client(job_name, message_queue)
	218	+ instance.unregister_operation_peer(operation_name, peer)
218	219
219	220	if self._is_instrumented:
220	221	if self.__peers[peer] > 1:

buildgrid/server/job.py

---

...	...	@@ -20,7 +20,7 @@ import uuid
20	20	from google.protobuf import duration_pb2, timestamp_pb2
21	21
22	22	from buildgrid._enums import LeaseState, OperationStage
23		-from buildgrid._exceptions import CancelledError
	23	+from buildgrid._exceptions import CancelledError, NotFoundError
24	24	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
25	25	from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
26	26	from buildgrid._protos.google.longrunning import operations_pb2
...	...	@@ -29,35 +29,70 @@ from buildgrid._protos.google.rpc import code_pb2
29	29
30	30	class Job:
31	31
32		- def __init__(self, action, action_digest):
	32	+ def __init__(self, action, action_digest, priority=0):
33	33	self.__logger = logging.getLogger(__name__)
34	34
35	35	self._name = str(uuid.uuid4())
	36	+ self._priority = priority
36	37	self._action = remote_execution_pb2.Action()
37		- self._operation = operations_pb2.Operation()
38	38	self._lease = None
39	39
40	40	self.__execute_response = None
41	41	self.__operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
	42	+ self.__operations_by_name = {} # Name to Operation 1:1 mapping
	43	+ self.__operations_by_peer = {} # Peer to Operation 1:1 mapping
42	44
43	45	self.__queued_timestamp = timestamp_pb2.Timestamp()
44	46	self.__queued_time_duration = duration_pb2.Duration()
45	47	self.__worker_start_timestamp = timestamp_pb2.Timestamp()
46	48	self.__worker_completed_timestamp = timestamp_pb2.Timestamp()
47	49
48		- self.__operation_cancelled = False
	50	+ self.__operations_message_queues = {}
	51	+ self.__operations_cancelled = set()
49	52	self.__lease_cancelled = False
	53	+ self.__job_cancelled = False
50	54
51	55	self.__operation_metadata.action_digest.CopyFrom(action_digest)
52	56	self.__operation_metadata.stage = OperationStage.UNKNOWN.value
53	57
54	58	self._action.CopyFrom(action)
55	59	self._do_not_cache = self._action.do_not_cache
56		- self._operation_update_queues = []
57		- self._operation.name = self._name
58		- self._operation.done = False
59	60	self._n_tries = 0
60	61
	62	+ self._done = False
	63	+
	64	+ def __lt__(self, other):
	65	+ try:
	66	+ return self.priority < other.priority
	67	+ except AttributeError:
	68	+ return NotImplemented
	69	+
	70	+ def __le__(self, other):
	71	+ try:
	72	+ return self.priority <= other.priority
	73	+ except AttributeError:
	74	+ return NotImplemented
	75	+
	76	+ def __eq__(self, other):
	77	+ if isinstance(other, Job):
	78	+ return self.name == other.name
	79	+ return False
	80	+
	81	+ def __ne__(self, other):
	82	+ return not self.__eq__(other)
	83	+
	84	+ def __gt__(self, other):
	85	+ try:
	86	+ return self.priority > other.priority
	87	+ except AttributeError:
	88	+ return NotImplemented
	89	+
	90	+ def __ge__(self, other):
	91	+ try:
	92	+ return self.priority >= other.priority
	93	+ except AttributeError:
	94	+ return NotImplemented
	95	+
61	96	# --- Public API ---
62	97
63	98	@property
...	...	@@ -65,17 +100,27 @@ class Job:
65	100	return self._name
66	101
67	102	@property
68		- def do_not_cache(self):
69		- return self._do_not_cache
	103	+ def priority(self):
	104	+ return self._priority
70	105
71	106	@property
72		- def action(self):
73		- return self._action
	107	+ def done(self):
	108	+ return self._done
	109	+
	110	+ # --- Public API: REAPI ---
	111	+
	112	+ @property
	113	+ def do_not_cache(self):
	114	+ return self._do_not_cache
74	115
75	116	@property
76	117	def action_digest(self):
77	118	return self.__operation_metadata.action_digest
78	119
	120	+ @property
	121	+ def operation_stage(self):
	122	+ return OperationStage(self.__operation_metadata.stage)
	123	+
79	124	@property
80	125	def action_result(self):
81	126	if self.__execute_response is not None:
...	...	@@ -84,19 +129,210 @@ class Job:
84	129	return None
85	130
86	131	@property
87		- def holds_cached_action_result(self):
	132	+ def holds_cached_result(self):
88	133	if self.__execute_response is not None:
89	134	return self.__execute_response.cached_result
90	135	else:
91	136	return False
92	137
93		- @property
94		- def operation(self):
95		- return self._operation
	138	+ def set_cached_result(self, action_result):
	139	+ """Allows specifying an action result form the action cache for the job.
	140	+
	141	+ Note:
	142	+ This won't trigger any :class:`Operation` stage transition.
	143	+
	144	+ Args:
	145	+ action_result (ActionResult): The result from cache.
	146	+ """
	147	+ self.__execute_response = remote_execution_pb2.ExecuteResponse()
	148	+ self.__execute_response.result.CopyFrom(action_result)
	149	+ self.__execute_response.cached_result = True
96	150
97	151	@property
98		- def operation_stage(self):
99		- return OperationStage(self.__operation_metadata.state)
	152	+ def n_peers(self):
	153	+ return len(self.__operations_message_queues)
	154	+
	155	+ def n_peers_for_operation(self, operation_name):
	156	+ return len([operation for operation in self.__operations_by_peer.values()
	157	+ if operation.name == operation_name])
	158	+
	159	+ def register_new_operation_peer(self, peer, message_queue):
	160	+ """Subscribes to a new job's :class:`Operation` stage changes.
	161	+
	162	+ Args:
	163	+ peer (str): a unique string identifying the client.
	164	+ message_queue (queue.Queue): the event queue to register.
	165	+
	166	+ Returns:
	167	+ str: The name of the subscribed :class:`Operation`.
	168	+ """
	169	+ new_operation = operations_pb2.Operation()
	170	+ # Copy state from first existing and non cancelled operation:
	171	+ for operation in self.__operations_by_name.values():
	172	+ if operation.name not in self.__operations_cancelled:
	173	+ new_operation.CopyFrom(operation)
	174	+ break
	175	+
	176	+ new_operation.name = str(uuid.uuid4())
	177	+
	178	+ self.__operations_by_name[new_operation.name] = new_operation
	179	+ self.__operations_by_peer[peer] = new_operation
	180	+ self.__operations_message_queues[peer] = message_queue
	181	+
	182	+ self._send_operations_updates(peers=[peer])
	183	+
	184	+ return new_operation.name
	185	+
	186	+ def register_operation_peer(self, operation_name, peer, message_queue):
	187	+ """Subscribes to one of the job's :class:`Operation` stage changes.
	188	+
	189	+ Args:
	190	+ operation_name (str, optinal): an existing operation's name to
	191	+ subscribe to.
	192	+ peer (str): a unique string identifying the client.
	193	+ message_queue (queue.Queue): the event queue to register.
	194	+
	195	+ Returns:
	196	+ str: The name of the subscribed :class:`Operation`.
	197	+
	198	+ Raises:
	199	+ NotFoundError: If no operation with `operation_name` exists.
	200	+ """
	201	+ try:
	202	+ operation = self.__operations_by_name[operation_name]
	203	+
	204	+ except KeyError:
	205	+ raise NotFoundError("Operation name does not exist: [{}]"
	206	+ .format(operation_name))
	207	+
	208	+ self.__operations_by_peer[peer] = operation
	209	+ self.__operations_message_queues[peer] = message_queue
	210	+
	211	+ self._send_operations_updates(peers=[peer])
	212	+
	213	+ def unregister_operation_peer(self, operation_name, peer):
	214	+ """Unsubscribes to the job's :class:`Operation` stage change.
	215	+
	216	+ Args:
	217	+ peer (str): a unique string identifying the client.
	218	+
	219	+ Raises:
	220	+ NotFoundError: If no operation with `operation_name` exists.
	221	+ """
	222	+ try:
	223	+ operation = self.__operations_by_name[operation_name]
	224	+
	225	+ except KeyError:
	226	+ raise NotFoundError("Operation name does not exist: [{}]"
	227	+ .format(operation_name))
	228	+
	229	+ if peer in self.__operations_message_queues:
	230	+ del self.__operations_message_queues[peer]
	231	+
	232	+ del self.__operations_by_peer[peer]
	233	+
	234	+ # Drop the operation if nobody is watching it anymore:
	235	+ if operation not in self.__operations_by_peer.values():
	236	+ del self.__operations_by_name[operation.name]
	237	+
	238	+ self.__operations_cancelled.discard(operation.name)
	239	+
	240	+ def list_operations(self):
	241	+ """Lists the :class:`Operation` related to a job.
	242	+
	243	+ Returns:
	244	+ list: A list of :class:`Operation` names.
	245	+ """
	246	+ return list(self.__operations_by_name.keys())
	247	+
	248	+ def get_operation(self, operation_name):
	249	+ """Returns a copy of the the job's :class:`Operation`.
	250	+
	251	+ Args:
	252	+ operation_name (str): the operation's name.
	253	+
	254	+ Raises:
	255	+ NotFoundError: If no operation with `operation_name` exists.
	256	+ """
	257	+ try:
	258	+ operation = self.__operations_by_name[operation_name]
	259	+
	260	+ except KeyError:
	261	+ raise NotFoundError("Operation name does not exist: [{}]"
	262	+ .format(operation_name))
	263	+
	264	+ return self._copy_operation(operation)
	265	+
	266	+ def update_operation_stage(self, stage):
	267	+ """Operates a stage transition for the job's :class:`Operation`.
	268	+
	269	+ Args:
	270	+ stage (OperationStage): the operation stage to transition to.
	271	+ """
	272	+ if stage.value == self.__operation_metadata.stage:
	273	+ return
	274	+
	275	+ self.__operation_metadata.stage = stage.value
	276	+
	277	+ if self.__operation_metadata.stage == OperationStage.QUEUED.value:
	278	+ if self.__queued_timestamp.ByteSize() == 0:
	279	+ self.__queued_timestamp.GetCurrentTime()
	280	+ self._n_tries += 1
	281	+
	282	+ elif self.__operation_metadata.stage == OperationStage.EXECUTING.value:
	283	+ queue_in, queue_out = self.__queued_timestamp.ToDatetime(), datetime.now()
	284	+ self.__queued_time_duration.FromTimedelta(queue_out - queue_in)
	285	+
	286	+ elif self.__operation_metadata.stage == OperationStage.COMPLETED.value:
	287	+ self._done = True
	288	+
	289	+ self._send_operations_updates()
	290	+
	291	+ def cancel_operation(self, operation_name):
	292	+ """Triggers a job's :class:`Operation` cancellation.
	293	+
	294	+ This may cancel any job's :class:`Lease` that may have been issued.
	295	+
	296	+ Args:
	297	+ operation_name (str): the operation's name.
	298	+
	299	+ Raises:
	300	+ NotFoundError: If no operation with `operation_name` exists.
	301	+ """
	302	+ try:
	303	+ operation = self.__operations_by_name[operation_name]
	304	+
	305	+ except KeyError:
	306	+ raise NotFoundError("Operation name does not exist: [{}]"
	307	+ .format(operation_name))
	308	+
	309	+ self.__operations_cancelled.add(operation.name)
	310	+
	311	+ ongoing_operations = set(self.__operations_by_name.keys())
	312	+ # Job is cancelled if all the operation are:
	313	+ self.__job_cancelled = ongoing_operations.issubset(self.__operations_cancelled)
	314	+
	315	+ if self.__job_cancelled and self._lease is not None:
	316	+ self.cancel_lease()
	317	+
	318	+ peers_to_notify = set()
	319	+ # If the job is not cancelled, notify all the peers watching the given
	320	+ # operation; if the job is cancelled, only notify the peers for which
	321	+ # the operation status changed.
	322	+ for peer, operation in self.__operations_by_peer.items():
	323	+ if self.__job_cancelled:
	324	+ if operation.name not in self.__operations_cancelled:
	325	+ peers_to_notify.add(peer)
	326	+ elif operation.name == operation_name:
	327	+ peers_to_notify.add(peer)
	328	+
	329	+ else:
	330	+ if operation.name == operation_name:
	331	+ peers_to_notify.add(peer)
	332	+
	333	+ self._send_operations_updates(peers=peers_to_notify, notify_cancelled=True)
	334	+
	335	+ # --- Public API: RWAPI ---
100	336
101	337	@property
102	338	def lease(self):
...	...	@@ -117,45 +353,15 @@ class Job:
117	353	def n_tries(self):
118	354	return self._n_tries
119	355
120		- @property
121		- def n_clients(self):
122		- return len(self._operation_update_queues)
123		-
124		- def register_client(self, queue):
125		- """Subscribes to the job's :class:`Operation` stage change events.
126		-
127		- Queues this :object:`Job` instance.
128		-
129		- Args:
130		- queue (queue.Queue): the event queue to register.
131		- """
132		- self._operation_update_queues.append(queue)
133		- queue.put(self)
134		-
135		- def unregister_client(self, queue):
136		- """Unsubscribes to the job's :class:`Operation` stage change events.
137		-
138		- Args:
139		- queue (queue.Queue): the event queue to unregister.
140		- """
141		- self._operation_update_queues.remove(queue)
142		-
143		- def set_cached_result(self, action_result):
144		- """Allows specifying an action result form the action cache for the job.
145		- """
146		- self.__execute_response = remote_execution_pb2.ExecuteResponse()
147		- self.__execute_response.result.CopyFrom(action_result)
148		- self.__execute_response.cached_result = True
149		-
150	356	def create_lease(self):
151	357	"""Emits a new :class:`Lease` for the job.
152	358
153	359	Only one :class:`Lease` can be emitted for a given job. This method
154		- should only be used once, any furhter calls are ignored.
	360	+ should only be used once, any further calls are ignored.
155	361	"""
156		- if self.__operation_cancelled:
157		- return None
158		- elif self._lease is not None:
	362	+ if self._lease is not None:
	363	+ return self._lease
	364	+ elif self.__job_cancelled:
159	365	return None
160	366
161	367	self._lease = bots_pb2.Lease()
...	...	@@ -166,14 +372,14 @@ class Job:
166	372	return self._lease
167	373
168	374	def update_lease_state(self, state, status=None, result=None):
169		- """Operates a state transition for the job's current :class:Lease.
	375	+ """Operates a state transition for the job's current :class:`Lease`.
170	376
171	377	Args:
172	378	state (LeaseState): the lease state to transition to.
173		- status (google.rpc.Status): the lease execution status, only
174		- required if `state` is `COMPLETED`.
175		- result (google.protobuf.Any): the lease execution result, only
176		- required if `state` is `COMPLETED`.
	379	+ status (google.rpc.Status, optional): the lease execution status,
	380	+ only required if `state` is `COMPLETED`.
	381	+ result (google.protobuf.Any, optional): the lease execution result,
	382	+ only required if `state` is `COMPLETED`.
177	383	"""
178	384	if state.value == self._lease.state:
179	385	return
...	...	@@ -214,79 +420,96 @@ class Job:
214	420	self.__execute_response.status.CopyFrom(status)
215	421
216	422	def cancel_lease(self):
217		- """Triggers a job's :class:Lease cancellation.
	423	+ """Triggers a job's :class:`Lease` cancellation.
218	424
219		- This will not cancel the job's :class:Operation.
	425	+ Note:
	426	+ This will not cancel the job's :class:`Operation`.
220	427	"""
221	428	self.__lease_cancelled = True
222	429	if self._lease is not None:
223	430	self.update_lease_state(LeaseState.CANCELLED)
224	431
225	432	def delete_lease(self):
226		- """Discard the job's :class:Lease."""
	433	+ """Discard the job's :class:`Lease`.
	434	+
	435	+ Note:
	436	+ This will not cancel the job's :class:`Operation`.
	437	+ """
227	438	self.__worker_start_timestamp.Clear()
228	439	self.__worker_completed_timestamp.Clear()
229	440
230	441	self._lease = None
231	442
232		- def update_operation_stage(self, stage):
233		- """Operates a stage transition for the job's :class:Operation.
	443	+ # --- Public API: Monitoring ---
234	444
235		- Args:
236		- stage (OperationStage): the operation stage to transition to.
237		- """
238		- if stage.value == self.__operation_metadata.stage:
239		- return
	445	+ def query_queue_time(self):
	446	+ return self.__queued_time_duration.ToTimedelta()
240	447
241		- self.__operation_metadata.stage = stage.value
	448	+ def query_n_retries(self):
	449	+ return self._n_tries - 1 if self._n_tries > 0 else 0
242	450
243		- if self.__operation_metadata.stage == OperationStage.QUEUED.value:
244		- if self.__queued_timestamp.ByteSize() == 0:
245		- self.__queued_timestamp.GetCurrentTime()
246		- self._n_tries += 1
	451	+ # --- Private API ---
247	452
248		- elif self.__operation_metadata.stage == OperationStage.EXECUTING.value:
249		- queue_in, queue_out = self.__queued_timestamp.ToDatetime(), datetime.now()
250		- self.__queued_time_duration.FromTimedelta(queue_out - queue_in)
	453	+ def _copy_operation(self, operation):
	454	+ """Simply duplicates a given :class:`Lease` object."""
	455	+ new_operation = operations_pb2.Operation()
	456	+ new_operation.CopyFrom(operation)
251	457
252		- elif self.__operation_metadata.stage == OperationStage.COMPLETED.value:
253		- if self.__execute_response is not None:
254		- self._operation.response.Pack(self.__execute_response)
255		- self._operation.done = True
	458	+ return new_operation
256	459
257		- self._operation.metadata.Pack(self.__operation_metadata)
	460	+ def _update_operation(self, operation, operation_metadata, execute_response=None, done=False):
	461	+ """Forges a :class:`Operation` message given input data."""
	462	+ operation.metadata.Pack(operation_metadata)
258	463
259		- for queue in self._operation_update_queues:
260		- queue.put(self)
	464	+ if execute_response is not None:
	465	+ operation.response.Pack(execute_response)
261	466
262		- def check_operation_status(self):
263		- """Reports errors on unexpected job's :class:Operation state.
	467	+ operation.done = done
264	468
265		- Raises:
266		- CancelledError: if the job's :class:Operation was cancelled.
267		- """
268		- if self.__operation_cancelled:
269		- raise CancelledError(self.__execute_response.status.message)
	469	+ def _update_cancelled_operation(self, operation, operation_metadata, execute_response=None):
	470	+ """Forges a cancelled :class:`Operation` message given input data."""
	471	+ cancelled_operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
	472	+ cancelled_operation_metadata.CopyFrom(operation_metadata)
	473	+ cancelled_operation_metadata.stage = OperationStage.COMPLETED.value
270	474
271		- def cancel_operation(self):
272		- """Triggers a job's :class:Operation cancellation.
	475	+ operation.metadata.Pack(cancelled_operation_metadata)
273	476
274		- This will also cancel any job's :class:Lease that may have been issued.
275		- """
276		- self.__operation_cancelled = True
277		- if self._lease is not None:
278		- self.cancel_lease()
	477	+ cancelled_execute_response = remote_execution_pb2.ExecuteResponse()
	478	+ if execute_response is not None:
	479	+ cancelled_execute_response.CopyFrom(self.__execute_response)
	480	+ cancelled_execute_response.status.code = code_pb2.CANCELLED
	481	+ cancelled_execute_response.status.message = "Operation cancelled by client."
279	482
280		- self.__execute_response = remote_execution_pb2.ExecuteResponse()
281		- self.__execute_response.status.code = code_pb2.CANCELLED
282		- self.__execute_response.status.message = "Operation cancelled by client."
	483	+ operation.response.Pack(cancelled_execute_response)
283	484
284		- self.update_operation_stage(OperationStage.COMPLETED)
	485	+ operation.done = True
285	486
286		- # --- Public API: Monitoring ---
	487	+ def _send_operations_updates(self, peers=None, notify_cancelled=False):
	488	+ """Sends :class:`Operation` stage change messages to watchers."""
	489	+ for operation in self.__operations_by_name.values():
	490	+ if operation.name in self.__operations_cancelled:
	491	+ self._update_cancelled_operation(operation, self.__operation_metadata,
	492	+ execute_response=self.__execute_response)
287	493
288		- def query_queue_time(self):
289		- return self.__queued_time_duration.ToTimedelta()
	494	+ else:
	495	+ self._update_operation(operation, self.__operation_metadata,
	496	+ execute_response=self.__execute_response,
	497	+ done=self._done)
290	498
291		- def query_n_retries(self):
292		- return self._n_tries - 1 if self._n_tries > 0 else 0
	499	+ for peer, message_queue in self.__operations_message_queues.items():
	500	+ if peer not in self.__operations_by_peer:
	501	+ continue
	502	+ elif peers and peer not in peers:
	503	+ continue
	504	+
	505	+ operation = self.__operations_by_peer[peer]
	506	+ # Messages are pairs of (Exception, Operation,):
	507	+ if not notify_cancelled and operation.name in self.__operations_cancelled:
	508	+ continue
	509	+ elif operation.name not in self.__operations_cancelled:
	510	+ message = (None, self._copy_operation(operation),)
	511	+ else:
	512	+ message = (CancelledError("Operation has been cancelled"),
	513	+ self._copy_operation(operation),)
	514	+
	515	+ message_queue.put(message)

buildgrid/server/operations/instance.py

---

...	...	@@ -21,7 +21,7 @@ An instance of the LongRunningOperations Service.
21	21
22	22	import logging
23	23
24		-from buildgrid._exceptions import InvalidArgumentError
	24	+from buildgrid._exceptions import InvalidArgumentError, NotFoundError
25	25	from buildgrid._protos.google.longrunning import operations_pb2
26	26
27	27
...	...	@@ -39,62 +39,43 @@ class OperationsInstance:
39	39	def register_instance_with_server(self, instance_name, server):
40	40	server.add_operations_instance(self, instance_name)
41	41
42		- def get_operation(self, name):
43		- job = self._scheduler.jobs.get(name)
	42	+ def get_operation(self, job_name):
	43	+ try:
	44	+ operation = self._scheduler.get_job_operation(job_name)
44	45
45		- if job is None:
46		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
	46	+ except NotFoundError:
	47	+ raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))
47	48
48		- else:
49		- return job.operation
	49	+ return operation
50	50
51	51	def list_operations(self, list_filter, page_size, page_token):
52	52	# TODO: Pages
53	53	# Spec says number of pages and length of a page are optional
54	54	response = operations_pb2.ListOperationsResponse()
	55	+
	56	+ operation_names = [operation_name for job_name in
	57	+ self._scheduler.list_current_jobs() for operation_name in
	58	+ self._scheduler.list_job_operations(job_name)]
	59	+
55	60	operations = []
56		- for job in self._scheduler.list_jobs():
57		- op = operations_pb2.Operation()
58		- op.CopyFrom(job.operation)
59		- operations.append(op)
	61	+ for operation_name in operation_names:
	62	+ operation = self._scheduler.get_job_operation(operation_name)
	63	+ operations.append(operation)
60	64
61	65	response.operations.extend(operations)
62	66
63	67	return response
64	68
65		- def delete_operation(self, name):
66		- try:
67		- self._scheduler.jobs.pop(name)
68		-
69		- except KeyError:
70		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
71		-
72		- def cancel_operation(self, name):
	69	+ def delete_operation(self, job_name):
73	70	try:
74		- self._scheduler.cancel_job_operation(name)
	71	+ self._scheduler.delete_job_operation(job_name)
75	72
76		- except KeyError:
77		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
	73	+ except NotFoundError:
	74	+ raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))
78	75
79		- def register_message_client(self, name, queue):
	76	+ def cancel_operation(self, job_name):
80	77	try:
81		- self._scheduler.register_client(name, queue)
82		-
83		- except KeyError:
84		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
85		-
86		- def unregister_message_client(self, name, queue):
87		- try:
88		- self._scheduler.unregister_client(name, queue)
89		-
90		- except KeyError:
91		- raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
92		-
93		- def stream_operation_updates(self, message_queue, operation_name):
94		- job = message_queue.get()
95		- while not job.operation.done:
96		- yield job.operation
97		- job = message_queue.get()
98		- job.check_operation_status()
	78	+ self._scheduler.cancel_job_operation(job_name)
99	79
100		- yield job.operation
	80	+ except NotFoundError:
	81	+ raise InvalidArgumentError("Operation name does not exist: [{}]".format(job_name))

buildgrid/server/scheduler.py

---

...	...	@@ -19,12 +19,13 @@ Scheduler
19	19	Schedules jobs.
20	20	"""
21	21
22		-from collections import deque
	22	+import bisect
23	23	from datetime import timedelta
24	24	import logging
25	25
26	26	from buildgrid._enums import LeaseState, OperationStage
27	27	from buildgrid._exceptions import NotFoundError
	28	+from buildgrid.server.job import Job
28	29
29	30
30	31	class Scheduler:
...	...	@@ -42,8 +43,12 @@ class Scheduler:
42	43	self.__retries_count = 0
43	44
44	45	self._action_cache = action_cache
45		- self.jobs = {}
46		- self.queue = deque()
	46	+
	47	+ self.__jobs_by_action = {} # Action to Job 1:1 mapping
	48	+ self.__jobs_by_operation = {} # Operation to Job 1:1 mapping
	49	+ self.__jobs_by_name = {} # Name to Job 1:1 mapping
	50	+
	51	+ self.__queue = []
47	52
48	53	self._is_instrumented = monitor
49	54
...	...	@@ -52,61 +57,209 @@ class Scheduler:
52	57
53	58	# --- Public API ---
54	59
55		- def register_client(self, job_name, queue):
56		- job = self.jobs[job_name]
	60	+ def list_current_jobs(self):
	61	+ """Returns a list of the :class:`Job` names currently managed."""
	62	+ return self.__jobs_by_name.keys()
	63	+
	64	+ def list_job_operations(self, job_name):
	65	+ """Returns a list of :class:`Operation` names for a :class:`Job`."""
	66	+ if job_name in self.__jobs_by_name:
	67	+ return self.__jobs_by_name[job_name].list_operations()
	68	+ else:
	69	+ return []
	70	+
	71	+ # --- Public API: REAPI ---
	72	+
	73	+ def register_job_peer(self, job_name, peer, message_queue):
	74	+ """Subscribes to the job's :class:`Operation` stage changes.
	75	+
	76	+ Args:
	77	+ job_name (str): name of the job to subscribe to.
	78	+ peer (str): a unique string identifying the client.
	79	+ message_queue (queue.Queue): the event queue to register.
	80	+
	81	+ Returns:
	82	+ str: The name of the subscribed :class:`Operation`.
	83	+
	84	+ Raises:
	85	+ NotFoundError: If no job with `job_name` exists.
	86	+ """
	87	+ try:
	88	+ job = self.__jobs_by_name[job_name]
	89	+
	90	+ except KeyError:
	91	+ raise NotFoundError("Job name does not exist: [{}]"
	92	+ .format(job_name))
	93	+
	94	+ operation_name = job.register_new_operation_peer(peer, message_queue)
	95	+
	96	+ self.__jobs_by_operation[operation_name] = job
	97	+
	98	+ return operation_name
57	99
58		- job.register_client(queue)
	100	+ def register_job_operation_peer(self, operation_name, peer, message_queue):
	101	+ """Subscribes to an existing the job's :class:`Operation` stage changes.
	102	+
	103	+ Args:
	104	+ operation_name (str): name of the operation to subscribe to.
	105	+ peer (str): a unique string identifying the client.
	106	+ message_queue (queue.Queue): the event queue to register.
	107	+
	108	+ Returns:
	109	+ str: The name of the subscribed :class:`Operation`.
	110	+
	111	+ Raises:
	112	+ NotFoundError: If no operation with `operation_name` exists.
	113	+ """
	114	+ try:
	115	+ job = self.__jobs_by_operation[operation_name]
	116	+
	117	+ except KeyError:
	118	+ raise NotFoundError("Operation name does not exist: [{}]"
	119	+ .format(operation_name))
	120	+
	121	+ job.register_operation_peer(operation_name, peer, message_queue)
	122	+
	123	+ def unregister_job_operation_peer(self, operation_name, peer):
	124	+ """Unsubscribes to one of the job's :class:`Operation` stage change.
	125	+
	126	+ Args:
	127	+ operation_name (str): name of the operation to unsubscribe from.
	128	+ peer (str): a unique string identifying the client.
	129	+
	130	+ Raises:
	131	+ NotFoundError: If no operation with `operation_name` exists.
	132	+ """
	133	+ try:
	134	+ job = self.__jobs_by_operation[operation_name]
	135	+
	136	+ except KeyError:
	137	+ raise NotFoundError("Operation name does not exist: [{}]"
	138	+ .format(operation_name))
59	139
60		- def unregister_client(self, job_name, queue):
61		- job = self.jobs[job_name]
	140	+ job.unregister_operation_peer(operation_name, peer)
62	141
63		- job.unregister_client(queue)
	142	+ if not job.n_peers_for_operation(operation_name):
	143	+ del self.__jobs_by_operation[operation_name]
64	144
65		- if not job.n_clients and job.operation.done and not job.lease:
	145	+ if not job.n_peers and job.done and not job.lease:
66	146	self._delete_job(job.name)
67	147
68		- def queue_job(self, job, skip_cache_lookup=False):
69		- self.jobs[job.name] = job
	148	+ def queue_job_operation(self, action, action_digest, priority=0, skip_cache_lookup=False):
	149	+ """Inserts a newly created job into the execution queue.
	150	+
	151	+ Warning:
	152	+ Priority is handle like a POSIX ``nice`` values: a higher value
	153	+ means a low priority, 0 being default priority.
	154	+
	155	+ Args:
	156	+ action (Action): the given action to queue for execution.
	157	+ action_digest (Digest): the digest of the given action.
	158	+ priority (int): the execution job's priority.
	159	+ skip_cache_lookup (bool): whether or not to look for pre-computed
	160	+ result for the given action.
	161	+
	162	+ Returns:
	163	+ str: the newly created operation's name.
	164	+ """
	165	+ if action_digest.hash in self.__jobs_by_action:
	166	+ job = self.__jobs_by_action[action_digest.hash]
	167	+
	168	+ # Reschedule if priority is now greater:
	169	+ if priority < job.priority:
	170	+ job.priority = priority
	171	+
	172	+ if job.operation_stage == OperationStage.QUEUED:
	173	+ self._queue_job(job.name)
	174	+
	175	+ return job.name
	176	+
	177	+ job = Job(action, action_digest, priority=priority)
	178	+
	179	+ self.__jobs_by_action[job.action_digest.hash] = job
	180	+ self.__jobs_by_name[job.name] = job
70	181
71	182	operation_stage = None
	183	+
72	184	if self._action_cache is not None and not skip_cache_lookup:
73	185	try:
74	186	action_result = self._action_cache.get_action_result(job.action_digest)
	187	+
75	188	except NotFoundError:
76	189	operation_stage = OperationStage.QUEUED
77		- self.queue.append(job)
	190	+ self._queue_job(job.name)
78	191
79	192	else:
80		- job.set_cached_result(action_result)
81	193	operation_stage = OperationStage.COMPLETED
	194	+ job.set_cached_result(action_result)
82	195
83	196	if self._is_instrumented:
84	197	self.__retries_count += 1
85	198
86	199	else:
87	200	operation_stage = OperationStage.QUEUED
88		- self.queue.append(job)
	201	+ self._queue_job(job.name)
89	202
90	203	self._update_job_operation_stage(job.name, operation_stage)
91	204
92		- def retry_job(self, job_name):
93		- job = self.jobs[job_name]
	205	+ return job.name
94	206
95		- operation_stage = None
96		- if job.n_tries >= self.MAX_N_TRIES:
97		- # TODO: Decide what to do with these jobs
98		- operation_stage = OperationStage.COMPLETED
99		- # TODO: Mark these jobs as done
	207	+ def get_job_operation(self, operation_name):
	208	+ """Retrieves a job's :class:`Operation` by name.
100	209
101		- else:
102		- operation_stage = OperationStage.QUEUED
103		- job.update_lease_state(LeaseState.PENDING)
104		- self.queue.append(job)
	210	+ Args:
	211	+ operation_name (str): name of the operation to query.
105	212
106		- self._update_job_operation_stage(job_name, operation_stage)
	213	+ Raises:
	214	+ NotFoundError: If no operation with `operation_name` exists.
	215	+ """
	216	+ try:
	217	+ job = self.__jobs_by_operation[operation_name]
	218	+
	219	+ except KeyError:
	220	+ raise NotFoundError("Operation name does not exist: [{}]"
	221	+ .format(operation_name))
	222	+
	223	+ return job.get_operation(operation_name)
	224	+
	225	+ def cancel_job_operation(self, operation_name):
	226	+ """"Cancels a job's :class:`Operation` by name.
	227	+
	228	+ Args:
	229	+ operation_name (str): name of the operation to cancel.
	230	+
	231	+ Raises:
	232	+ NotFoundError: If no operation with `operation_name` exists.
	233	+ """
	234	+ try:
	235	+ job = self.__jobs_by_operation[operation_name]
	236	+
	237	+ except KeyError:
	238	+ raise NotFoundError("Operation name does not exist: [{}]"
	239	+ .format(operation_name))
	240	+
	241	+ job.cancel_operation(operation_name)
	242	+
	243	+ def delete_job_operation(self, operation_name):
	244	+ """"Removes a job.
	245	+
	246	+ Args:
	247	+ operation_name (str): name of the operation to delete.
	248	+
	249	+ Raises:
	250	+ NotFoundError: If no operation with `operation_name` exists.
	251	+ """
	252	+ try:
	253	+ job = self.__jobs_by_operation[operation_name]
	254	+
	255	+ except KeyError:
	256	+ raise NotFoundError("Operation name does not exist: [{}]"
	257	+ .format(operation_name))
	258	+
	259	+ if not job.n_peers and job.done and not job.lease:
	260	+ self._delete_job(job.name)
107	261
108		- def list_jobs(self):
109		- return self.jobs.values()
	262	+ # --- Public API: RWAPI ---
110	263
111	264	def request_job_leases(self, worker_capabilities):
112	265	"""Generates a list of the highest priority leases to be run.
...	...	@@ -115,11 +268,14 @@ class Scheduler:
115	268	worker_capabilities (dict): a set of key-value pairs decribing the
116	269	worker properties, configuration and state at the time of the
117	270	request.
	271	+
	272	+ Warning: Worker capabilities handling is not implemented at the moment!
118	273	"""
119		- if not self.queue:
	274	+ if not self.__queue:
120	275	return []
121	276
122		- job = self.queue.popleft()
	277	+ # TODO: Try to match worker_capabilities with jobs properties.
	278	+ job = self.__queue.pop()
123	279
124	280	lease = job.lease
125	281
...	...	@@ -132,18 +288,25 @@ class Scheduler:
132	288
133	289	return None
134	290
135		- def update_job_lease(self, lease):
	291	+ def update_job_lease_state(self, job_name, lease):
136	292	"""Requests a state transition for a job's current :class:Lease.
137	293
	294	+ Note:
	295	+ This may trigger a job's :class:`Operation` stage transition.
	296	+
138	297	Args:
139		- job_name (str): name of the job to query.
140		- lease_state (LeaseState): the lease state to transition to.
141		- lease_status (google.rpc.Status): the lease execution status, only
142		- required if `lease_state` is `COMPLETED`.
143		- lease_result (google.protobuf.Any): the lease execution result, only
144		- required if `lease_state` is `COMPLETED`.
	298	+ job_name (str): name of the job to update lease state from.
	299	+ lease (Lease): the lease holding the new state.
	300	+
	301	+ Raises:
	302	+ NotFoundError: If no job with `job_name` exists.
145	303	"""
146		- job = self.jobs[lease.id]
	304	+ try:
	305	+ job = self.__jobs_by_name[job_name]
	306	+
	307	+ except KeyError:
	308	+ raise NotFoundError("Job name does not exist: [{}]".format(job_name))
	309	+
147	310	lease_state = LeaseState(lease.state)
148	311
149	312	operation_stage = None
...	...	@@ -179,38 +342,93 @@ class Scheduler:
179	342	self.__leases_by_state[LeaseState.ACTIVE].discard(lease.id)
180	343	self.__leases_by_state[LeaseState.COMPLETED].add(lease.id)
181	344
182		- self._update_job_operation_stage(lease.id, operation_stage)
	345	+ self._update_job_operation_stage(job_name, operation_stage)
	346	+
	347	+ def retry_job_lease(self, job_name):
	348	+ """Re-queues a job on lease execution failure.
	349	+
	350	+ Note:
	351	+ This may trigger a job's :class:`Operation` stage transition.
	352	+
	353	+ Args:
	354	+ job_name (str): name of the job to retry the lease from.
	355	+
	356	+ Raises:
	357	+ NotFoundError: If no job with `job_name` exists.
	358	+ """
	359	+ try:
	360	+ job = self.__jobs_by_name[job_name]
	361	+
	362	+ except KeyError:
	363	+ raise NotFoundError("Job name does not exist: [{}]".format(job_name))
	364	+
	365	+ operation_stage = None
	366	+ if job.n_tries >= self.MAX_N_TRIES:
	367	+ # TODO: Decide what to do with these jobs
	368	+ operation_stage = OperationStage.COMPLETED
	369	+ # TODO: Mark these jobs as done
	370	+
	371	+ else:
	372	+ operation_stage = OperationStage.QUEUED
	373	+ self._queue_job(job.name)
	374	+
	375	+ job.update_lease_state(LeaseState.PENDING)
	376	+
	377	+ self._update_job_operation_stage(job_name, operation_stage)
183	378
184	379	def get_job_lease(self, job_name):
185		- """Returns the lease associated to job, if any have been emitted yet."""
186		- return self.jobs[job_name].lease
	380	+ """Returns the lease associated to job, if any have been emitted yet.
187	381
188		- def get_job_lease_cancelled(self, job_name):
189		- """Returns true if the lease is cancelled"""
190		- return self.jobs[job_name].lease_cancelled
	382	+ Args:
	383	+ job_name (str): name of the job to query the lease from.
	384	+
	385	+ Raises:
	386	+ NotFoundError: If no job with `job_name` exists.
	387	+ """
	388	+ try:
	389	+ job = self.__jobs_by_name[job_name]
	390	+
	391	+ except KeyError:
	392	+ raise NotFoundError("Job name does not exist: [{}]".format(job_name))
	393	+
	394	+ return job.lease
191	395
192	396	def delete_job_lease(self, job_name):
193		- """Discards the lease associated to a job."""
194		- job = self.jobs[job_name]
	397	+ """Discards the lease associated with a job.
195	398
196		- self.jobs[job.name].delete_lease()
	399	+ Args:
	400	+ job_name (str): name of the job to delete the lease from.
197	401
198		- if not job.n_clients and job.operation.done:
199		- self._delete_job(job.name)
	402	+ Raises:
	403	+ NotFoundError: If no job with `job_name` exists.
	404	+ """
	405	+ try:
	406	+ job = self.__jobs_by_name[job_name]
200	407
201		- def get_job_operation(self, job_name):
202		- """Returns the operation associated to job."""
203		- return self.jobs[job_name].operation
	408	+ except KeyError:
	409	+ raise NotFoundError("Job name does not exist: [{}]".format(job_name))
204	410
205		- def cancel_job_operation(self, job_name):
206		- """"Cancels the underlying operation of a given job.
	411	+ job.delete_lease()
207	412
208		- This will also cancel any job's lease that may have been issued.
	413	+ if not job.n_peers and job.operation.done:
	414	+ self._delete_job(job.name)
	415	+
	416	+ def get_job_lease_cancelled(self, job_name):
	417	+ """Returns true if the lease is cancelled.
209	418
210	419	Args:
211		- job_name (str): name of the job holding the operation to cancel.
	420	+ job_name (str): name of the job to query the lease state from.
	421	+
	422	+ Raises:
	423	+ NotFoundError: If no job with `job_name` exists.
212	424	"""
213		- self.jobs[job_name].cancel_operation()
	425	+ try:
	426	+ job = self.__jobs_by_name[job_name]
	427	+
	428	+ except KeyError:
	429	+ raise NotFoundError("Job name does not exist: [{}]".format(job_name))
	430	+
	431	+ return job.lease_cancelled
214	432
215	433	# --- Public API: Monitoring ---
216	434
...	...	@@ -260,11 +478,11 @@ class Scheduler:
260	478	self.__build_metadata_queues.append(message_queue)
261	479
262	480	def query_n_jobs(self):
263		- return len(self.jobs)
	481	+ return len(self.__jobs_by_name)
264	482
265	483	def query_n_operations(self):
266	484	# For now n_operations == n_jobs:
267		- return len(self.jobs)
	485	+ return len(self.__jobs_by_operation)
268	486
269	487	def query_n_operations_by_stage(self, operation_stage):
270	488	try:
...	...	@@ -275,7 +493,7 @@ class Scheduler:
275	493	return 0
276	494
277	495	def query_n_leases(self):
278		- return len(self.jobs)
	496	+ return len(self.__jobs_by_name)
279	497
280	498	def query_n_leases_by_state(self, lease_state):
281	499	try:
...	...	@@ -295,19 +513,35 @@ class Scheduler:
295	513
296	514	# --- Private API ---
297	515
	516	+ def _queue_job(self, job_name):
	517	+ """Schedules or reschedules a job."""
	518	+ job = self.__jobs_by_name[job_name]
	519	+
	520	+ if job.operation_stage == OperationStage.QUEUED:
	521	+ self.__queue.sort()
	522	+
	523	+ else:
	524	+ bisect.insort(self.__queue, job)
	525	+
298	526	def _delete_job(self, job_name):
299	527	"""Drops an entry from the internal list of jobs."""
300		- del self.jobs[job_name]
	528	+ job = self.__jobs_by_name[job_name]
	529	+
	530	+ if job.operation_stage == OperationStage.QUEUED:
	531	+ self.__queue.remove(job)
	532	+
	533	+ del self.__jobs_by_action[job.action_digest.hash]
	534	+ del self.__jobs_by_name[job.name]
301	535
302	536	if self._is_instrumented:
303		- self.__operations_by_stage[OperationStage.CACHE_CHECK].discard(job_name)
304		- self.__operations_by_stage[OperationStage.QUEUED].discard(job_name)
305		- self.__operations_by_stage[OperationStage.EXECUTING].discard(job_name)
306		- self.__operations_by_stage[OperationStage.COMPLETED].discard(job_name)
	537	+ self.__operations_by_stage[OperationStage.CACHE_CHECK].discard(job.name)
	538	+ self.__operations_by_stage[OperationStage.QUEUED].discard(job.name)
	539	+ self.__operations_by_stage[OperationStage.EXECUTING].discard(job.name)
	540	+ self.__operations_by_stage[OperationStage.COMPLETED].discard(job.name)
307	541
308		- self.__leases_by_state[LeaseState.PENDING].discard(job_name)
309		- self.__leases_by_state[LeaseState.ACTIVE].discard(job_name)
310		- self.__leases_by_state[LeaseState.COMPLETED].discard(job_name)
	542	+ self.__leases_by_state[LeaseState.PENDING].discard(job.name)
	543	+ self.__leases_by_state[LeaseState.ACTIVE].discard(job.name)
	544	+ self.__leases_by_state[LeaseState.COMPLETED].discard(job.name)
311	545
312	546	def _update_job_operation_stage(self, job_name, operation_stage):
313	547	"""Requests a stage transition for the job's :class:Operations.
...	...	@@ -316,7 +550,7 @@ class Scheduler:
316	550	job_name (str): name of the job to query.
317	551	operation_stage (OperationStage): the stage to transition to.
318	552	"""
319		- job = self.jobs[job_name]
	553	+ job = self.__jobs_by_name[job_name]
320	554
321	555	if operation_stage == OperationStage.CACHE_CHECK:
322	556	job.update_operation_stage(OperationStage.CACHE_CHECK)
...	...	@@ -365,7 +599,7 @@ class Scheduler:
365	599
366	600	self.__queue_time_average = average_order, average_time
367	601
368		- if not job.holds_cached_action_result:
	602	+ if not job.holds_cached_result:
369	603	execution_metadata = job.action_result.execution_metadata
370	604	context_metadata = {'job-is': job.name}
371	605

docs/source/conf.py

---

...	...	@@ -182,3 +182,11 @@ texinfo_documents = [
182	182	author, 'BuildGrid', 'One line description of project.',
183	183	'Miscellaneous'),
184	184	]
	185	+
	186	+# -- Options for the autodoc extension ----------------------------------------
	187	+
	188	+# This value selects if automatically documented members are sorted
	189	+# alphabetical (value 'alphabetical'), by member type (value 'groupwise') or
	190	+# by source order (value 'bysource'). The default is alphabetical.
	191	+autodoc_member_order = 'bysource'
	192	+

tests/integration/bots_service.py

---

...	...	@@ -25,7 +25,6 @@ import pytest
25	25
26	26	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
27	27	from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
28		-from buildgrid.server import job
29	28	from buildgrid.server.controller import ExecutionController
30	29	from buildgrid.server.job import LeaseState
31	30	from buildgrid.server.bots import service
...	...	@@ -159,7 +158,8 @@ def test_post_bot_event_temp(context, instance):
159	158	def _inject_work(scheduler, action=None, action_digest=None):
160	159	if not action:
161	160	action = remote_execution_pb2.Action()
	161	+
162	162	if not action_digest:
163	163	action_digest = remote_execution_pb2.Digest()
164		- j = job.Job(action, action_digest)
165		- scheduler.queue_job(j, True)
	164	+
	165	+ scheduler.queue_job_operation(action, action_digest, skip_cache_lookup=True)

tests/integration/execution_service.py

---

...	...	@@ -17,14 +17,15 @@
17	17
18	18	# pylint: disable=redefined-outer-name
19	19
	20	+import queue
20	21	import uuid
21	22	from unittest import mock
22	23
23		-from google.protobuf import any_pb2
24	24	import grpc
25	25	from grpc._server import _Context
26	26	import pytest
27	27
	28	+from buildgrid._enums import OperationStage
28	29	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
29	30	from buildgrid._protos.google.longrunning import operations_pb2
30	31
...	...	@@ -82,7 +83,7 @@ def test_execute(skip_cache_lookup, instance, context):
82	83	assert isinstance(result, operations_pb2.Operation)
83	84	metadata = remote_execution_pb2.ExecuteOperationMetadata()
84	85	result.metadata.Unpack(metadata)
85		- assert metadata.stage == job.OperationStage.QUEUED.value
	86	+ assert metadata.stage == OperationStage.QUEUED.value
86	87	operation_uuid = result.name.split('/')[-1]
87	88	assert uuid.UUID(operation_uuid, version=4)
88	89	assert result.done is False
...	...	@@ -106,18 +107,19 @@ def test_no_action_digest_in_storage(instance, context):
106	107
107	108
108	109	def test_wait_execution(instance, controller, context):
109		- j = job.Job(action, action_digest)
110		- j._operation.done = True
	110	+ job_name = controller.execution_instance._scheduler.queue_job_operation(action,
	111	+ action_digest,
	112	+ skip_cache_lookup=True)
111	113
112		- request = remote_execution_pb2.WaitExecutionRequest(name=j.name)
	114	+ message_queue = queue.Queue()
	115	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	116	+ context.peer(),
	117	+ message_queue)
113	118
114		- controller.execution_instance._scheduler.jobs[j.name] = j
	119	+ controller.execution_instance._scheduler._update_job_operation_stage(job_name,
	120	+ OperationStage.COMPLETED)
115	121
116		- action_result_any = any_pb2.Any()
117		- action_result = remote_execution_pb2.ActionResult()
118		- action_result_any.Pack(action_result)
119		-
120		- j.update_operation_stage(job.OperationStage.COMPLETED)
	122	+ request = remote_execution_pb2.WaitExecutionRequest(name=operation_name)
121	123
122	124	response = instance.WaitExecution(request, context)
123	125
...	...	@@ -127,7 +129,6 @@ def test_wait_execution(instance, controller, context):
127	129	metadata = remote_execution_pb2.ExecuteOperationMetadata()
128	130	result.metadata.Unpack(metadata)
129	131	assert metadata.stage == job.OperationStage.COMPLETED.value
130		- assert uuid.UUID(result.name, version=4)
131	132	assert result.done is True
132	133
133	134

tests/integration/operations_service.py

---

...	...	@@ -17,6 +17,7 @@
17	17
18	18	# pylint: disable=redefined-outer-name
19	19
	20	+import queue
20	21	from unittest import mock
21	22
22	23	from google.protobuf import any_pb2
...	...	@@ -86,8 +87,13 @@ def blank_instance(controller):
86	87
87	88	# Queue an execution, get operation corresponding to that request
88	89	def test_get_operation(instance, controller, execute_request, context):
89		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
90		- execute_request.skip_cache_lookup)
	90	+ job_name = controller.execution_instance.execute(execute_request.action_digest,
	91	+ execute_request.skip_cache_lookup)
	92	+
	93	+ message_queue = queue.Queue()
	94	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	95	+ context.peer(),
	96	+ message_queue)
91	97
92	98	request = operations_pb2.GetOperationRequest()
93	99
...	...	@@ -95,25 +101,28 @@ def test_get_operation(instance, controller, execute_request, context):
95	101	# we're manually creating the instance here, it doesn't get a name.
96	102	# Therefore we need to manually add the instance name to the operation
97	103	# name in the GetOperation request.
98		- request.name = "{}/{}".format(instance_name, response_execute.name)
	104	+ request.name = "{}/{}".format(instance_name, operation_name)
99	105
100	106	response = instance.GetOperation(request, context)
101		- assert response.name == "{}/{}".format(instance_name, response_execute.name)
102		- assert response.done == response_execute.done
	107	+ assert response.name == "{}/{}".format(instance_name, operation_name)
103	108
104	109
105	110	# Queue an execution, get operation corresponding to that request
106	111	def test_get_operation_blank(blank_instance, controller, execute_request, context):
107		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
108		- execute_request.skip_cache_lookup)
	112	+ job_name = controller.execution_instance.execute(execute_request.action_digest,
	113	+ execute_request.skip_cache_lookup)
	114	+
	115	+ message_queue = queue.Queue()
	116	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	117	+ context.peer(),
	118	+ message_queue)
109	119
110	120	request = operations_pb2.GetOperationRequest()
111	121
112		- request.name = response_execute.name
	122	+ request.name = operation_name
113	123
114	124	response = blank_instance.GetOperation(request, context)
115		- assert response.name == response_execute.name
116		- assert response.done == response_execute.done
	125	+ assert response.name == operation_name
117	126
118	127
119	128	def test_get_operation_fail(instance, context):
...	...	@@ -133,25 +142,35 @@ def test_get_operation_instance_fail(instance, context):
133	142
134	143
135	144	def test_list_operations(instance, controller, execute_request, context):
136		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
137		- execute_request.skip_cache_lookup)
	145	+ job_name = controller.execution_instance.execute(execute_request.action_digest,
	146	+ execute_request.skip_cache_lookup)
	147	+
	148	+ message_queue = queue.Queue()
	149	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	150	+ context.peer(),
	151	+ message_queue)
138	152
139	153	request = operations_pb2.ListOperationsRequest(name=instance_name)
140	154	response = instance.ListOperations(request, context)
141	155
142	156	names = response.operations[0].name.split('/')
143	157	assert names[0] == instance_name
144		- assert names[1] == response_execute.name
	158	+ assert names[1] == operation_name
145	159
146	160
147	161	def test_list_operations_blank(blank_instance, controller, execute_request, context):
148		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
149		- execute_request.skip_cache_lookup)
	162	+ job_name = controller.execution_instance.execute(execute_request.action_digest,
	163	+ execute_request.skip_cache_lookup)
	164	+
	165	+ message_queue = queue.Queue()
	166	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	167	+ context.peer(),
	168	+ message_queue)
150	169
151	170	request = operations_pb2.ListOperationsRequest(name='')
152	171	response = blank_instance.ListOperations(request, context)
153	172
154		- assert response.operations[0].name.split('/')[-1] == response_execute.name
	173	+ assert response.operations[0].name.split('/')[-1] == operation_name
155	174
156	175
157	176	def test_list_operations_instance_fail(instance, controller, execute_request, context):
...	...	@@ -174,14 +193,19 @@ def test_list_operations_empty(instance, context):
174	193
175	194	# Send execution off, delete, try to find operation should fail
176	195	def test_delete_operation(instance, controller, execute_request, context):
177		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
178		- execute_request.skip_cache_lookup)
	196	+ job_name = controller.execution_instance.execute(execute_request.action_digest,
	197	+ execute_request.skip_cache_lookup)
	198	+
	199	+ message_queue = queue.Queue()
	200	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	201	+ context.peer(),
	202	+ message_queue)
179	203
180	204	request = operations_pb2.DeleteOperationRequest()
181		- request.name = response_execute.name
	205	+ request.name = operation_name
182	206	instance.DeleteOperation(request, context)
183	207
184		- request_name = "{}/{}".format(instance_name, response_execute.name)
	208	+ request_name = "{}/{}".format(instance_name, operation_name)
185	209
186	210	with pytest.raises(InvalidArgumentError):
187	211	controller.operations_instance.get_operation(request_name)
...	...	@@ -189,17 +213,11 @@ def test_delete_operation(instance, controller, execute_request, context):
189	213
190	214	# Send execution off, delete, try to find operation should fail
191	215	def test_delete_operation_blank(blank_instance, controller, execute_request, context):
192		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
193		- execute_request.skip_cache_lookup)
194		-
195	216	request = operations_pb2.DeleteOperationRequest()
196		- request.name = response_execute.name
	217	+ request.name = "runner"
197	218	blank_instance.DeleteOperation(request, context)
198	219
199		- request_name = response_execute.name
200		-
201		- with pytest.raises(InvalidArgumentError):
202		- controller.operations_instance.get_operation(request_name)
	220	+ context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
203	221
204	222
205	223	def test_delete_operation_fail(instance, context):
...	...	@@ -211,11 +229,16 @@ def test_delete_operation_fail(instance, context):
211	229
212	230
213	231	def test_cancel_operation(instance, controller, execute_request, context):
214		- response_execute = controller.execution_instance.execute(execute_request.action_digest,
215		- execute_request.skip_cache_lookup)
	232	+ job_name = controller.execution_instance.execute(execute_request.action_digest,
	233	+ execute_request.skip_cache_lookup)
	234	+
	235	+ message_queue = queue.Queue()
	236	+ operation_name = controller.execution_instance.register_job_peer(job_name,
	237	+ context.peer(),
	238	+ message_queue)
216	239
217	240	request = operations_pb2.CancelOperationRequest()
218		- request.name = "{}/{}".format(instance_name, response_execute.name)
	241	+ request.name = "{}/{}".format(instance_name, operation_name)
219	242
220	243	instance.CancelOperation(request, context)
221	244
...	...	@@ -238,7 +261,7 @@ def test_cancel_operation_blank(blank_instance, context):
238	261	context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
239	262
240	263
241		-def test_cancel_operation_instance_fail(instance, context):
	264	+def test_cancel_operation__fail(instance, context):
242	265	request = operations_pb2.CancelOperationRequest()
243	266	instance.CancelOperation(request, context)
244	267

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