[Notes] [Git][BuildStream/buildstream][Qinusty/message-helpers] messaging: Overhaul of messaging throughout codebase

[Qinusty <gitlab mg gitlab com>]

Title: GitLab

Qinusty pushed to branch Qinusty/message-helpers at BuildStream / buildstream

Commits:

• 9a143dce
  by Josh Smith at 2018-08-16T14:20:58Z

  messaging: Overhaul of messaging throughout codebase
  
  job.py: Remove _message() and use context directly
  
  Each helper method now uses the context method directly
  
  queue.py: Added message helpers
  
  elementjob.py: Added message helpers to add kwarg scheduler=True
  
  Lots of unnecessary imports after previous changes.
  

13 changed files:

• buildstream/_artifactcache/artifactcache.py
• buildstream/_artifactcache/cascache.py
• buildstream/_context.py
• buildstream/_frontend/app.py
• buildstream/_pipeline.py
• buildstream/_platform/linux.py
• buildstream/_project.py
• buildstream/_scheduler/jobs/elementjob.py
• buildstream/_scheduler/jobs/job.py
• buildstream/_scheduler/queues/buildqueue.py
• buildstream/_scheduler/queues/queue.py
• buildstream/_stream.py
• buildstream/plugin.py

Changes:

buildstream/_artifactcache/artifactcache.py

---

...	...	@@ -23,7 +23,6 @@ from collections import Mapping, namedtuple
23	23
24	24	from ..element import _KeyStrength
25	25	from .._exceptions import ArtifactError, ImplError, LoadError, LoadErrorReason
26		-from .._message import Message, MessageType
27	26	from .. import utils
28	27	from .. import _yaml
29	28

buildstream/_artifactcache/cascache.py

---

...	...	@@ -33,7 +33,6 @@ from .._protos.google.bytestream import bytestream_pb2, bytestream_pb2_grpc
33	33	from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
34	34	from .._protos.buildstream.v2 import buildstream_pb2, buildstream_pb2_grpc
35	35
36		-from .._message import MessageType, Message
37	36	from .. import _signals, utils
38	37	from .._exceptions import ArtifactError
39	38
...	...	@@ -348,12 +347,10 @@ class CASCache(ArtifactCache):
348	347	raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
349	348
350	349	if skipped_remote:
351		- self.context.message(Message(
352		- None,
353		- MessageType.SKIPPED,
	350	+ self.context.skipped(
354	351	"Remote ({}) already has {} cached".format(
355	352	remote.spec.url, element._get_brief_display_key())
356		- ))
	353	+ )
357	354	return pushed
358	355
359	356	################################################

buildstream/_context.py

---

...	...	@@ -447,6 +447,9 @@ class Context():
447	447	def failure(self, text, *, plugin=None, **kwargs):
448	448	self._message(text, plugin=plugin, msg_type=MessageType.FAIL, **kwargs)
449	449
	450	+ def skipped(self, text, *, plugin=None, **kwargs):
	451	+ self._message(text, plugin=plugin, msg_type=MessageType.SKIPPED, **kwargs)
	452	+
450	453	def debug(self, text, *, plugin=None, **kwargs):
451	454	self._message(text, plugin=plugin, msg_type=MessageType.DEBUG, **kwargs)
452	455

buildstream/_frontend/app.py

---

...	...	@@ -36,7 +36,7 @@ from .._context import Context
36	36	from .._platform import Platform
37	37	from .._project import Project
38	38	from .._exceptions import BstError, StreamError, LoadError, LoadErrorReason, AppError
39		-from .._message import Message, MessageType, unconditional_messages
	39	+from .._message import MessageType, unconditional_messages
40	40	from .._stream import Stream
41	41	from .._versions import BST_FORMAT_VERSION
42	42	from .. import _yaml

buildstream/_pipeline.py

---

...	...	@@ -24,7 +24,6 @@ import itertools
24	24	from operator import itemgetter
25	25
26	26	from ._exceptions import PipelineError
27		-from ._message import Message, MessageType
28	27	from ._profile import Topics, profile_start, profile_end
29	28	from . import Scope, Consistency
30	29	from ._project import ProjectRefStorage

buildstream/_platform/linux.py

---

...	...	@@ -22,7 +22,6 @@ import subprocess
22	22	from .. import _site
23	23	from .. import utils
24	24	from .._artifactcache.cascache import CASCache
25		-from .._message import Message, MessageType
26	25	from ..sandbox import SandboxBwrap
27	26
28	27	from . import Platform

buildstream/_project.py

---

