diff --git a/projects/caliper/engine/parameter_matrix.py b/projects/caliper/engine/parameter_matrix.py index 69c95d3e..03e720e1 100644 --- a/projects/caliper/engine/parameter_matrix.py +++ b/projects/caliper/engine/parameter_matrix.py @@ -237,8 +237,4 @@ def create_legend_name( legend_name = ", ".join(param_pairs) - # Truncate if too long - if len(legend_name) > max_length: - legend_name = legend_name[: max_length - 3] + "..." - return legend_name diff --git a/projects/guidellm/toolbox/run_guidellm_benchmark/main.py b/projects/guidellm/toolbox/run_guidellm_benchmark/main.py index fa82ff1e..c7fe0b2a 100644 --- a/projects/guidellm/toolbox/run_guidellm_benchmark/main.py +++ b/projects/guidellm/toolbox/run_guidellm_benchmark/main.py @@ -120,19 +120,12 @@ def _best_effort_delete(description: str, *oc_args: str) -> None: @task def create_guidellm_resources_task(args, ctx): - """Create the GuideLLM benchmark PVC and job""" + """Create the GuideLLM benchmark job and PVC with job as owner""" # Ensure src directory exists (args.artifact_dir / "src").mkdir(parents=True, exist_ok=True) - oc_apply( - args.artifact_dir / "src" / "guidellm-pvc.yaml", - render_guidellm_pvc_from_parts( - namespace=ctx.target_namespace, - name=ctx.benchmark_name, - pvc_size=args.pvc_size, - ), - ) + # Create the job first oc_apply( args.artifact_dir / "src" / "guidellm-job.yaml", render_guidellm_job_from_parts( @@ -144,10 +137,35 @@ def create_guidellm_resources_task(args, ctx): hf_token_secret=args.hf_token_secret, ), ) - return f"GuideLLM benchmark {ctx.benchmark_name} created" + + # Get the job metadata for owner reference + job_data = oc_get_json("job", name=ctx.benchmark_name, namespace=ctx.target_namespace) + + # Create owner reference from job metadata + owner_reference = { + "apiVersion": "batch/v1", + "kind": "Job", + "name": job_data["metadata"]["name"], + "uid": job_data["metadata"]["uid"], + "controller": True, + "blockOwnerDeletion": True, + } + + # Create the PVC with job as owner + oc_apply( + args.artifact_dir / "src" / "guidellm-pvc.yaml", + render_guidellm_pvc_from_parts( + namespace=ctx.target_namespace, + name=ctx.benchmark_name, + pvc_size=args.pvc_size, + owner_reference=owner_reference, + ), + ) + + return f"GuideLLM benchmark {ctx.benchmark_name} created with job as PVC owner" -@retry(attempts=180, delay=10, backoff=1.0) +@retry(attempts=1080, delay=10, backoff=1.0) @task def wait_guidellm_benchmark_task(args, ctx): """Wait for the GuideLLM benchmark job to complete""" diff --git a/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py b/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py index 7cab6a2c..6366acce 100644 --- a/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py +++ b/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py @@ -146,13 +146,16 @@ def _build_multi_run_script(*, endpoint_url: str, runs: list[GuideLLMRun]) -> st return "\n".join(lines) -def render_guidellm_pvc_from_parts(*, namespace: str, name: str, pvc_size: str) -> dict[str, Any]: +def render_guidellm_pvc_from_parts( + *, namespace: str, name: str, pvc_size: str, owner_reference: dict[str, Any] | None = None +) -> dict[str, Any]: """Render a GuideLL-M PVC manifest from individual components. Args: namespace: Target namespace name: Name of the benchmark job and PVC pvc_size: Size of the PVC + owner_reference: Optional owner reference to set (e.g., for job ownership) Returns: PVC manifest as dict @@ -165,7 +168,13 @@ def render_guidellm_pvc_from_parts(*, namespace: str, name: str, pvc_size: str) "pvc_size": pvc_size, }, ) - return yaml.safe_load(rendered_yaml) + manifest = yaml.safe_load(rendered_yaml) + + # Add owner reference if provided + if owner_reference: + manifest["metadata"]["ownerReferences"] = [owner_reference] + + return manifest def render_guidellm_job_from_parts( diff --git a/projects/guidellm/toolbox/run_smoke_request/main.py b/projects/guidellm/toolbox/run_smoke_request/main.py index 5905beff..eba195b8 100644 --- a/projects/guidellm/toolbox/run_smoke_request/main.py +++ b/projects/guidellm/toolbox/run_smoke_request/main.py @@ -247,6 +247,7 @@ def capture_smoke_pod_debug_info(args, ctx): "-o", "yaml", check=False, + stdout_dest=artifacts_dir / f"{ctx.pod_name}.yaml", ) # Capture pod description @@ -283,6 +284,7 @@ def cleanup_smoke_pod(args, ctx): "-n", args.namespace, "--ignore-not-found=true", + "--wait=false", check=False, ) return f"Cleaned up smoke pod {ctx.pod_name}" diff --git a/projects/kserve/toolbox/capture_llmisvc_state/main.py b/projects/kserve/toolbox/capture_llmisvc_state/main.py index 12efa193..06096344 100644 --- a/projects/kserve/toolbox/capture_llmisvc_state/main.py +++ b/projects/kserve/toolbox/capture_llmisvc_state/main.py @@ -26,6 +26,7 @@ def setup_directories(args, context): """Create the artifacts directory""" shell.mkdir("artifacts") + shell.mkdir("artifacts/logs") return "Artifacts directory created" @@ -121,7 +122,7 @@ def capture_namespace_pods(args, context): """Capture all pods in the namespace with wide output""" shell.run( f"oc get pods -owide -n {context.target_namespace}", - stdout_dest=args.artifact_dir / "artifacts/namespace.pods.status", + stdout_dest=args.artifact_dir / "artifacts/namespace.pods.status.txt", check=False, ) return "Namespace pods status captured" @@ -132,7 +133,7 @@ def capture_namespace_services(args, context): """Capture all services in the namespace""" shell.run( f"oc get svc -n {context.target_namespace}", - stdout_dest=args.artifact_dir / "artifacts/namespace.services.status", + stdout_dest=args.artifact_dir / "artifacts/namespace.services.status.txt", check=False, ) return "Namespace services captured" @@ -173,20 +174,19 @@ def capture_pod_logs(args, context): if not pod_names or not result.stdout.strip(): return "No pods found to capture logs" - log_file = args.artifact_dir / "artifacts/llminferenceservice.pods.logs" + logs_dir = args.artifact_dir / "artifacts/logs" + captured_count = 0 - with open(log_file, "w") as handle: - for pod_name in pod_names: - handle.write(f"=== Logs for pod: {pod_name} ===\n") - log_result = shell.run( - f"oc logs {pod_name} -n {context.target_namespace} --all-containers=true", - check=False, - log_stdout=False, - ) - handle.write(log_result.stdout) - handle.write("\n") + for pod_name in pod_names: + log_file = logs_dir / f"{pod_name}.log" + shell.run( + f"oc logs {pod_name} -n {context.target_namespace} --all-containers=true", + stdout_dest=log_file, + check=False, + ) + captured_count += 1 - return f"Pod logs captured for {len(pod_names)} pods" + return f"Pod logs captured for {captured_count} pods in dedicated files" @task @@ -195,26 +195,26 @@ def capture_pod_previous_logs(args, context): result = shell.run( f'oc get pods -l "app.kubernetes.io/name={args.llmisvc_name}" -n {context.target_namespace} -o jsonpath="{{.items[*].metadata.name}}"', check=False, + log_stdout=False, ) pod_names = result.stdout.strip().split() if not pod_names or not result.stdout.strip(): return "No pods found to capture previous logs" - log_file = args.artifact_dir / "artifacts/llminferenceservice.pods.previous.logs" + logs_dir = args.artifact_dir / "artifacts/logs" + captured_count = 0 - with open(log_file, "w") as handle: - for pod_name in pod_names: - handle.write(f"=== Previous logs for pod: {pod_name} ===\n") - log_result = shell.run( - f"oc logs {pod_name} -n {context.target_namespace} --previous --all-containers=true", - check=False, - log_stdout=False, - ) - handle.write(log_result.stdout) - handle.write("\n") + for pod_name in pod_names: + log_file = logs_dir / f"{pod_name}.previous.log" + shell.run( + f"oc logs {pod_name} -n {context.target_namespace} --previous --all-containers=true", + stdout_dest=log_file, + check=False, + ) + captured_count += 1 - return f"Pod previous logs captured for {len(pod_names)} pods" + return f"Pod previous logs captured for {captured_count} pods in dedicated files" @task diff --git a/projects/kserve/toolbox/deploy_llmisvc/__init__.py b/projects/kserve/toolbox/deploy_llmisvc/__init__.py index e69de29b..68f67b1f 100644 --- a/projects/kserve/toolbox/deploy_llmisvc/__init__.py +++ b/projects/kserve/toolbox/deploy_llmisvc/__init__.py @@ -0,0 +1,5 @@ +from __future__ import annotations + +from projects.kserve.toolbox.deploy_llmisvc.main import try_resolve_endpoint_url + +__all__ = ["try_resolve_endpoint_url"] diff --git a/projects/kserve/toolbox/deploy_llmisvc/main.py b/projects/kserve/toolbox/deploy_llmisvc/main.py index cf064757..c3efade8 100644 --- a/projects/kserve/toolbox/deploy_llmisvc/main.py +++ b/projects/kserve/toolbox/deploy_llmisvc/main.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from pathlib import Path import yaml @@ -24,6 +25,8 @@ from .on_failure_helpers import on_wait_pods_appear_failure +logger = logging.getLogger(__name__) + def load_yaml(path: Path): with path.open(encoding="utf-8") as handle: @@ -35,8 +38,9 @@ def run( *, namespace: str, inference_service_manifest_path: str, - gateway_status_address_name: str = "gateway-external", + gateway_status_address_name: str | None = "gateway-external", dry_run: bool = False, + wait_pods_scheduled: bool = False, ) -> str: """ Deploy an LLMInferenceService and wait for its endpoint. @@ -46,6 +50,7 @@ def run( inference_service_manifest_path: Path to the InferenceService YAML manifest file gateway_status_address_name: Gateway status address name for endpoint resolution dry_run: If True, only prepare the manifest without deploying + wait_pods_scheduled: If True, wait for all pods to be scheduled before checking service readiness """ ctx = execute_tasks(locals()) @@ -119,10 +124,20 @@ def delete_existing_service(args, ctx): def wait_old_pods_gone(args, ctx): """Wait for old llm-d pods to disappear""" - pods = oc_get_json( - "pods", namespace=args.namespace, selector=ctx.selector, ignore_not_found=True + result = oc( + "get", + "pods", + "-n", + args.namespace, + "-l", + ctx.selector, + "--ignore-not-found=true", + "--no-headers", + check=False, ) - if not pods or not pods.get("items"): + + # Check if output is empty (no pods found) + if not result.stdout.strip(): return f"Old pods gone for {ctx.inference_service_name}" return False # Retry @@ -154,6 +169,311 @@ def wait_pods_appear(args, ctx): return False # Retry +@task +def query_service_status(args, ctx): + """Query the status of the LLMInferenceService""" + + service_name = ctx.inference_service_name + + # Query only the Ready condition status + result = oc( + "get", + "llminferenceservice", + service_name, + "-n", + args.namespace, + "-o", + "jsonpath={.status.conditions[?(@.type=='Ready')].status}", + log_stdout=False, + ) + + ready_status = result.stdout.strip() + ctx.is_ready = ready_status == "True" + + if ctx.is_ready: + return f"LLMInferenceService {service_name} status: Ready" + else: + return f"LLMInferenceService {service_name} status: Not Ready" + + +@task +def query_service_message(args, ctx): + """Query detailed message from LLMInferenceService""" + + service_name = ctx.inference_service_name + + # Query the Ready condition details + result = oc( + "get", + "llminferenceservice", + service_name, + "-n", + args.namespace, + "-o", + "jsonpath={.status.conditions[?(@.type=='Ready')]}", + log_stdout=False, + ) + + if result.stdout.strip(): + try: + import json + + condition = json.loads(result.stdout) + reason = condition.get("reason", "Unknown") + message = condition.get("message", "No message") + + if not ctx.is_ready: + return f"Not ready - Reason: {reason}, Message: {message}" + else: + return "Ready - Service is operational" + except (json.JSONDecodeError, KeyError) as e: + return f"Failed to parse Ready condition: {e}" + else: + return "No Ready condition found in status" + + +@retry(attempts=999999, delay=30, backoff=1.0) +@task +def wait_pods_scheduled(args, ctx): + """Wait for all pods to be scheduled (optional task)""" + + # Check if this task is enabled + if not args.wait_pods_scheduled: + return "Pod scheduling wait disabled by parameter" + + service_name = ctx.inference_service_name + + # Get pod status using plain text output + result = oc( + "get", + "pods", + "-l", + ctx.selector, + "-n", + args.namespace, + "--no-headers", + check=False, + log_stdout=False, + ) + + if not result.stdout.strip(): + return False, "No pods found for the service yet" + + # Keep waiting if any pod is Pending or SchedulingGated + if "Pending" in result.stdout: + return False, "Waiting for pods to exit Pending state" + + if "SchedulingGated" in result.stdout: + return False, "Waiting for pods to exit SchedulingGated state" + + return f"All pods for {service_name} are scheduled successfully" + + +@retry(attempts=90, delay=10, backoff=1.0) +@task +def wait_service_ready(args, ctx): + """Wait for LLMInferenceService to be ready""" + + service_name = ctx.inference_service_name + + # Query the current status and show diagnostic info + result = oc( + "get", + "llminferenceservice", + service_name, + "-n", + args.namespace, + "-o", + "jsonpath={.status.conditions[?(@.type=='Ready')]}", + ) + + # Also show pod status for debugging + oc( + "get", + "pods", + "-l", + ctx.selector, + "-n", + args.namespace, + ) + + # Check for pod restarts and abort if any pods have restarted + restart_result = oc( + "get", + "pods", + "-l", + ctx.selector, + "-n", + args.namespace, + "-o", + "jsonpath={range .items[*]}{.metadata.name}:{.status.containerStatuses[*].restartCount}{'\\n'}{end}", + log_stdout=False, + ) + + if restart_result.stdout.strip(): + for line in restart_result.stdout.strip().split("\n"): + if ":" in line: + pod_name, restart_counts = line.split(":", 1) + # Check if any container has restarted + counts = restart_counts.split() + for count_str in counts: + try: + if int(count_str) > 0: + raise RuntimeError( + f"Pod {pod_name} has restarted (restart count: {count_str}). Aborting wait due to pod restart." + ) + except ValueError: + # Skip non-numeric restart counts + pass + + if result.stdout.strip(): + try: + import json + + condition = json.loads(result.stdout) + status = condition.get("status", "Unknown") + reason = condition.get("reason", "Unknown") + message = condition.get("message", "No message") + + if status == "True": + return f"LLMInferenceService {service_name} is ready" + else: + return ( + False, + f"Service not ready - Status: {status}, Reason: {reason}, Message: {message}", + ) + + except (json.JSONDecodeError, KeyError) as e: + return (False, f"Failed to parse Ready condition: {e}") + else: + return (False, f"No Ready condition found in status for {service_name}") + + +def try_resolve_endpoint_url( + *, namespace: str, inference_service_name: str, gateway_status_address_name: str | None +) -> str | None: + logger.info( + f"=== Resolving endpoint URL for {inference_service_name} in namespace {namespace} ===" + ) + logger.info(f"Target gateway_status_address_name: {gateway_status_address_name}") + + payload = oc_get_json("llminferenceservice", name=inference_service_name, namespace=namespace) + + # Log the entire status section for debugging + status = payload.get("status", {}) + logger.info(f"Status section keys found: {list(status.keys())}") + + # Check status.address first + status_address = status.get("address") + logger.info(f"status.address content: {status_address}") + + if status_address: + logger.info("✓ status.address exists") + if isinstance(status_address, dict): + logger.info("✓ status.address is a dict") + if status_address.get("url"): + url = status_address["url"] + logger.info(f"✓ Found URL in status.address: '{url}'") + + # When gateway_status_address_name is None, append port 8000 if needed + if gateway_status_address_name is None: + logger.info("Mode: No gateway (will append port 8000 if needed)") + if ":" not in url.split("/")[-1]: # Check if no port in the hostname part + url = f"{url}:8000" + logger.info(f"✓ Appended port 8000: '{url}'") + else: + logger.info("✓ URL already has port, using as-is") + else: + logger.info( + f"Mode: Gateway '{gateway_status_address_name}' (no port modification)" + ) + + logger.info(f"🎯 RESOLVED from status.address: '{url}'") + return url + else: + logger.info("✗ status.address has no 'url' field") + logger.info(f" Available fields: {list(status_address.keys())}") + else: + logger.info(f"✗ status.address is not a dict, type: {type(status_address)}") + else: + logger.info("✗ No status.address found") + + # Fallback to existing status.addresses logic + logger.info("--- Falling back to status.addresses lookup ---") + status_addresses = status.get("addresses", []) + logger.info(f"status.addresses content: {status_addresses}") + + if not status_addresses: + logger.info("✗ No status.addresses found - RESOLUTION FAILED") + return None + + logger.info(f"✓ Found {len(status_addresses)} address(es) in status.addresses") + + for i, address in enumerate(status_addresses): + logger.info(f"Checking address[{i}]: {address}") + + # When gateway_status_address_name is None, return the first address with a URL and append port 8000 + if gateway_status_address_name is None: + logger.info(" Mode: No gateway filter (looking for any URL)") + if address.get("url"): + url = address["url"] + logger.info(f" ✓ Found URL in address[{i}]: '{url}'") + # Append port 8000 when not using gateway if no port is already specified + if ":" not in url.split("/")[-1]: # Check if no port in the hostname part + url = f"{url}:8000" + logger.info(f" ✓ Appended port 8000: '{url}'") + else: + logger.info(" ✓ URL already has port, using as-is") + + logger.info(f"🎯 RESOLVED from status.addresses[{i}]: '{url}'") + return url + else: + logger.info(f" ✗ Address[{i}] has no 'url' field") + logger.info(f" Available fields: {list(address.keys())}") + # Otherwise, match by name + else: + address_name = address.get("name") + logger.info( + f" Mode: Gateway filter (looking for name='{gateway_status_address_name}')" + ) + logger.info(f" Address[{i}] name: '{address_name}'") + + if address_name == gateway_status_address_name: + logger.info(f" ✓ Name matches '{gateway_status_address_name}'") + if address.get("url"): + url = address["url"] + logger.info(f" ✓ Found URL: '{url}'") + logger.info(f"🎯 RESOLVED from status.addresses[{i}] by name: '{url}'") + return url + else: + logger.info(" ✗ Matching address has no 'url' field") + logger.info(f" Available fields: {list(address.keys())}") + else: + logger.info( + f" ✗ Name mismatch: '{address_name}' != '{gateway_status_address_name}'" + ) + + logger.info("❌ RESOLUTION FAILED - No suitable URL found in any address") + return None + + +@retry(attempts=30, delay=10, backoff=1.0) +@task +def resolve_endpoint_task(args, ctx): + """Resolve the gateway endpoint URL""" + + endpoint_url = try_resolve_endpoint_url( + namespace=args.namespace, + inference_service_name=ctx.inference_service_name, + gateway_status_address_name=args.gateway_status_address_name, + ) + if endpoint_url: + ctx.endpoint_url = endpoint_url + write_text(args.artifact_dir / "artifacts" / "endpoint.url", f"{endpoint_url}\n") + return f"Endpoint resolved: {endpoint_url}" + return False, "No endpoint URL available" + + @always @task def capture_llmisv_description(args, ctx): @@ -257,148 +577,255 @@ def capture_replicaset_description(args, ctx): return f"Failed to capture ReplicaSet description: {e}" +@always @task -def query_service_status(args, ctx): - """Query the status of the LLMInferenceService""" +def capture_final_llmisvc_yaml(args, ctx): + """Capture the final YAML state of the LLMInferenceService""" - service_name = ctx.inference_service_name + if args.dry_run: + return "Dry-run, nothing to do" - # Query only the Ready condition status - result = oc( - "get", - "llminferenceservice", - service_name, - "-n", - args.namespace, - "-o", - "jsonpath={.status.conditions[?(@.type=='Ready')].status}", - log_stdout=False, - ) + try: + # Ensure artifacts directory exists + artifacts_dir = args.artifact_dir / "artifacts" + artifacts_dir.mkdir(parents=True, exist_ok=True) - ready_status = result.stdout.strip() - ctx.is_ready = ready_status == "True" + # Use LLMInferenceService name from context + service_name = getattr(ctx, "inference_service_name", None) + if not service_name: + return "No service name available" - if ctx.is_ready: - return f"LLMInferenceService {service_name} status: Ready" - else: - return f"LLMInferenceService {service_name} status: Not Ready" + # Capture final YAML state + result = oc( + "get", + "llminferenceservice", + service_name, + "-n", + args.namespace, + "-o", + "yaml", + log_stdout=False, + check=False, + ) + llmisvc_yaml_path = artifacts_dir / "llmisvc_final.yaml" -@task -def query_service_message(args, ctx): - """Query detailed message from LLMInferenceService""" + if result.returncode != 0: + # Handle the case where the LLMInferenceService is not found + if "not found" in result.stderr.lower(): + logger.warning( + f"LLMInferenceService '{service_name}' not found in namespace '{args.namespace}'" + ) + with open(llmisvc_yaml_path, "w", encoding="utf-8") as f: + f.write( + f"# LLMInferenceService '{service_name}' not found in namespace '{args.namespace}'\n" + ) + f.write(f"# Error: {result.stderr.strip()}\n") + return f"LLMInferenceService not found, logged error to {llmisvc_yaml_path}" + else: + # Handle other error cases + logger.error( + f"Failed to get LLMInferenceService '{service_name}': {result.stderr.strip()}" + ) + with open(llmisvc_yaml_path, "w", encoding="utf-8") as f: + f.write(f"# Error getting LLMInferenceService '{service_name}'\n") + f.write(f"# Error: {result.stderr.strip()}\n") + return f"Error getting LLMInferenceService, logged error to {llmisvc_yaml_path}" - service_name = ctx.inference_service_name + # Success case - write the YAML content + with open(llmisvc_yaml_path, "w", encoding="utf-8") as f: + f.write(result.stdout) - # Query the Ready condition details - result = oc( - "get", - "llminferenceservice", - service_name, - "-n", - args.namespace, - "-o", - "jsonpath={.status.conditions[?(@.type=='Ready')]}", - log_stdout=False, - ) + return f"Captured final LLMInferenceService YAML to {llmisvc_yaml_path}" - if result.stdout.strip(): - try: - import json + except Exception as e: + return f"Failed to capture final LLMInferenceService YAML: {e}" - condition = json.loads(result.stdout) - reason = condition.get("reason", "Unknown") - message = condition.get("message", "No message") - if not ctx.is_ready: - return f"Not ready - Reason: {reason}, Message: {message}" - else: - return "Ready - Service is operational" - except (json.JSONDecodeError, KeyError) as e: - return f"Failed to parse Ready condition: {e}" - else: - return "No Ready condition found in status" +@always +@task +def capture_workload_overview(args, ctx): + """Capture deployment, replicaset, and pod overview for debugging""" + if args.dry_run: + return "Dry-run, nothing to do" -@retry(attempts=90, delay=10, backoff=1.0) -@task -def wait_service_ready(args, ctx): - """Wait for LLMInferenceService to be ready""" + # Ensure artifacts directory exists + artifacts_dir = args.artifact_dir / "artifacts" + artifacts_dir.mkdir(parents=True, exist_ok=True) - service_name = ctx.inference_service_name + # Use selector from context + selector = getattr(ctx, "selector", None) + if not selector: + return "No selector available" - # Query the current status and show diagnostic info - result = oc( - "get", - "llminferenceservice", - service_name, - "-n", - args.namespace, - "-o", - "jsonpath={.status.conditions[?(@.type=='Ready')]}", - log_stdout=True, - ) + workload_overview_path = artifacts_dir / "workload_overview.txt" - # Also show pod status for debugging + # Capture deployment, replicaset, and pod overview oc( "get", - "pods", + "deploy,rs,pod", "-l", - ctx.selector, + selector, "-n", args.namespace, - log_stdout=True, # Show pod status in logs + "-o", + "wide", + check=False, + stdout_dest=workload_overview_path, ) - if result.stdout.strip(): - try: - import json + return f"Captured workload overview to {workload_overview_path}" - condition = json.loads(result.stdout) - status = condition.get("status", "Unknown") - reason = condition.get("reason", "Unknown") - message = condition.get("message", "No message") - if status == "True": - return f"LLMInferenceService {service_name} is ready" - else: - return ( - False, - f"Service not ready - Status: {status}, Reason: {reason}, Message: {message}", - ) +@always +@task +def capture_pod_status(args, ctx): + """Capture pod status for debugging""" - except (json.JSONDecodeError, KeyError) as e: - return (False, f"Failed to parse Ready condition: {e}") - else: - return (False, f"No Ready condition found in status for {service_name}") + if args.dry_run: + return "Dry-run, nothing to do" + try: + # Ensure artifacts directory exists + artifacts_dir = args.artifact_dir / "artifacts" + artifacts_dir.mkdir(parents=True, exist_ok=True) -@retry(attempts=90, delay=10, backoff=1.0) + # Use selector from context + selector = getattr(ctx, "selector", None) + if not selector: + return "No selector available" + + # Capture pod status with wide output + result = oc( + "get", + "pods", + "-l", + selector, + "-n", + args.namespace, + "-o", + "wide", + log_stdout=False, + check=False, + ) + + pod_status_path = artifacts_dir / "pod_status.txt" + with open(pod_status_path, "w", encoding="utf-8") as f: + f.write(result.stdout) + + return f"Captured pod status to {pod_status_path}" + + except Exception as e: + return f"Failed to capture pod status: {e}" + + +@always @task -def resolve_endpoint_task(args, ctx): - """Resolve the gateway endpoint URL""" +def capture_pod_descriptions(args, ctx): + """Capture pod descriptions for debugging""" - endpoint_url = try_resolve_endpoint_url( - namespace=args.namespace, - inference_service_name=ctx.inference_service_name, - gateway_status_address_name=args.gateway_status_address_name, - ) - if endpoint_url: - ctx.endpoint_url = endpoint_url - write_text(args.artifact_dir / "artifacts" / "endpoint.url", f"{endpoint_url}\n") - return f"Endpoint resolved: {endpoint_url}" - return False # Retry + if args.dry_run: + return "Dry-run, nothing to do" + try: + # Ensure artifacts directory exists + artifacts_dir = args.artifact_dir / "artifacts" + artifacts_dir.mkdir(parents=True, exist_ok=True) -def try_resolve_endpoint_url( - *, namespace: str, inference_service_name: str, gateway_status_address_name: str -) -> str | None: - payload = oc_get_json("llminferenceservice", name=inference_service_name, namespace=namespace) + # Use selector from context + selector = getattr(ctx, "selector", None) + if not selector: + return "No selector available" - for address in payload.get("status", {}).get("addresses", []): - if address.get("name") == gateway_status_address_name and address.get("url"): - return address["url"] - return None + # Get pod names + pod_result = oc( + "get", + "pods", + "-l", + selector, + "-n", + args.namespace, + "-o", + "jsonpath={.items[*].metadata.name}", + log_stdout=False, + check=False, + ) + + pod_names = pod_result.stdout.strip().split() + if not pod_names or not pod_result.stdout.strip(): + pod_desc_path = artifacts_dir / "pod_descriptions.txt" + with open(pod_desc_path, "w", encoding="utf-8") as f: + f.write("No pods found for the service") + return f"No pods found, wrote empty file to {pod_desc_path}" + + # Describe each pod + pod_descriptions = [] + for pod_name in pod_names: + describe_result = oc( + "describe", + "pod", + pod_name, + "-n", + args.namespace, + log_stdout=False, + check=False, + ) + pod_descriptions.append( + f"=== Description for pod: {pod_name} ===\n{describe_result.stdout}" + ) + + # Save all pod descriptions + pod_desc_path = artifacts_dir / "pod_descriptions.txt" + with open(pod_desc_path, "w", encoding="utf-8") as f: + f.write("\n\n".join(pod_descriptions)) + + return f"Captured descriptions for {len(pod_names)} pods to {pod_desc_path}" + + except Exception as e: + return f"Failed to capture pod descriptions: {e}" + + +@always +@task +def capture_pod_yaml(args, ctx): + """Capture pod YAML definitions for debugging""" + + if args.dry_run: + return "Dry-run, nothing to do" + + try: + # Ensure artifacts directory exists + artifacts_dir = args.artifact_dir / "artifacts" + artifacts_dir.mkdir(parents=True, exist_ok=True) + + # Use selector from context + selector = getattr(ctx, "selector", None) + if not selector: + return "No selector available" + + # Capture all pod YAMLs + result = oc( + "get", + "pods", + "-l", + selector, + "-n", + args.namespace, + "-o", + "yaml", + log_stdout=False, + check=False, + ) + + pod_yaml_path = artifacts_dir / "pod_definitions.yaml" + with open(pod_yaml_path, "w", encoding="utf-8") as f: + f.write(result.stdout) + + return f"Captured pod YAML definitions to {pod_yaml_path}" + + except Exception as e: + return f"Failed to capture pod YAML: {e}" if __name__ == "__main__": diff --git a/projects/kserve/toolbox/prepare_hf_model_cache/main.py b/projects/kserve/toolbox/prepare_hf_model_cache/main.py index a25544e8..1e01c173 100644 --- a/projects/kserve/toolbox/prepare_hf_model_cache/main.py +++ b/projects/kserve/toolbox/prepare_hf_model_cache/main.py @@ -286,7 +286,45 @@ def create_download_job(args, ctx): return f"Download job {cache_spec['download_job_name']} created" -@retry(attempts=30, delay=30, backoff=1.0) +@retry(attempts=24, delay=5, backoff=1.0) # 24 attempts * 5 seconds = 120 seconds (2 minutes) +@task +def wait_for_pods_running(args, ctx): + """Wait for the download job pods to start running within 2 minutes""" + + if getattr(ctx, "cache_ready", False): + return "Skipping pod wait - cache already ready" + + cache_spec = ctx.cache_spec + job_name = cache_spec["download_job_name"] + namespace = cache_spec["namespace"] + + # Get pod status using plain text output + result = oc( + "get", + "pod", + "--no-headers", + "-l", + f"job-name={job_name}", + "-n", + namespace, + "-o", + "custom-columns=NAME:.metadata.name,STATUS:.status.phase", + check=False, + log_stdout=False, + ) + + if not result.stdout.strip(): + return False # No pods yet, retry + + # Keep waiting only if "Pending" appears in the output + if "Pending" in result.stdout: + return False, "Waiting for pods to exit Pending state" + + # All pods are past Pending state + return "All pods past Pending state" + + +@retry(attempts=120, delay=30, backoff=1.0) @task def wait_for_download(args, ctx): """Wait for the download job to complete""" diff --git a/projects/llm_d/tests/test_deployment_profiles.py b/projects/llm_d/tests/test_deployment_profiles.py index 631ab48a..0663760a 100644 --- a/projects/llm_d/tests/test_deployment_profiles.py +++ b/projects/llm_d/tests/test_deployment_profiles.py @@ -67,13 +67,25 @@ def _test_preset_generates_expected_llmisvc(preset: str, tmp_path: Path): ci_script = PROJECT_ROOT / "projects" / "llm_d" / "orchestration" / "ci.py" cmd = [str(ci_script), "--preset", preset, "test"] - result = subprocess.run( - cmd, - capture_output=True, - text=True, - cwd=PROJECT_ROOT, - env=env_vars, - ) + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + cwd=PROJECT_ROOT, + env=env_vars, + timeout=60, # 60-second timeout + ) + except subprocess.TimeoutExpired as e: + # Show stdout when timeout occurs + stdout_output = e.stdout.decode() if e.stdout else "No stdout available" + stderr_output = e.stderr.decode() if e.stderr else "No stderr available" + pytest.fail( + f"CI command timed out after 60 seconds for preset {preset}:\n" + f"Command: {' '.join(cmd)}\n\n" + f"STDOUT (before timeout):\n{stdout_output}\n\n" + f"STDERR (before timeout):\n{stderr_output}\n\n" + ) # Check that the command succeeded if result.returncode != 0: