Merge pull request #360 from alexhsamuel/fix/run-metadata

clean up run metadata

Author: alexhsamuel

.github/workflows/makefile.yml

@@ -17,6 +17,10 @@ jobs:
     - name: Install Litestream
       run: wget --quiet https://github.com/benbjohnson/litestream/releases/download/v0.3.13/litestream-v0.3.13-linux-amd64.deb && sudo dpkg -i litestream-v0.3.13-linux-amd64.deb
 
+    - uses: actions/setup-python@v5
+      with:
+        python-version: '3.10'
+
     - name: Upgrade pip
       run: pip3 install --upgrade pip 'setuptools>=61'
 

python/apsis/actions/test.py

@@ -111,3 +111,36 @@ def from_jso(cls, jso):
 
 
 
+class CheckLabelAction(BaseAction):
+    """
+    Action that raises an exception if a run doesn't have a certain label.
+    """
+
+    def __init__(self, label, *, condition=None):
+        super().__init__(condition=condition)
+        self.__label = str(label)
+
+
+    @classmethod
+    def from_jso(cls, jso):
+        with check_schema(jso) as pop:
+            label       = pop("label", str, None)
+            condition   = pop("if", Condition.from_jso, None)
+        return cls(label, condition=condition)
+
+
+    def to_jso(self):
+        jso = super().to_jso()
+        jso["label"] = self.__label
+        return jso
+
+
+    def __call__(self, apsis, run):
+        if self.condition is not None and not self.condition(run):
+            return
+        labels = run.meta.get("job", {}).get("labels", [])
+        if self.__label not in labels:
+            raise RuntimeError(f"run missing label: {self.__label}")
+
+
+

python/apsis/apsis.py

@@ -276,8 +276,9 @@ def __prepare_run(self, run):
                 return False
 
         # Attach job labels to the run.
-        if run.meta.get("labels") is None:
-            run.meta["labels"] = job.meta.get("labels", [])
+        run.meta["job"] = {
+            "labels": job.meta.get("labels", []),
+        }
 
         return True
 
@@ -365,19 +366,20 @@ def _run_exc(self, run, *, message=None):
 
 
     # FIXME: persist is a hack.
-    def _update_metadata(self, run, meta, *, persist=True):
+    def _update_metadata(self, run, meta):
         """
         Updates run metadata, without transitioning.
         """
         assert run.state in {State.starting, State.running}
 
-        run._update(meta=meta)
-        # Persist the new state.
-        if persist:
-            self.run_store.update(run, now())
+        if meta is None or len(meta) == 0:
+            return
 
+        run.meta.update(meta)
+        # Persist the new state.
+        self.run_store.update(run, now())
         # Publish to run update subscribers.
-        self.run_update_publisher.publish(run.run_id, {"meta": meta})
+        self.run_update_publisher.publish(run.run_id, {"meta": run.meta})
 
 
     def _update_output_data(self, run, outputs, persist):
@@ -410,6 +412,10 @@ def _update_output_data(self, run, outputs, persist):
     def _transition(self, run, state, *, meta={}, **kw_args):
         """
         Transitions `run` to `state`, updating it with `kw_args`.
+
+        :param meta:
+          Metadata updates.  Sets or replaces run metadata keys from this
+          mapping.
         """
         time = now()
         run_id = run.run_id
@@ -419,10 +425,6 @@ def _transition(self, run, state, *, meta={}, **kw_args):
             self.__db.run_log_db.flush(run.run_id)
             run.expected = False
 
-        # Update metadata.  Don't persist; will persist with the run.
-        # FIXME: Separate out metadata from Run and clean this up.
-        if meta:
-            self._update_metadata(run, meta, persist=False)
         # Transition the run object.
         run._transition(time, state, meta=meta, **kw_args)
         # Persist the new state.
@@ -431,6 +433,8 @@ def _transition(self, run, state, *, meta={}, **kw_args):
         # Publish to run update subscribers.
         if run_id in self.run_update_publisher:
             msg = {"run": run_to_summary_jso(run)}
+            if len(meta) > 0:
+                msg["meta"] = run.meta
             self.run_update_publisher.publish(run_id, msg)
         # Publish to summary subscribers.
         self.summary_publisher.publish(messages.make_run_transition(run))
@@ -797,7 +801,7 @@ async def _process_updates(apsis, run, updates):
                     apsis._transition(
                         run, State.running,
                         run_state   =running.run_state,
-                        meta        =running.meta,
+                        meta        ={"program": running.meta},
                         times       =running.times,
                     )
 
@@ -806,7 +810,7 @@ async def _process_updates(apsis, run, updates):
                     apsis._update_output_data(run, error.outputs, persist=True)
                     apsis._transition(
                         run, State.error,
-                        meta        =error.meta,
+                        meta        ={"program": error.meta},
                         times       =error.times,
                     )
                     return
@@ -823,7 +827,7 @@ async def _process_updates(apsis, run, updates):
                     if update.outputs is not None:
                         apsis._update_output_data(run, update.outputs, False)
                     if update.meta is not None:
-                        apsis._update_metadata(run, update.meta)
+                        apsis._update_metadata(run, {"program": update.meta})
 
                 case ProgramSuccess() as success:
                     apsis.run_log.record(run, "success")
@@ -834,7 +838,7 @@ async def _process_updates(apsis, run, updates):
                     )
                     apsis._transition(
                         run, State.success,
-                        meta        =success.meta,
+                        meta        ={"program": success.meta},
                         times       =success.times,
                     )
 
@@ -848,7 +852,7 @@ async def _process_updates(apsis, run, updates):
                     )
                     apsis._transition(
                         run, State.failure,
-                        meta        =failure.meta,
+                        meta        ={"program": failure.meta},
                         times       =failure.times,
                     )
 
@@ -861,7 +865,7 @@ async def _process_updates(apsis, run, updates):
                     )
                     apsis._transition(
                         run, State.error,
-                        meta        =error.meta,
+                        meta        ={"program": error.meta},
                         times       =error.times,
                     )
 

python/apsis/lib/api.py

@@ -117,7 +117,7 @@ def run_to_summary_jso(run):
         "run_id"        : run.run_id,
         "state"         : run.state.name,
         "times"         : { n: time_to_jso(t) for n, t in run.times.items() },
-        "labels"        : run.meta.get("labels", []),
+        "labels"        : run.meta.get("job", {}).get("labels", []),
     }
     if run.expected:
         jso["expected"] = run.expected

python/apsis/runs.py

@@ -201,16 +201,14 @@ def __str__(self):
         return f"{self.run_id} {self.state.name} {self.inst}"
 
 
-    def _update(self, *, meta=None):
-        if meta is not None:
-            self.meta = meta
-
-
     def _transition(self, timestamp, state, *, meta={}, times={},
                     message=None, run_state=None, force=False):
         """
         :param force:
           Transition outside of the state model.
+        :param meta:
+          Metadata updates.  Sets or replaces run metadata keys from this
+          mapping.
         """
         # Check that this is a valid transition.
         if not force and self.state not in TRANSITIONS[state]:

python/apsis/service/vue/index.html

@@ -1 +1 @@
-<!DOCTYPE html><html><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1"><link rel=stylesheet href=/static/css/roboto.css><link rel=stylesheet href=/static/css/roboto-condensed.css><link rel=stylesheet href=/static/css/roboto-mono.css><title>apsis</title><link href=/static/css/app.2c00a3a669051c996f8e52bf4cc3abed.css rel=stylesheet></head><body><div id=app></div><script type=text/javascript src=/static/js/manifest.2ae2e69a05c33dfc65f8.js></script><script type=text/javascript src=/static/js/vendor.c925f914b965eeaf0032.js></script><script type=text/javascript src=/static/js/app.1759c586f7305af1844d.js></script></body></html>
+<!DOCTYPE html><html><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1"><link rel=stylesheet href=/static/css/roboto.css><link rel=stylesheet href=/static/css/roboto-condensed.css><link rel=stylesheet href=/static/css/roboto-mono.css><title>apsis</title><link href=/static/css/app.74e09c3bce8167f746a4f0af24143daf.css rel=stylesheet></head><body><div id=app></div><script type=text/javascript src=/static/js/manifest.2ae2e69a05c33dfc65f8.js></script><script type=text/javascript src=/static/js/vendor.c925f914b965eeaf0032.js></script><script type=text/javascript src=/static/js/app.7a05013375372941a20d.js></script></body></html>