...	...	@@ -36,7 +36,6 @@ from ._projectrefs import ProjectRefs, ProjectRefStorage
36	36	from ._versions import BST_FORMAT_VERSION
37	37	from ._loader import Loader
38	38	from .element import Element
39		-from ._message import Message, MessageType
40	39	from ._includes import Includes
41	40
42	41
...	...	@@ -334,8 +333,7 @@ class Project():
334	333	for source, ref in redundant_refs
335	334	]
336	335	detail += "\n".join(lines)
337		- self._context.message(
338		- Message(None, MessageType.WARN, "Ignoring redundant source references", detail=detail))
	336	+ self._context.warn("Ignoring redundant source references", detail=detail)
339	337
340	338	return elements
341	339
...	...	@@ -492,13 +490,9 @@ class Project():
492	490
493	491	# Deprecation check
494	492	if fail_on_overlap is not None:
495		- self._context.message(
496		- Message(
497		- None,
498		- MessageType.WARN,
499		- "Use of fail-on-overlap within project.conf " +
500		- "is deprecated. Consider using fatal-warnings instead."
501		- )
	493	+ self._context.warn(
	494	+ "Use of fail-on-overlap within project.conf " +
	495	+ "is deprecated. Consider using fatal-warnings instead."
502	496	)
503	497
504	498	# Load project.refs if it exists, this may be ignored.

buildstream/_scheduler/jobs/elementjob.py

---

...	...	@@ -18,8 +18,6 @@
18	18	#
19	19	from ruamel import yaml
20	20
21		-from ..._message import Message, MessageType
22		-
23	21	from .job import Job
24	22
25	23
...	...	@@ -86,9 +84,8 @@ class ElementJob(Job):
86	84	# This should probably be omitted for non-build tasks but it's harmless here
87	85	elt_env = self._element.get_environment()
88	86	env_dump = yaml.round_trip_dump(elt_env, default_flow_style=False, allow_unicode=True)
89		- self.message(MessageType.LOG,
90		- "Build environment for element {}".format(self._element.name),
91		- detail=env_dump)
	87	+ self._log("Build environment for element {}".format(self._element.name),
	88	+ detail=env_dump)
92	89
93	90	# Run the action
94	91	return self._action_cb(self._element)
...	...	@@ -96,15 +93,6 @@ class ElementJob(Job):
96	93	def parent_complete(self, success, result):
97	94	self._complete_cb(self, self._element, success, self._result)
98	95
99		- def message(self, message_type, message, **kwargs):
100		- args = dict(kwargs)
101		- args['scheduler'] = True
102		- self._scheduler.context.message(
103		- Message(self._element._get_unique_id(),
104		- message_type,
105		- message,
106		- **args))
107		-
108	96	def child_process_data(self):
109	97	data = {}
110	98
...	...	@@ -119,3 +107,33 @@ class ElementJob(Job):
119	107	data['cache_size'] = cache_size
120	108
121	109	return data
	110	+
	111	+ def _start(self, text, **kwargs):
	112	+ self._scheduler.context.start(text, scheduler=True, **kwargs)
	113	+
	114	+ def _success(self, text, **kwargs):
	115	+ self._scheduler.context.success(text, scheduler=True, **kwargs)
	116	+
	117	+ def _failure(self, text, **kwargs):
	118	+ self._scheduler.context.failure(text, scheduler=True, **kwargs)
	119	+
	120	+ def _debug(self, text, **kwargs):
	121	+ self._scheduler.context.debug(text, scheduler=True, **kwargs)
	122	+
	123	+ def _status(self, text, **kwargs):
	124	+ self._scheduler.context.status(text, scheduler=True, **kwargs)
	125	+
	126	+ def _info(self, text, **kwargs):
	127	+ self._scheduler.context.info(text, scheduler=True, **kwargs)
	128	+
	129	+ def _warn(self, text, **kwargs):
	130	+ self._scheduler.context.warn(text, scheduler=True, **kwargs)
	131	+
	132	+ def _error(self, text, **kwargs):
	133	+ self._scheduler.context.error(text, scheduler=True, **kwargs)
	134	+
	135	+ def _bug(self, text, **kwargs):
	136	+ self._scheduler.context.bug(text, scheduler=True, **kwargs)
	137	+
	138	+ def _log(self, text, **kwargs):
	139	+ self._scheduler.context.log(text, scheduler=True, **kwargs)

buildstream/_scheduler/jobs/job.py

---

...	...	@@ -32,7 +32,7 @@ import psutil
32	32
33	33	# BuildStream toplevel imports
34	34	from ..._exceptions import ImplError, BstError, set_last_task_error
35		-from ..._message import Message, MessageType, unconditional_messages
	35	+from ..._message import MessageType, unconditional_messages
36	36	from ... import _signals, utils
37	37
38	38	# Return code values shutdown of job handling child processes
...	...	@@ -179,7 +179,7 @@ class Job():
179	179	# First resume the job if it's suspended
180	180	self.resume(silent=True)
181	181
182		- self._status("{} terminating".format(self.action_name), self.element)
	182	+ self._status("{} terminating".format(self.action_name))
183	183
184	184	# Make sure there is no garbage on the queue
185	185	self._parent_stop_listening()
...	...	@@ -211,7 +211,7 @@ class Job():
211	211
212	212	# Force kill
213	213	self._warn("{} did not terminate gracefully, killing"
214		- .format(self.action_name), self.element)
	214	+ .format(self.action_name))
