feat: introduce inspect command

Signed-off-by: thxCode <thxcode0824@gmail.com>

Author: thxCode

gpustack_runtime/__main__.py

@@ -19,6 +19,7 @@
     DetectDevicesSubCommand,
     ExecWorkloadSubCommand,
     GetWorkloadSubCommand,
+    InspectWorkloadSubCommand,
     ListImagesSubCommand,
     ListWorkloadsSubCommand,
     LogsWorkloadSubCommand,
@@ -65,6 +66,7 @@ def main():
     ListWorkloadsSubCommand.register(subcommand_parser)
     LogsWorkloadSubCommand.register(subcommand_parser)
     ExecWorkloadSubCommand.register(subcommand_parser)
+    InspectWorkloadSubCommand.register(subcommand_parser)
     DetectDevicesSubCommand.register(subcommand_parser)
     ListImagesSubCommand.register(subcommand_parser)
     SaveImagesSubCommand.register(subcommand_parser)

gpustack_runtime/cmds/__init__.py

@@ -7,6 +7,7 @@
     DeleteWorkloadSubCommand,
     ExecWorkloadSubCommand,
     GetWorkloadSubCommand,
+    InspectWorkloadSubCommand,
     ListWorkloadsSubCommand,
     LogsWorkloadSubCommand,
 )
@@ -29,6 +30,7 @@
     "DetectDevicesSubCommand",
     "ExecWorkloadSubCommand",
     "GetWorkloadSubCommand",
+    "InspectWorkloadSubCommand",
     "ListImagesSubCommand",
     "ListWorkloadsSubCommand",
     "LogsWorkloadSubCommand",

gpustack_runtime/cmds/deployer.py

@@ -30,6 +30,7 @@
     get_workload,
     list_workloads,
 )
+from ..deployer.__utils__ import safe_json, safe_yaml
 from ..detector import supported_backends
 from .__types__ import SubCommand
 
@@ -77,6 +78,19 @@
     "_DRIVER_CAPABILITIES",
 )
 
+_IGNORE_SENSITIVE_ENVS_SUFFIX = (
+    "_KEY",
+    "_key",
+    "_TOKEN",
+    "_token",
+    "_SECRET",
+    "_secret",
+    "_PASSWORD",
+    "_password",
+    "_PASS",
+    "_pass",
+)
+
 
 class CreateRunnerWorkloadSubCommand(SubCommand):
     """
@@ -966,6 +980,72 @@ def sockets(self):
         pty.PseudoTerminal(None, exec_op).start()
 
 
+class InspectWorkloadSubCommand(SubCommand):
+    """
+    Command to diagnose a workload deployment.
+    """
+
+    @staticmethod
+    def register(parser: _SubParsersAction):
+        inspect_parser = parser.add_parser(
+            "inspect",
+            help="Inspect a workload deployment",
+        )
+
+        inspect_parser.add_argument(
+            "--namespace",
+            type=str,
+            help="Namespace of the workloads",
+        )
+
+        inspect_parser.add_argument(
+            "name",
+            type=str,
+            help="Name of the workload",
+        )
+
+        inspect_parser.set_defaults(func=InspectWorkloadSubCommand)
+
+    def __init__(self, args: Namespace):
+        self.namespace = args.namespace
+        self.name = args.name
+
+        if not self.name:
+            msg = "The name argument is required."
+            raise ValueError(msg)
+
+    def run(self):
+        workload = get_workload(self.name, self.namespace)
+        if not workload:
+            print(f"Workload '{self.name}' not found.")
+            return
+
+        if hasattr(workload, "_d_containers"):
+            result = []
+            for c in workload._d_containers:  # noqa: SLF001
+                c_attrs = c.attrs
+                # Mask sensitive environment variables
+                if "Env" in c_attrs["Config"]:
+                    for i, env in enumerate(c_attrs["Config"]["Env"] or []):
+                        env_name, _ = env.split("=", maxsplit=1)
+                        if env_name.endswith(_IGNORE_SENSITIVE_ENVS_SUFFIX):
+                            c_attrs["Config"]["Env"][i] = f"{env_name}=******"
+                result.append(c_attrs)
+            print(safe_json(result, indent=2))
+        elif hasattr(workload, "_k_pod"):
+            k_pod = workload._k_pod  # noqa: SLF001
+            # Remove managed fields to reduce output size
+            k_pod.metadata.managed_fields = None
+            # Mask sensitive environment variables
+            for c in k_pod.spec.containers:
+                for env in c.env or []:
+                    if env.name.endswith(_IGNORE_SENSITIVE_ENVS_SUFFIX):
+                        env.value = "******"
+            print(safe_yaml(k_pod, indent=2, sort_keys=False))
+        else:
+            print("No detailed inspection information available for this workload.")
+
+
 def format_workloads_json(sts: list[WorkloadStatus]) -> str:
     return json.dumps([st.to_dict() for st in sts], indent=2)