Use run_dir from Config for monitoring path (#3100)

Prior to this PR, the default was hard-coded to "runinfo/" (in PR #2136)
which works most of the time, as most of the time, users do not change
the Config run_dir.

However, PR #3003 sets Config.run_dir for most tests, resulting in a
(mostly silent) monitoring database manager failure. This PR fixes that
by allowing monitoring to see that new path, as well as fixing it for
all other run_dir changing users.

Note that in Config/DFK there are two uses of the "run_dir" : in
Config, that name refers to the root in which .../000/... /NNN/
directories will be created. In the DFK, that name refers to the
numbered config.DFK/NNN/ directory.

Since #3003 (which I bisected back to), monitoring has been mostly
silently failing with a database opening error appearing in
database_manager.log and a missing monitoring.db, but no exception
raised in the workflow process. This PR does not address that
loss-of-failure-info problem.

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

Author: benclifford

parsl/dataflow/dflow.py

@@ -113,7 +113,7 @@ def __init__(self, config: Config) -> None:
             if self.monitoring.logdir is None:
                 self.monitoring.logdir = self.run_dir
             self.hub_address = self.monitoring.hub_address
-            self.hub_interchange_port = self.monitoring.start(self.run_id, self.run_dir)
+            self.hub_interchange_port = self.monitoring.start(self.run_id, self.run_dir, self.config.run_dir)
 
         self.time_began = datetime.datetime.now()
         self.time_completed: Optional[datetime.datetime] = None

parsl/monitoring/monitoring.py

@@ -84,7 +84,7 @@ def __init__(self,
 
                  workflow_name: Optional[str] = None,
                  workflow_version: Optional[str] = None,
-                 logging_endpoint: str = 'sqlite:///runinfo/monitoring.db',
+                 logging_endpoint: Optional[str] = None,
                  logdir: Optional[str] = None,
                  monitoring_debug: bool = False,
                  resource_monitoring_enabled: bool = True,
@@ -118,7 +118,7 @@ def __init__(self,
         logging_endpoint : str
              The database connection url for monitoring to log the information.
              These URLs follow RFC-1738, and can include username, password, hostname, database name.
-             Default: 'sqlite:///monitoring.db'
+             Default: sqlite, in the configured run_dir.
         logdir : str
              Parsl log directory paths. Logs and temp files go here. Default: '.'
         monitoring_debug : Bool
@@ -162,11 +162,14 @@ def __init__(self,
         self.resource_monitoring_enabled = resource_monitoring_enabled
         self.resource_monitoring_interval = resource_monitoring_interval
 
-    def start(self, run_id: str, run_dir: str) -> int:
+    def start(self, run_id: str, dfk_run_dir: str, config_run_dir: Union[str, os.PathLike]) -> int:
 
         if self.logdir is None:
             self.logdir = "."
 
+        if self.logging_endpoint is None:
+            self.logging_endpoint = f"sqlite:///{os.fspath(config_run_dir)}/monitoring.db"
+
         os.makedirs(self.logdir, exist_ok=True)
 
         # Initialize the ZMQ pipe to the Parsl Client
@@ -231,7 +234,7 @@ def start(self, run_id: str, run_dir: str) -> int:
         self.logger.info("Started the router process {} and DBM process {}".format(self.router_proc.pid, self.dbm_proc.pid))
 
         self.filesystem_proc = Process(target=filesystem_receiver,
-                                       args=(self.logdir, self.resource_msgs, run_dir),
+                                       args=(self.logdir, self.resource_msgs, dfk_run_dir),
                                        name="Monitoring-Filesystem-Process",
                                        daemon=True
                                        )