Feature / Runtime config plugin (#451)

* Provide a capability to load 3rd party plugins when launching from the CLI

* Provide a plugin interface for config loading, and local implementation

* Special load method for config plugins in plugin manager

* Separate config loading and config parsing using a new config manager

* Fix TRAC deprecation warnings

* Update unit tests for config loading / parsing

* Handle URL schemes for Windows file paths

* Fix path resolution in the config manager on Windows

* Allow specifying plugin_package as a launch arg for launch_model and launch_job

* Test case for external config loader

* Fixes for using external config loaders

* Fix path resolution in the config manager on Windows

* Fix path resolution in the config manager on Windows

Author: Martin Traverse

tracdap-runtime/python/src/tracdap/rt/_exec/dev_mode.py

@@ -46,7 +46,7 @@ class DevModeTranslator:
     _log: tp.Optional[_util.logging.Logger] = None
 
     @classmethod
-    def translate_sys_config(cls, sys_config: _cfg.RuntimeConfig, config_dir: tp.Optional[pathlib.Path]):
+    def translate_sys_config(cls, sys_config: _cfg.RuntimeConfig, config_mgr: _cfg_p.ConfigManager):
 
         cls._log.info(f"Applying dev mode config translation to system config")
 
@@ -56,7 +56,7 @@ def translate_sys_config(cls, sys_config: _cfg.RuntimeConfig, config_dir: tp.Opt
             sys_config.storage = _cfg.StorageConfig()
 
         sys_config = cls._add_integrated_repo(sys_config)
-        sys_config = cls._resolve_relative_storage_root(sys_config, config_dir)
+        sys_config = cls._resolve_relative_storage_root(sys_config, config_mgr)
 
         return sys_config
 
@@ -66,7 +66,7 @@ def translate_job_config(
             sys_config: _cfg.RuntimeConfig,
             job_config: _cfg.JobConfig,
             scratch_dir: pathlib.Path,
-            config_dir: tp.Optional[pathlib.Path],
+            config_mgr: _cfg_p.ConfigManager,
             model_class: tp.Optional[_api.TracModel.__class__]) \
             -> _cfg.JobConfig:
 
@@ -84,7 +84,7 @@ def translate_job_config(
 
         # Fow flows, load external flow definitions then perform auto-wiring and type inference
         if job_config.job.jobType == _meta.JobType.RUN_FLOW:
-            job_config = cls._process_flow_definition(job_config, config_dir)
+            job_config = cls._process_flow_definition(job_config, config_mgr)
 
         # For run (model|flow) jobs, apply processing to the parameters, inputs and outputs
         if job_config.job.jobType in [_meta.JobType.RUN_MODEL, _meta.JobType.RUN_FLOW]:
@@ -109,7 +109,7 @@ def _add_integrated_repo(cls, sys_config: _cfg.RuntimeConfig) -> _cfg.RuntimeCon
     @classmethod
     def _resolve_relative_storage_root(
             cls, sys_config: _cfg.RuntimeConfig,
-            sys_config_path: tp.Optional[pathlib.Path]):
+            config_mgr: _cfg_p.ConfigManager):
 
         storage_config = copy.deepcopy(sys_config.storage)
 
@@ -128,6 +128,7 @@ def _resolve_relative_storage_root(
 
             cls._log.info(f"Resolving relative path for [{bucket_key}] local storage...")
 
+            sys_config_path = config_mgr.config_dir_path()
             if sys_config_path is not None:
                 absolute_path = sys_config_path.joinpath(root_path).resolve()
                 if absolute_path.exists():
@@ -291,7 +292,7 @@ def _generate_model_for_entry_point(
         return model_id, model_object
 
     @classmethod
-    def _process_flow_definition(cls, job_config: _cfg.JobConfig, config_dir: pathlib.Path) -> _cfg.JobConfig:
+    def _process_flow_definition(cls, job_config: _cfg.JobConfig, config_mgr: _cfg_p.ConfigManager) -> _cfg.JobConfig:
 
         flow_details = job_config.job.runFlow.flow
 
@@ -305,21 +306,12 @@ def _process_flow_definition(cls, job_config: _cfg.JobConfig, config_dir: pathli
             cls._log.error(err)
             raise _ex.EConfigParse(err)
 
-        flow_path = config_dir.joinpath(flow_details) if config_dir is not None else pathlib.Path(flow_details)
-
-        if not flow_path.exists():
-            err = f"Flow definition not available for [{flow_details}]: File not found ({flow_path})"
-            cls._log.error(err)
-            raise _ex.EConfigParse(err)
-
         flow_id = _util.new_object_id(_meta.ObjectType.FLOW)
         flow_key = _util.object_key(flow_id)
 
-        cls._log.info(f"Generating flow definition for [{flow_details}] with ID = [{flow_key}]")
+        cls._log.info(f"Generating flow definition from [{flow_details}] with ID = [{flow_key}]")
 
-        flow_parser = _cfg_p.ConfigParser(_meta.FlowDefinition)
-        flow_raw_data = flow_parser.load_raw_config(flow_path, flow_path.name)
-        flow_def = flow_parser.parse(flow_raw_data, flow_path.name)
+        flow_def = config_mgr.load_config_object(flow_details, _meta.FlowDefinition)
 
         # Auto-wiring and inference only applied to externally loaded flows for now
         flow_def = cls._autowire_flow(flow_def, job_config)

tracdap-runtime/python/src/tracdap/rt/_exec/runtime.py

@@ -64,6 +64,7 @@ def __init__(
             job_result_format: tp.Optional[str] = None,
             scratch_dir: tp.Union[str, pathlib.Path, None] = None,
             scratch_dir_persist: bool = False,
+            plugin_packages: tp.List[str] = None,
             dev_mode: bool = False):
 
         trac_version = _version.__version__
@@ -86,12 +87,13 @@ def __init__(
         self._log.info(f"TRAC D.A.P. Python Runtime {trac_version}")
 
         self._sys_config = sys_config if isinstance(sys_config, _cfg.RuntimeConfig) else None
-        self._sys_config_path = pathlib.Path(sys_config) if not self._sys_config else None
+        self._sys_config_path = sys_config if not self._sys_config else None
         self._job_result_dir = job_result_dir
         self._job_result_format = job_result_format
         self._scratch_dir = scratch_dir
         self._scratch_dir_provided = True if scratch_dir is not None else False
         self._scratch_dir_persist = scratch_dir_persist
+        self._plugin_packages = plugin_packages or []
         self._dev_mode = dev_mode
 
         # Runtime control
@@ -102,6 +104,7 @@ def __init__(
         self._oneshot_job = None
 
         # Top level resources
+        self._config_mgr: tp.Optional[_cparse.ConfigManager] = None
         self._models: tp.Optional[_models.ModelLoader] = None
         self._storage: tp.Optional[_storage.StorageManager] = None
 
@@ -141,30 +144,36 @@ def pre_start(self):
 
             self._prepare_scratch_dir()
 
-            # Plugin manager and static API impl are singletons
-            # If these methods are called multiple times, the second and subsequent calls are ignored
+            # Plugin manager, static API and guard rails are singletons
+            # Calling these methods multiple times is safe (e.g. for embedded or testing scenarios)
+            # However, plugins are never un-registered for the lifetime of the processes
 
             _plugins.PluginManager.register_core_plugins()
+
+            for plugin_package in self._plugin_packages:
+                _plugins.PluginManager.register_plugin_package(plugin_package)
+
             _static_api.StaticApiImpl.register_impl()
             _guard.PythonGuardRails.protect_dangerous_functions()
 
             # Load sys config (or use embedded), config errors are detected before start()
             # Job config can also be checked before start() by using load_job_config()
 
+            self._config_mgr = _cparse.ConfigManager.for_root_config(self._sys_config_path)
+
             if self._sys_config is None:
                 sys_config_dev_mode = _dev_mode.DEV_MODE_SYS_CONFIG if self._dev_mode else None
-                sys_config_parser = _cparse.ConfigParser(_cfg.RuntimeConfig, sys_config_dev_mode)
-                sys_config_raw = sys_config_parser.load_raw_config(self._sys_config_path, config_file_name="system")
-                self._sys_config = sys_config_parser.parse(sys_config_raw, self._sys_config_path)
+                self._sys_config = self._config_mgr.load_root_object(
+                    _cfg.RuntimeConfig, sys_config_dev_mode,
+                    config_file_name="system")
             else:
                 self._log.info("Using embedded system config")
 
             # Dev mode translation is controlled by the dev mode flag
             # I.e. it can be applied to embedded configs
 
             if self._dev_mode:
-                config_dir = self._sys_config_path.parent if self._sys_config_path is not None else None
-                self._sys_config = _dev_mode.DevModeTranslator.translate_sys_config(self._sys_config, config_dir)
+                self._sys_config = _dev_mode.DevModeTranslator.translate_sys_config(self._sys_config, self._config_mgr)
 
             # Runtime API server is controlled by the sys config
 
@@ -311,20 +320,18 @@ def load_job_config(
 
         if isinstance(job_config, _cfg.JobConfig):
             self._log.info("Using embedded job config")
-            job_config_path = None
 
         else:
-            job_config_path = job_config
             job_config_dev_mode = _dev_mode.DEV_MODE_JOB_CONFIG if self._dev_mode else None
-            job_config_parser = _cparse.ConfigParser(_cfg.JobConfig, job_config_dev_mode)
-            job_config_raw = job_config_parser.load_raw_config(job_config_path, config_file_name="job")
-            job_config = job_config_parser.parse(job_config_raw, job_config_path)
+            job_config = self._config_mgr.load_config_object(
+                job_config, _cfg.JobConfig,
+                job_config_dev_mode,
+                config_file_name="job")
 
         if self._dev_mode:
-            config_dir = job_config_path.parent if job_config_path is not None else None
             job_config = _dev_mode.DevModeTranslator.translate_job_config(
                 self._sys_config, job_config,
-                self._scratch_dir, config_dir,
+                self._scratch_dir, self._config_mgr,
                 model_class)
 
         return job_config