[Notes] [Git][BuildGrid/buildgrid][finn/74-operation-cancelation] 6 commits: Implement cancelletion methods in job class.

[finn <gitlab mg gitlab com>]

Title: GitLab

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

Commits:

• 75d927c2
  by Finn at 2018-11-01T11:06:52Z

  Implement cancelletion methods in job class.
  

• 68254f2b
  by Finn at 2018-11-01T11:06:56Z

  Added cancellation method to scheduler.
  

• 34c9e33f
  by Finn at 2018-11-01T11:06:56Z

  Added cancellation to operations instance.
  

• 13efe2a0
  by Finn at 2018-11-01T11:41:32Z

  Added cancel command to app.
  

• c5118986
  by Finn at 2018-11-01T13:27:26Z

  Updating unittests for cancellation of operation.
  

• 0c891929
  by Finn at 2018-11-01T13:28:54Z

  Working commit
  

6 changed files:

• buildgrid/_app/commands/cmd_operation.py
• buildgrid/server/job.py
• buildgrid/server/operations/instance.py
• buildgrid/server/operations/service.py
• buildgrid/server/scheduler.py
• tests/integration/operations_service.py

Changes:

buildgrid/_app/commands/cmd_operation.py

---

...	...	@@ -155,6 +155,19 @@ def status(context, operation_name, json):
155	155	click.echo(json_format.MessageToJson(operation))
156	156
157	157
	158	+@cli.command('cancel', short_help="Cancel an operation.")
	159	+@click.argument('operation-name', nargs=1, type=click.STRING, required=True)
	160	+@pass_context
	161	+def cancel(context, operation_name):
	162	+ context.logger.info("Cancelling an operation...")
	163	+ stub = operations_pb2_grpc.OperationsStub(context.channel)
	164	+
	165	+ request = operations_pb2.CancelOperationRequest(name=operation_name)
	166	+
	167	+ stub.CancelOperation(request)
	168	+ context.logger.info("Operation cancelled: [{}]".format(request))
	169	+
	170	+
158	171	@cli.command('list', short_help="List operations.")
159	172	@click.option('--json', is_flag=True, show_default=True,
160	173	help="Print operations list in JSON format.")

buildgrid/server/job.py

---

...	...	@@ -19,6 +19,7 @@ import uuid
19	19	from google.protobuf import timestamp_pb2
20	20
21	21	from buildgrid._enums import LeaseState, OperationStage
	22	+from buildgrid._exceptions import CancelledError
22	23	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
23	24	from buildgrid._protos.google.devtools.remoteworkers.v1test2 import bots_pb2
24	25	from buildgrid._protos.google.longrunning import operations_pb2
...	...	@@ -37,10 +38,14 @@ class Job:
37	38
38	39	self.__execute_response = None
39	40	self.__operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
	41	+
40	42	self.__queued_timestamp = timestamp_pb2.Timestamp()
41	43	self.__worker_start_timestamp = timestamp_pb2.Timestamp()
42	44	self.__worker_completed_timestamp = timestamp_pb2.Timestamp()
43	45
	46	+ self.__operation_cancelled = False
	47	+ self.__lease_cancelled = False
	48	+
44	49	self.__operation_metadata.action_digest.CopyFrom(action_digest)
45	50	self.__operation_metadata.stage = OperationStage.UNKNOWN.value
46	51
...	...	@@ -131,7 +136,9 @@ class Job:
131	136	Only one :class:`Lease` can be emitted for a given job. This method
132	137	should only be used once, any furhter calls are ignored.
133	138	"""
134		- if self._lease is not None:
	139	+ if self.__operation_cancelled:
	140	+ return None
	141	+ elif self._lease is not None:
135	142	return None
136	143
137	144	self._lease = bots_pb2.Lease()
...	...	@@ -189,6 +196,15 @@ class Job:
189	196	self.__execute_response.cached_result = False
190	197	self.__execute_response.status.CopyFrom(status)
191	198
	199	+ def cancel_lease(self):
	200	+ """Triggers a job's :class:Lease cancellation.
	201	+
	202	+ This will not cancel the job's :class:Operation.
	203	+ """
	204	+ self.__lease_cancelled = True
	205	+ if self._lease is not None:
	206	+ self.update_lease_state(LeaseState.CANCELLED)
	207	+
192	208	def update_operation_stage(self, stage):
193	209	"""Operates a stage transition for the job's :class:Operation.
194	210
...	...	@@ -214,3 +230,20 @@ class Job:
214	230
215	231	for queue in self._operation_update_queues:
216	232	queue.put(self._operation)
	233	+
	234	+ def cancel_operation(self):
	235	+ """Triggers a job's :class:Operation cancellation.
	236	+
	237	+ This will also cancel any job's :class:Lease that may have been issued.
	238	+ """
	239	+ self.__operation_cancelled = True
	240	+ if self._lease is not None:
	241	+ self.cancel_lease()
	242	+
	243	+ self.__execute_response = remote_execution_pb2.ExecuteResponse()
	244	+ self.__execute_response.status.code = code_pb2.CANCELLED
	245	+ self.__execute_response.status.message = "Operation cancelled by client."
	246	+
	247	+ self.update_operation_stage(OperationStage.COMPLETED)
	248	+
	249	+ raise CancelledError("Operation cancelled: {}".format(self._name))

