Merge branch 'aevri/set_message_unique_id' into 'master'

jobs: refactor, use new set_message_unique_id

See merge request BuildStream/buildstream!1337

3 files changed, 43 insertions, 26 deletions

diff --git a/src/buildstream/_scheduler/jobs/elementjob.py b/src/buildstream/_scheduler/jobs/elementjob.py
index fb5d38e11..d6aa81567 100644
--- a/src/buildstream/_scheduler/jobs/elementjob.py
+++ b/src/buildstream/_scheduler/jobs/elementjob.py
@@ -18,7 +18,7 @@
 #
 from ruamel import yaml
 
-from ..._message import Message, MessageType
+from ..._message import MessageType
 
 from .job import Job
 
@@ -72,7 +72,8 @@ class ElementJob(Job):
         self._action_cb = action_cb            # The action callable function
         self._complete_cb = complete_cb        # The complete callable function
 
-        # Set the task wide ID for logging purposes
+        # Set the ID for logging purposes
+        self.set_message_unique_id(element._unique_id)
         self.set_task_id(element._unique_id)
 
     @property
@@ -96,15 +97,6 @@ class ElementJob(Job):
     def parent_complete(self, status, result):
         self._complete_cb(self, self._element, status, self._result)
 
-    def message(self, message_type, message, **kwargs):
-        args = dict(kwargs)
-        args['scheduler'] = True
-        self._scheduler.context.message(
-            Message(self._element._unique_id,
-                    message_type,
-                    message,
-                    **args))
-
     def child_process_data(self):
         data = {}
 
diff --git a/src/buildstream/_scheduler/jobs/job.py b/src/buildstream/_scheduler/jobs/job.py
index 1c19d0221..203564af7 100644
--- a/src/buildstream/_scheduler/jobs/job.py
+++ b/src/buildstream/_scheduler/jobs/job.py
@@ -112,6 +112,7 @@ class Job():
         self._terminated = False               # Whether this job has been explicitly terminated
 
         self._logfile = logfile
+        self._message_unique_id = None
         self._task_id = None
 
     # spawn()
@@ -254,12 +255,25 @@ class Job():
             os.kill(self._process.pid, signal.SIGCONT)
             self._suspended = False
 
+    # set_message_unique_id()
+    #
+    # This is called by Job subclasses to set the plugin ID
+    # issuing the message (if an element is related to the Job).
+    #
+    # Args:
+    #     unique_id (int): The id to be supplied to the Message() constructor
+    #
+    def set_message_unique_id(self, unique_id):
+        self._message_unique_id = unique_id
+
     # set_task_id()
     #
     # This is called by Job subclasses to set a plugin ID
     # associated with the task at large (if any element is related
     # to the task).
     #
+    # This will only be used in the child process running the task.
+    #
     # The task ID helps keep messages in the frontend coherent
     # in the case that multiple plugins log in the context of
     # a single task (e.g. running integration commands should appear
@@ -272,6 +286,26 @@ class Job():
     def set_task_id(self, task_id):
         self._task_id = task_id
 
+    # message():
+    #
+    # Logs a message, this will be logged in the task's logfile and
+    # conditionally also be sent to the frontend.
+    #
+    # Args:
+    #    message_type (MessageType): The type of message to send
+    #    message (str): The message
+    #    kwargs: Remaining Message() constructor arguments, note that you can
+    #            override 'unique_id' this way.
+    #
+    def message(self, message_type, message, **kwargs):
+        kwargs['scheduler'] = True
+        unique_id = self._message_unique_id
+        if "unique_id" in kwargs:
+            unique_id = kwargs["unique_id"]
+            del kwargs["unique_id"]
+        self._scheduler.context.message(
+            Message(unique_id, message_type, message, **kwargs))
+
     # send_message()
     #
     # To be called from inside Job.child_process() implementations
@@ -329,21 +363,6 @@ class Job():
         raise ImplError("Job '{kind}' does not implement child_process()"
                         .format(kind=type(self).__name__))
 
-    # message():
-    #
-    # Logs a message, this will be logged in the task's logfile and
-    # conditionally also be sent to the frontend.
-    #
-    # Args:
-    #    message_type (MessageType): The type of message to send
-    #    message (str): The message
-    #    kwargs: Remaining Message() constructor arguments
-    #
-    def message(self, message_type, message, **kwargs):
-        args = dict(kwargs)
-        args['scheduler'] = True
-        self._scheduler.context.message(Message(None, message_type, message, **args))
-
     # child_process_data()
     #
     # Abstract method to retrieve additional data that should be
diff --git a/tests/frontend/logging.py b/tests/frontend/logging.py
index 31a2dd909..a8f894976 100644
--- a/tests/frontend/logging.py
+++ b/tests/frontend/logging.py
@@ -125,3 +125,9 @@ def test_failed_build_listing(cli, datafiles):
         assert m.start() in failure_summary_range
         assert m.end() in failure_summary_range
     assert len(matches) == 3  # each element should be matched once.
+
+    # Note that if we mess up the 'unique_id' of Messages, they won't be printed
+    # with the name of the relevant element, e.g. 'testfail-1.bst'. Check that
+    # they have the name as expected.
+    pattern = r"\[..:..:..\] FAILURE testfail-.\.bst: Staged artifacts do not provide command 'sh'"
+    assert len(re.findall(pattern, result.stderr, re.MULTILINE)) == 6  # each element should be matched twice.