log xtrigger errors, workflow_state validation.

changes.d/5809.break.d

@@ -0,0 +1,3 @@
+The workflow-state command and xtrigger now looks for task outputs (a.k.a triggers)
+not the corresponding task messages, as well as task status. This is a consequence
+of the new optional output support in Cylc 8.

changes.d/5809.feat.d

@@ -0,0 +1 @@
+The workflow-state command and xtrigger is now flow-specific.

changes.d/5809.fix.d

@@ -0,0 +1 @@
+Workflow-state command and xtrigger: handle status and outputs the same way.

cylc/flow/scripts/workflow_state.py

@@ -199,7 +199,7 @@ def get_option_parser() -> COP:
     parser.add_option(
         "-O", "--output", "--message", metavar="OUTPUT",
         help="Check for a given task output"
-        " (--message is deprecated and aliased to task output)",
+        " (--message is now aliased to task output)",
         action="store", dest="output", default=None)
 
     parser.add_option(

cylc/flow/xtrigger_mgr.py

@@ -360,6 +360,7 @@ def check_xtrigger(
             )
         except TypeError as exc:
             raise XtriggerConfigError(label, f"{sig_str}: {exc}")
+
         # Specific xtrigger.validate(), if available.
         XtriggerManager.try_xtrig_validate_func(
             label, fname, fdir, bound_args, sig_str
@@ -406,6 +407,7 @@ def try_xtrig_validate_func(
         """Call an xtrigger's `validate()` function if it is implemented.
 
         Raise XtriggerConfigError if validation fails.
+
         """
         try:
             xtrig_validate_func = get_xtrig_func(fname, 'validate', fdir)
@@ -416,7 +418,7 @@ def try_xtrig_validate_func(
             xtrig_validate_func(bound_args.arguments)
         except Exception as exc:  # Note: catch all errors
             raise XtriggerConfigError(
-                label, f"{signature_str} validation failed: {exc}"
+                label, f"{signature_str}\n{exc}"
             )
 
     def add_trig(self, label: str, fctx: 'SubFuncContext', fdir: str) -> None:
@@ -623,22 +625,35 @@ def callback(self, ctx: 'SubFuncContext'):
 
         Record satisfaction status and function results dict.
 
+        Log a warning if the xtrigger functions errors, to distinguish
+        errors from not-satisfied.
+
         Args:
             ctx (SubFuncContext): function context
         Raises:
             ValueError: if the context given is not active
         """
         sig = ctx.get_signature()
         self.active.remove(sig)
+
+        if ctx.ret_code != 0:
+            msg = f"ERROR in xtrigger {sig}"
+            if ctx.err:
+                msg += f"\n{ctx.err}"
+            LOG.warning(msg)
+
         try:
             satisfied, results = json.loads(ctx.out)
         except (ValueError, TypeError):
             return
+
         LOG.debug('%s: returned %s', sig, results)
-        if satisfied:
-            # Newly satisfied
-            self.data_store_mgr.delta_task_xtrigger(sig, True)
-            self.workflow_db_mgr.put_xtriggers({sig: results})
-            LOG.info('xtrigger satisfied: %s = %s', ctx.label, sig)
-            self.sat_xtrig[sig] = results
-            self.do_housekeeping = True
+        if not satisfied:
+            return
+
+        # Newly satisfied
+        self.data_store_mgr.delta_task_xtrigger(sig, True)
+        self.workflow_db_mgr.put_xtriggers({sig: results})
+        LOG.info('xtrigger satisfied: %s = %s', ctx.label, sig)
+        self.sat_xtrig[sig] = results
+        self.do_housekeeping = True

cylc/flow/xtriggers/suite_state.py

@@ -78,6 +78,6 @@ def suite_state(suite, task, point, offset=None, status='succeeded',
         point=point,
         offset=offset,
         status=status,
-        message=message,
+        output=message,
         cylc_run_dir=cylc_run_dir
     )

cylc/flow/xtriggers/workflow_state.py

@@ -14,15 +14,16 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import sqlite3
-from typing import Dict, Optional, Tuple
+from typing import Dict, Optional, Tuple, Any
 
 from metomi.isodatetime.parsers import TimePointParser
 
 from cylc.flow.cycling.util import add_offset
 from cylc.flow.dbstatecheck import CylcWorkflowDBChecker
 from cylc.flow.pathutil import get_cylc_run_dir
 from cylc.flow.workflow_files import infer_latest_run_from_id
+from cylc.flow.exceptions import WorkflowConfigError
+from cylc.flow.task_state import TASK_STATUSES_ALL
 
 
 def workflow_state(
@@ -31,9 +32,10 @@ def workflow_state(
     point: str,
     offset: Optional[str] = None,
     status: Optional[str] = None,
-    message: Optional[str] = None,
+    output: Optional[str] = None,
+    flow_num: Optional[int] = None,
     cylc_run_dir: Optional[str] = None
-) -> Tuple[bool, Optional[Dict[str, Optional[str]]]]:
+) -> Tuple[bool, Dict[str, Optional[str]]]:
     """Connect to a workflow DB and query the requested task state.
 
     * Reports satisfied only if the remote workflow state has been achieved.
@@ -82,32 +84,69 @@ def workflow_state(
     if offset is not None:
         point = str(add_offset(point, offset))
 
-    try:
-        checker = CylcWorkflowDBChecker(cylc_run_dir, workflow)
-    except (OSError, sqlite3.Error):
-        # Failed to connect to DB; target workflow may not be started.
-        return (False, None)
-
-    point = str(
-        TimePointParser().parse(
-            point, dump_format=checker.point_fmt
+    # Failure to connect to DB will raise exceptions here.
+    # It could mean the target workflow as not started yet,
+    # but it could also mean a typo in the workflow ID, so
+    # so don't hide the error.
+    checker = CylcWorkflowDBChecker(cylc_run_dir, workflow)
+
+    # Point validity can only be checked at run time.
+    # Bad function arg templating can cause a syntax error.
+    if checker.point_fmt is None:
+        # Integer cycling: raises ValueError if bad.
+        int(point)
+    else:
+        # Datetime cycling: raises ISO8601SyntaxError if bad
+        point = str(
+            TimePointParser().parse(
+                point, dump_format=checker.point_fmt
+            )
         )
-    )
 
-    if not message and not status:
+    if not output and not status:
         status = "succeeded"
 
-    satisfied = checker.task_state_met(
-        task, str(point), output=message, status=status
+    satisfied: bool = checker.task_state_met(
+        task, point, output=output, status=status
     )
 
     results = {
         'workflow': workflow,
         'task': task,
-        'point': point,
+        'point': str(point),
         'offset': offset,
         'status': status,
-        'message': message,
+        'output': output,
+        'flow_num': str(flow_num),
         'cylc_run_dir': cylc_run_dir
     }
     return satisfied, results
+
+
+def validate(args: Dict[str, Any]):
+    """Validate workflow_state function args from the workflow config.
+
+    The rules for are:
+    * output/status: one at most (defaults to succeeded status)
+    * flow_num: Must be an integer
+    * status: Must be a valid status
+
+    """
+    output = args['output']
+    status = args['status']
+    flow_num = args['flow_num']
+
+    if output is not None and status is not None:
+        raise WorkflowConfigError(
+            "Give `status` or `output`, not both"
+        )
+
+    if status is not None and status not in TASK_STATUSES_ALL:
+        raise WorkflowConfigError(
+            f"Invalid tasks status '{status}'"
+        )
+
+    if flow_num is not None and not isinstance(flow_num, int):
+        raise WorkflowConfigError(
+            "flow_num must be an integer"
+        )

tests/functional/xtriggers/03-sequence.t

@@ -60,4 +60,3 @@ __END__
 
 cylc stop --now --max-polls=10 --interval=2 "${WORKFLOW_NAME}"
 purge
-exit

tests/functional/xtriggers/04-suite_state.t

@@ -29,7 +29,7 @@ init_workflow "$TEST_NAME_BASE" << __FLOW_CONFIG__
         [[[R1]]]
             graph = @upstream => foo
     [[xtriggers]]
-        upstream = suite_state(suite=thorin/oin/gloin, task=mithril, point=1)
+        upstream = suite_state(suite=thorin/oin/gloin, task=mithril, point=1, status="succeeded")
 [runtime]
     [[foo]]
 __FLOW_CONFIG__

tests/integration/test_config.py

@@ -357,7 +357,7 @@ def test_xtrig_validation_wall_clock(
         }
     })
     with pytest.raises(WorkflowConfigError, match=(
-        r'\[@myxt\] wall_clock\(offset=PT7MH\) validation failed: '
+        r'\[@myxt\] wall_clock\(offset=PT7MH\)\n'
         r'Invalid offset: PT7MH'
     )):
         validate(id_)
@@ -392,7 +392,7 @@ def test_xtrig_validation_echo(
     })
     with pytest.raises(
         WorkflowConfigError,
-        match=r'echo.* Requires \'succeed=True/False\' arg'
+        match=r'Requires \'succeed=True/False\' arg'
     ):
         validate(id_)