215	215
216	216	try:
217	217	utils._kill_process_tree(self._process.pid)
...	...	@@ -226,7 +226,7 @@ class Job():
226	226	#
227	227	def suspend(self):
228	228	if not self._suspended:
229		- self._status("{} suspending".format(self.action_name), self.element)
	229	+ self._status("{} suspending".format(self.action_name))
230	230
231	231	try:
232	232	# Use SIGTSTP so that child processes may handle and propagate
...	...	@@ -250,7 +250,7 @@ class Job():
250	250	def resume(self, silent=False):
251	251	if self._suspended:
252	252	if not silent:
253		- self._status("{} resuming".format(self.action_name), self.element)
	253	+ self._status("{} resuming".format(self.action_name))
254	254
255	255	os.kill(self._process.pid, signal.SIGCONT)
256	256	self._suspended = False
...	...	@@ -303,26 +303,6 @@ class Job():
303	303	raise ImplError("Job '{kind}' does not implement child_process()"
304	304	.format(kind=type(self).__name__))
305	305
306		- # message():
307		- #
308		- # Logs a message, this will be logged in the task's logfile and
309		- # conditionally also be sent to the frontend.
310		- #
311		- # Args:
312		- # message (str): The message
313		- # message_type (MessageType): The type of message to send
314		- # plugin (Plugin): The plugin sending the message.
315		- # kwargs: Remaining Message() constructor arguments
316		- #
317		- def _message(self, text, plugin, *, msg_type=None, **kwargs):
318		- self._scheduler.context._message(
319		- text,
320		- plugin=plugin._get_unique_id(),
321		- msg_type=msg_type,
322		- scheduler=True,
323		- **kwargs
324		- )
325		-
326	306	# child_process_data()
327	307	#
328	308	# Abstract method to retrieve additional data that should be
...	...	@@ -348,35 +328,36 @@ class Job():
348	328	# Other methods can be called in both child or parent processes
349	329	#
350	330	#######################################################
351		- def _start(self, *args, **kwargs):
352		- self._message(*args, msg_type=MessageType.START, **kwargs)
353	331
354		- def _success(self, *args, **kwargs):
355		- self._message(*args, msg_type=MessageType.SUCCESS, **kwargs)
	332	+ def _start(self, text, **kwargs):
	333	+ self._scheduler.context.start(text, **kwargs)
	334	+
	335	+ def _success(self, text, **kwargs):
	336	+ self._scheduler.context.success(text, **kwargs)
356	337
357		- def _failure(self, *args, **kwargs):
358		- self._message(*args, msg_type=MessageType.FAIL, **kwargs)
	338	+ def _failure(self, text, **kwargs):
	339	+ self._scheduler.context.failure(text, **kwargs)
359	340
360		- def _debug(self, *args, **kwargs):
361		- self._message(*args, msg_type=MessageType.DEBUG, **kwargs)
	341	+ def _debug(self, text, **kwargs):
	342	+ self._scheduler.context.debug(text, **kwargs)
362	343
363		- def _status(self, *args, **kwargs):
364		- self._message(*args, msg_type=MessageType.STATUS, **kwargs)
	344	+ def _status(self, text, **kwargs):
	345	+ self._scheduler.context.status(text, **kwargs)
365	346
366		- def _info(self, *args, **kwargs):
367		- self._message(*args, msg_type=MessageType.INFO, **kwargs)
	347	+ def _info(self, text, **kwargs):
	348	+ self._scheduler.context.info(text, **kwargs)
368	349
369		- def _warn(self, *args, **kwargs):
370		- self._message(*args, msg_type=MessageType.WARN, **kwargs)
	350	+ def _warn(self, text, **kwargs):
	351	+ self._scheduler.context.warn(text, **kwargs)
