[RAPTOR-14422] Drum fast fail when config file has issues (#1688)

* fast fails when config file is faulty

* fast fail when config file can not be loaded

* refactoring

* rebased

---------

Co-authored-by: Reza Mahdiani <reza.mahdiani@datarobot.com>

Author: reza-mahdiani-m

custom_model_runner/datarobot_drum/drum/drum.py

@@ -4,6 +4,7 @@
 This is proprietary source code of DataRobot, Inc. and its affiliates.
 Released under the terms of DataRobot Tool and Utility Agreement.
 """
+
 import contextlib
 import copy
 import json
@@ -775,12 +776,16 @@ def _prepare_prediction_server_or_batch_pipeline(self, run_language):
             "triton_grpc_port": int(options.triton_grpc_port),
             "api_token": options.api_token,
             "allow_dr_api_access": options.allow_dr_api_access,
-            "query_params": '"{}"'.format(options.query)
-            if getattr(options, "query", None) is not None
-            else "null",
-            "content_type": '"{}"'.format(options.content_type)
-            if getattr(options, "content_type", None) is not None
-            else "null",
+            "query_params": (
+                '"{}"'.format(options.query)
+                if getattr(options, "query", None) is not None
+                else "null"
+            ),
+            "content_type": (
+                '"{}"'.format(options.content_type)
+                if getattr(options, "content_type", None) is not None
+                else "null"
+            ),
             "target_type": self.target_type.value,
             "user_secrets_mount_path": getattr(options, "user_secrets_mount_path", None),
             "user_secrets_prefix": getattr(options, "user_secrets_prefix", None),
@@ -806,9 +811,11 @@ def _prepare_prediction_server_or_batch_pipeline(self, run_language):
                     "engine_type": "Generic",
                     "component_type": "prediction_server",
                     "processes": options.max_workers if getattr(options, "max_workers") else "null",
-                    "deployment_config": '"{}"'.format(options.deployment_config)
-                    if getattr(options, "deployment_config", None) is not None
-                    else "null",
+                    "deployment_config": (
+                        '"{}"'.format(options.deployment_config)
+                        if getattr(options, "deployment_config", None) is not None
+                        else "null"
+                    ),
                 }
             )
 

custom_model_runner/datarobot_drum/drum/exceptions.py

@@ -44,3 +44,11 @@ class DrumSerializationError(DrumException):
 
 class DrumRootComponentException(DrumException):
     """Raised when there is an issue specific to root components."""
+
+
+class UnrecoverableError(DrumException):
+    """A base exception for any error that is considered fatal and main runner should terminate immediately."""
+
+
+class UnrecoverableConfigurationError(UnrecoverableError):
+    """Raised when failure in parsing or validating configuration file."""

custom_model_runner/datarobot_drum/drum/gpu_predictors/vllm_predictor.py

@@ -4,13 +4,14 @@
 This is proprietary source code of DataRobot, Inc. and its affiliates.
 Released under the terms of DataRobot Tool and Utility Agreement.
 """
+
 import json
 import os
 import subprocess
 from pathlib import Path
 
 from datarobot_drum.drum.enum import CustomHooks
-from datarobot_drum.drum.exceptions import DrumCommonException
+from datarobot_drum.drum.exceptions import DrumCommonException, UnrecoverableConfigurationError
 from datarobot_drum.drum.gpu_predictors.base import BaseOpenAiGpuPredictor
 
 
@@ -60,8 +61,19 @@ def download_and_serve_model(self):
         # For advanced users, allow them to specify arbitrary CLI options that we haven't exposed
         # via runtime parameters.
         if engine_config_file.is_file():
-            config = json.loads(engine_config_file.read_text())
+            try:
+                config = json.loads(engine_config_file.read_text())
+            except Exception as e:
+                # Catch any other file-related errors
+                raise UnrecoverableConfigurationError(
+                    f"Failed to read or parse critical config file engine_config.json: {e}"
+                ) from e
             if "args" in config:
+                cli_args = config["args"]
+                if not isinstance(cli_args, list):
+                    raise UnrecoverableConfigurationError(
+                        f"Invalid configuration in engine_config.json: 'args' must be a list, but found type '{type(cli_args).__name__}'."
+                    )
                 self.logger.info(f"Loading CLI args from config file: {engine_config_file}...")
                 cmd.extend(config["args"])
 

custom_model_runner/datarobot_drum/drum/main.py

@@ -4,9 +4,15 @@
 This is proprietary source code of DataRobot, Inc. and its affiliates.
 Released under the terms of DataRobot Tool and Utility Agreement.
 """
+
+import logging
+import threading
+
 from flask import Flask
 from datarobot_drum.drum.gunicorn.context import WorkerCtx
 
+from datarobot_drum.drum.exceptions import UnrecoverableError
+
 #!/usr/bin/env python3
 
 """
@@ -142,5 +148,19 @@ def signal_handler(sig, frame):
             sys.exit(ExitCodes.SCHEMA_VALIDATION_ERROR.value)
 
 
+def _handle_thread_exception(args):
+    """
+    This global hook is called for any unhandled exception in any thread.
+    """
+    if issubclass(args.exc_type, UnrecoverableError):
+        logging.critical(
+            f"CRITICAL: An unrecoverable error occurred in thread '{args.thread.name}': {args.exc_value}. Terminating process immediately.",
+            exc_info=(args.exc_type, args.exc_value, args.exc_traceback),
+        )
+        os._exit(1)
+
+
+threading.excepthook = _handle_thread_exception
+
 if __name__ == "__main__":
     main()