buildgrid/server/operations/instance.py

---

...	...	@@ -64,6 +64,13 @@ class OperationsInstance:
64	64	except KeyError:
65	65	raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
66	66
	67	+ def cancel_operation(self, name):
	68	+ try:
	69	+ self._scheduler.cancel_job_operation(name)
	70	+
	71	+ except KeyError:
	72	+ raise InvalidArgumentError("Operation name does not exist: [{}]".format(name))
	73	+
67	74	def register_message_client(self, name, queue):
68	75	try:
69	76	self._scheduler.register_client(name, queue)
...	...	@@ -84,7 +91,3 @@ class OperationsInstance:
84	91	yield operation
85	92	operation = message_queue.get()
86	93	yield operation
87		-
88		- def cancel_operation(self, name):
89		- # TODO: Cancel leases
90		- raise NotImplementedError("Cancelled operations not supported")

buildgrid/server/operations/service.py

---

...	...	@@ -115,7 +115,7 @@ class OperationsService(operations_pb2_grpc.OperationsServicer):
115	115	except CancelledError as e:
116	116	self.logger.error(e)
117	117	context.set_details(str(e))
118		- context.set_code(grpc.StatusCode.UNIMPLEMENTED)
	118	+ context.set_code(grpc.StatusCode.CANCELLED)
119	119
120	120	except InvalidArgumentError as e:
121	121	self.logger.error(e)

buildgrid/server/scheduler.py

---

...	...	@@ -94,7 +94,10 @@ class Scheduler:
94	94	# For now, one lease at a time:
95	95	lease = job.create_lease()
96	96
97		- return [lease]
	97	+ if lease:
	98	+ return [lease]
	99	+
	100	+ return None
98	101
99	102	def update_job_lease_state(self, job_name, lease_state, lease_status=None, lease_result=None):
100	103	"""Requests a state transition for a job's current :class:Lease.
...	...	@@ -133,3 +136,13 @@ class Scheduler:
133	136	def get_job_operation(self, job_name):
134	137	"""Returns the operation associated to job."""
135	138	return self.jobs[job_name].operation
	139	+
	140	+ def cancel_job_operation(self, job_name):
	141	+ """"Cancels the underlying operation of a given job.
	142	+
	143	+ This will also cancel any job's lease that may have been issued.
	144	+
	145	+ Args:
	146	+ job_name (str): name of the job holding the operation to cancel.
	147	+ """
	148	+ self.jobs[job_name].cancel_operation()

tests/integration/operations_service.py

---

...	...	@@ -24,6 +24,7 @@ import grpc
24	24	from grpc._server import _Context
25	25	import pytest
26	26
	27	+from buildgrid._enums import OperationStage
27	28	from buildgrid._exceptions import InvalidArgumentError
28	29	from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
29	30	from buildgrid._protos.google.longrunning import operations_pb2
...	...	@@ -236,12 +237,26 @@ def test_delete_operation_fail(instance, context):
236	237	context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
237	238
238	239
239		-def test_cancel_operation(instance, context):
	240	+def test_cancel_operation(instance, controller, execute_request, context):
	241	+ response_execute = controller.execution_instance.execute(execute_request.action_digest,
	242	+ execute_request.skip_cache_lookup)
	243	+
240	244	request = operations_pb2.CancelOperationRequest()
241		- request.name = "{}/{}".format(instance_name, "runner")
	245	+ request.name = "{}/{}".format(instance_name, response_execute.name)
	246	+
242	247	instance.CancelOperation(request, context)
243	248
244		- context.set_code.assert_called_once_with(grpc.StatusCode.UNIMPLEMENTED)
	249	+ context.set_code.assert_called_once_with(grpc.StatusCode.CANCELLED)
	250	+
	251	+ request = operations_pb2.ListOperationsRequest(name=instance_name)
	252	+ response = instance.ListOperations(request, context)
	253	+
	254	+ assert len(response.operations) is 1
	255	+
	256	+ for operation in response.operations:
	257	+ operation_metadata = remote_execution_pb2.ExecuteOperationMetadata()
	258	+ operation.metadata.Unpack(operation_metadata)
	259	+ assert operation_metadata.stage == OperationStage.COMPLETED.value
245	260
246	261
247	262	def test_cancel_operation_blank(blank_instance, context):
...	...	@@ -249,7 +264,7 @@ def test_cancel_operation_blank(blank_instance, context):
249	264	request.name = "runner"
250	265	blank_instance.CancelOperation(request, context)
251	266
252		- context.set_code.assert_called_once_with(grpc.StatusCode.UNIMPLEMENTED)
	267	+ context.set_code.assert_called_once_with(grpc.StatusCode.INVALID_ARGUMENT)
253	268
254	269
255	270	def test_cancel_operation_instance_fail(instance, context):

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