371	352
372		- def _error(self, *args, **kwargs):
373		- self._message(*args, msg_type=MessageType.ERROR, **kwargs)
	353	+ def _error(self, text, **kwargs):
	354	+ self._scheduler.context.error(text, **kwargs)
374	355
375		- def _bug(self, *args, **kwargs):
376		- self._message(*args, msg_type=MessageType.BUG, **kwargs)
	356	+ def _bug(self, text, **kwargs):
	357	+ self._scheduler.context.bug(text, **kwargs)
377	358
378		- def _log(self, *args, **kwargs):
379		- self._message(*args, msg_type=MessageType.LOG, **kwargs)
	359	+ def _log(self, text, **kwargs):
	360	+ self._scheduler.context.log(text, **kwargs)
380	361
381	362	# _child_action()
382	363	#
...	...	@@ -423,7 +404,7 @@ class Job():
423	404	with _signals.suspendable(stop_time, resume_time), \
424	405	self._scheduler.context.recorded_messages(self._logfile) as filename:
425	406
426		- self._start(self.action_name, element, logfile=filename)
	407	+ self._start(self.action_name, logfile=filename)
427	408
428	409	try:
429	410	# Try the task action
...	...	@@ -433,10 +414,9 @@ class Job():
433	414	self._retry_flag = e.temporary
434	415
435	416	if self._retry_flag and (self._tries <= self._max_retries):
436		- self._failure("Try #{} failed, retrying".format(self._tries),
437		- element, elapsed=elapsed)
	417	+ self._failure("Try #{} failed, retrying".format(self._tries), elapsed=elapsed)
