Rework monitoring recording of invalid stdout/err streams (#3344)

This PR stops treating None as an invalid stdout/err: it is a legitimate
value for stdout/err. Prior to this PR, using None was resulting in
the stdout/err being recorded in monitoring as a stringified exception,
like this:

```
sqlite> select task_stderr from task;
type of argument "stdfspec" must be one of (os.PathLike, str, Tuple[str, str], Tuple[os.PathLike, str]); got NoneType instead
```

Before this PR, an invalid stdout/err stream was logged at WARNING level
and the filename as recorded in the monitoring database was set to the
exception message.

This PR elevates that log to an ERROR, because this is an error,
including the exception message to give a debugging hint.

In this error case, a blank stdout or stderr will be recorded in the
monitoring database. Generally the task will fail later due to other
parts of the code being unable to interpret the specification.

This PR adds a few checks around stdout/err tests that should not have
any error raised - this test checks for ERROR rather than WARNING because
Parsl will legitimately output WARNING messages in some configurations
(such as when checkpointing is enabled but Parsl notices it hasn't
checkpointed anything)

Co-authored-by: Kevin Hunter Kesling <[email protected]>

parsl/dataflow/dflow.py

@@ -240,16 +240,29 @@ def _create_task_log_info(self, task_record):
         task_log_info['task_stdin'] = task_record['kwargs'].get('stdin', None)
         stdout_spec = task_record['kwargs'].get('stdout', None)
         stderr_spec = task_record['kwargs'].get('stderr', None)
-        try:
-            stdout_name, _ = get_std_fname_mode('stdout', stdout_spec)
-        except Exception as e:
-            logger.warning("Incorrect stdout format {} for Task {}".format(stdout_spec, task_record['id']))
-            stdout_name = str(e)
-        try:
-            stderr_name, _ = get_std_fname_mode('stderr', stderr_spec)
-        except Exception as e:
-            logger.warning("Incorrect stderr format {} for Task {}".format(stderr_spec, task_record['id']))
-            stderr_name = str(e)
+
+        # stdout and stderr strings are set to the filename if we can
+        # interpret the specification; otherwise, set to the empty string
+        # (on exception, or when not specified)
+
+        if stdout_spec is not None:
+            try:
+                stdout_name, _ = get_std_fname_mode('stdout', stdout_spec)
+            except Exception:
+                logger.exception("Could not parse stdout specification {} for task {}".format(stdout_spec, task_record['id']))
+                stdout_name = ""
+        else:
+            stdout_name = ""
+
+        if stderr_spec is not None:
+            try:
+                stderr_name, _ = get_std_fname_mode('stderr', stderr_spec)
+            except Exception:
+                logger.exception("Could not parse stderr specification {} for task {}".format(stderr_spec, task_record['id']))
+                stderr_name = ""
+        else:
+            stderr_name = ""
+
         task_log_info['task_stdout'] = stdout_name
         task_log_info['task_stderr'] = stderr_name
         task_log_info['task_fail_history'] = ",".join(task_record['fail_history'])

parsl/tests/test_bash_apps/test_basic.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import random
 import re
@@ -37,7 +38,7 @@ def test_command_format_1(tmpd_cwd):
     assert so_content == "1 4 10"
 
 
-def test_auto_log_filename_format():
+def test_auto_log_filename_format(caplog):
     """Testing auto log filename format for BashApps
     """
     app_label = "label_test_auto_log_filename_format"
@@ -59,6 +60,9 @@ def test_auto_log_filename_format():
     assert contents == '1 {0} 10\n'.format(rand_int), \
         'Output does not match expected string "1 {0} 10", Got: "{1}"'.format(rand_int, contents)
 
+    for record in caplog.records:
+        assert record.levelno < logging.ERROR
+
 
 def test_parallel_for(tmpd_cwd, n=3):
     """Testing a simple parallel for loop"""

parsl/tests/test_bash_apps/test_stdout.py

@@ -1,3 +1,4 @@
+import logging
 import os
 
 import pytest
@@ -72,7 +73,7 @@ def test_bad_stderr_file():
 
 
 @pytest.mark.executor_supports_std_stream_tuples
-def test_stdout_truncate(tmpd_cwd):
+def test_stdout_truncate(tmpd_cwd, caplog):
     """Testing truncation of prior content of stdout"""
 
     out = (str(tmpd_cwd / 't1.out'), 'w')
@@ -87,8 +88,11 @@ def test_stdout_truncate(tmpd_cwd):
     assert len1 == 1
     assert len1 == len2
 
+    for record in caplog.records:
+        assert record.levelno < logging.ERROR
 
-def test_stdout_append(tmpd_cwd):
+
+def test_stdout_append(tmpd_cwd, caplog):
     """Testing appending to prior content of stdout (default open() mode)"""
 
     out = str(tmpd_cwd / 't1.out')
@@ -101,3 +105,6 @@ def test_stdout_append(tmpd_cwd):
     len2 = len(open(out).readlines())
 
     assert len1 == 1 and len2 == 2
+
+    for record in caplog.records:
+        assert record.levelno < logging.ERROR