438	418	else:
439		- self._failure(str(e), element, elapsed=elapsed,
	419	+ self._failure(str(e), elapsed=elapsed,
440	420	detail=e.detail, logfile=filename,
441	421	sandbox=e.sandbox)
442	422
...	...	@@ -458,8 +438,7 @@ class Job():
458	438	elapsed = datetime.datetime.now() - starttime
459	439	detail = "An unhandled exception occured:\n\n{}".format(traceback.format_exc())
460	440
461		- self._bug(self.action_name, element, elapsed=elapsed,
462		- detail=detail, logfile=filename)
	441	+ self._bug(self.action_name, elapsed=elapsed, detail=detail, logfile=filename)
463	442	self._child_shutdown(RC_FAIL)
464	443	else:
465	444	# No exception occurred in the action
...	...	@@ -467,8 +446,7 @@ class Job():
467	446	self._child_send_result(result)
468	447
469	448	elapsed = datetime.datetime.now() - starttime
470		- self._success(self.action_name, element,
471		- elapsed=elapsed, logfile=filename)
	449	+ self._success(self.action_name, elapsed=elapsed, logfile=filename)
472	450
473	451	# Shutdown needs to stay outside of the above context manager,
474	452	# make sure we dont try to handle SIGTERM while the process
...	...	@@ -603,7 +581,7 @@ class Job():
603	581	if envelope._message_type == 'message':
604	582	# Propagate received messages from children
605	583	# back through the context.
606		- self._scheduler.context.message(envelope._message)
	584	+ self._scheduler.context._send_message(envelope._message)
607	585	elif envelope._message_type == 'error':
608	586	# For regression tests only, save the last error domain / reason
609	587	# reported from a child task in the main process, this global state

buildstream/_scheduler/queues/buildqueue.py

---

...	...	@@ -23,7 +23,6 @@ from datetime import timedelta
23	23	from . import Queue, QueueStatus
24	24	from ..jobs import ElementJob
25	25	from ..resources import ResourceType
26		-from ..._message import MessageType
27	26
28	27
29	28	# A queue which assembles elements
...	...	@@ -50,7 +49,7 @@ class BuildQueue(Queue):
50	49	self._tried.add(element)
51	50	_, description, detail = element._get_build_result()
52	51	logfile = element._get_build_log()
53		- self._message(element, MessageType.FAIL, description,
	52	+ self._failure(description, plugin=element,
54	53	detail=detail, action_name=self.action_name,
55	54	elapsed=timedelta(seconds=0),
56	55	logfile=logfile)

buildstream/_scheduler/queues/queue.py

---

...	...	@@ -30,7 +30,6 @@ from ..resources import ResourceType
30	30
31	31	# BuildStream toplevel imports
32	32	from ..._exceptions import BstError, set_last_task_error
33		-from ..._message import Message, MessageType
34	33
35	34
36	35	# Queue status for a given element
...	...	@@ -284,9 +283,8 @@ class Queue():
284	283	detail=str(e),
285	284	plugin=element._get_unique_id())
286	285	except Exception as e: # pylint: disable=broad-except
287		- self._message(element, MessageType.BUG,
288		- "Unhandled exception while saving workspaces",
289		- detail=traceback.format_exc())
	286	+ self._bug("Unhandled exception while saving workspaces",
	287	+ plugin=element, detail=traceback.format_exc())
290	288
291	289	# _job_done()
292	290	#
...	...	@@ -316,7 +314,9 @@ class Queue():
316	314
317	315	# Report error and mark as failed
318	316	#
319		- self._message(element, MessageType.ERROR, "Post processing error", detail=str(e))
	317	+ self._error("Post processing error",
	318	+ plugin=element,
	319	+ detail=str(e))
320	320	self.failed_elements.append(element)
321	321
322	322	# Treat this as a task error as it's related to a task
...	...	@@ -330,9 +330,9 @@ class Queue():
330	330
331	331	# Report unhandled exceptions and mark as failed
332	332	#
333		- self._message(element, MessageType.BUG,
334		- "Unhandled exception in post processing",
335		- detail=traceback.format_exc())
	333	+ self._bug("Unhandled exception in post processing",
	334	+ plugin=element,
	335	+ detail=traceback.format_exc())
336	336	self.failed_elements.append(element)
337	337	else:
338	338
...	...	@@ -348,12 +348,35 @@ class Queue():
348	348	else:
349	349	self.failed_elements.append(element)
350	350
351		- # Convenience wrapper for Queue implementations to send
352		- # a message for the element they are processing
353		- def _message(self, element, message_type, brief, **kwargs):
354		- context = element._get_context()
355		- context._message(brief, plugin=element._get_unique_id(),
356		- msg_type=message_type, **kwargs)
	351	+ def _start(self, text, **kwargs):
	352	+ self._scheduler.context.start(text, **kwargs)
	353	+
	354	+ def _success(self, text, **kwargs):
	355	+ self._scheduler.context.success(text, **kwargs)
	356	+
	357	+ def _failure(self, text, **kwargs):
	358	+ self._scheduler.context.failure(text, **kwargs)
	359	+
	360	+ def _debug(self, text, **kwargs):
	361	+ self._scheduler.context.debug(text, **kwargs)
	362	+
	363	+ def _status(self, text, **kwargs):
	364	+ self._scheduler.context.status(text, **kwargs)
	365	+
	366	+ def _info(self, text, **kwargs):
	367	+ self._scheduler.context.info(text, **kwargs)
	368	+
	369	+ def _warn(self, text, **kwargs):
	370	+ self._scheduler.context.warn(text, **kwargs)
	371	+
	372	+ def _error(self, text, **kwargs):
	373	+ self._scheduler.context.error(text, **kwargs)
	374	+
	375	+ def _bug(self, text, **kwargs):
	376	+ self._scheduler.context.bug(text, **kwargs)
	377	+
	378	+ def _log(self, text, **kwargs):
	379	+ self._scheduler.context.log(text, **kwargs)
357	380
358	381	def _element_log_path(self, element):
359	382	project = element._get_project()

buildstream/_stream.py

---

...	...	@@ -29,7 +29,6 @@ from contextlib import contextmanager
29	29	from tempfile import TemporaryDirectory
30	30
31	31	from ._exceptions import StreamError, ImplError, BstError, set_last_task_error
32		-from ._message import Message, MessageType
33	32	from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
34	33	from ._pipeline import Pipeline, PipelineSelection
35	34	from ._platform import Platform

buildstream/plugin.py

---

...	...	@@ -511,6 +511,15 @@ class Plugin():
511	511
512	512	self.__context.warn(brief, detail=detail)
513	513
	514	+ def skipped(self, brief, *, detail=None):
	515	+ """Prints a message indicating that an action has been skipped.
	516	+
	517	+ Args:
	518	+ brief (str): The brief message
	519	+ detail (str): An optional detailed message, can be multiline output
	520	+ """
	521	+ self.__context.skipped(brief, detail=detail)
	522	+
514	523	def log(self, brief, *, detail=None):
515	524	"""Log a message into the plugin's log file
516	525
...	...	@@ -717,10 +726,6 @@ class Plugin():
717	726
718	727	return (exit_code, output)
719	728
720		- def __message(self, message_type, brief, **kwargs):
721		- self.__context._message(brief, plugin=self.__unique_id,
722		- msg_type=message_type, **kwargs)
723		-
724	729	def __note_command(self, output, *popenargs, **kwargs):
725	730	workdir = os.getcwd()
726	731	if 'cwd' in kwargs:

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