diff --git a/plugins/nemo-auditor/src/nemo_auditor/api/v2/configs.py b/plugins/nemo-auditor/src/nemo_auditor/api/v2/configs.py index 1407ea4304..6a7641cbc8 100644 --- a/plugins/nemo-auditor/src/nemo_auditor/api/v2/configs.py +++ b/plugins/nemo-auditor/src/nemo_auditor/api/v2/configs.py @@ -25,6 +25,7 @@ get_entity_client, ) from nemo_platform_plugin.jobs.openapi_utils import generate_openapi_extra_params +from nemo_platform_plugin.log_utils import sanitize_for_log logger = logging.getLogger(__name__) @@ -165,7 +166,19 @@ async def update_config( detail=f"AuditConfig '{name}' not found in workspace '{workspace}'.", ) from exc except NemoEntityConflictError as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc + logger.info( + "Conflict updating audit config '%s' in workspace '%s'", + sanitize_for_log(name), + sanitize_for_log(workspace), + exc_info=True, + ) + raise HTTPException( + status_code=409, + detail=( + f"AuditConfig '{name}' was modified by another request in workspace '{workspace}'. " + "Refresh the config and try again." + ), + ) from exc except Exception as exc: logger.exception("Failed to update audit config '%s'", name) raise HTTPException(status_code=500, detail="Failed to update audit config.") from exc diff --git a/plugins/nemo-auditor/src/nemo_auditor/api/v2/targets.py b/plugins/nemo-auditor/src/nemo_auditor/api/v2/targets.py index b3acd291e3..e42429498e 100644 --- a/plugins/nemo-auditor/src/nemo_auditor/api/v2/targets.py +++ b/plugins/nemo-auditor/src/nemo_auditor/api/v2/targets.py @@ -24,6 +24,7 @@ get_entity_client, ) from nemo_platform_plugin.jobs.openapi_utils import generate_openapi_extra_params +from nemo_platform_plugin.log_utils import sanitize_for_log logger = logging.getLogger(__name__) @@ -162,7 +163,19 @@ async def update_target( detail=f"AuditTarget '{name}' not found in workspace '{workspace}'.", ) from exc except NemoEntityConflictError as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc + logger.info( + "Conflict updating audit target '%s' in workspace '%s'", + sanitize_for_log(name), + sanitize_for_log(workspace), + exc_info=True, + ) + raise HTTPException( + status_code=409, + detail=( + f"AuditTarget '{name}' was modified by another request in workspace '{workspace}'. " + "Refresh the target and try again." + ), + ) from exc except Exception as exc: logger.exception("Failed to update audit target '%s'", name) raise HTTPException(status_code=500, detail="Failed to update audit target.") from exc diff --git a/plugins/nemo-auditor/tests/test_api_configs.py b/plugins/nemo-auditor/tests/test_api_configs.py index 6f07d45a12..25e6124144 100644 --- a/plugins/nemo-auditor/tests/test_api_configs.py +++ b/plugins/nemo-auditor/tests/test_api_configs.py @@ -11,6 +11,7 @@ from __future__ import annotations +import logging from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock @@ -199,6 +200,41 @@ def test_invalid_payload_returns_422(self, client, mock_entity_client) -> None: ) assert resp.status_code == 422 + def test_conflict_hides_raw_exception_details(self, client, mock_entity_client) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_config("cfg-1")) + mock_entity_client.update = AsyncMock( + side_effect=NemoEntityConflictError("Error code: 409 - {'detail': 'db_version mismatch'}") + ) + + resp = client.put( + "/apis/auditor/v2/workspaces/default/configs/cfg-1", + json={"description": "new"}, + ) + + assert resp.status_code == 409 + detail = resp.json()["detail"] + assert "AuditConfig 'cfg-1'" in detail + assert "Refresh the config" in detail + assert "Error code" not in detail + assert "db_version" not in detail + + def test_conflict_sanitizes_log_fields(self, client, mock_entity_client, caplog) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_config("cfg-1")) + mock_entity_client.update = AsyncMock(side_effect=NemoEntityConflictError("conflict")) + + with caplog.at_level(logging.INFO, logger=configs_router_module.__name__): + resp = client.put( + "/apis/auditor/v2/workspaces/default%0Aforged/configs/cfg-1%0D%0Aforged", + json={"description": "new"}, + ) + + assert resp.status_code == 409 + message = next( + record.getMessage() for record in caplog.records if "Conflict updating audit config" in record.msg + ) + assert "\r" not in message + assert "\n" not in message + class TestDeleteConfig: def test_returns_204(self, client, mock_entity_client) -> None: diff --git a/plugins/nemo-auditor/tests/test_api_targets.py b/plugins/nemo-auditor/tests/test_api_targets.py index db2919404f..7fd227ffcf 100644 --- a/plugins/nemo-auditor/tests/test_api_targets.py +++ b/plugins/nemo-auditor/tests/test_api_targets.py @@ -5,6 +5,7 @@ from __future__ import annotations +import logging from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock @@ -144,6 +145,41 @@ def test_404_when_missing(self, client, mock_entity_client) -> None: ) assert resp.status_code == 404 + def test_conflict_hides_raw_exception_details(self, client, mock_entity_client) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_target("tgt-1")) + mock_entity_client.update = AsyncMock( + side_effect=NemoEntityConflictError("Error code: 409 - {'detail': 'db_version mismatch'}") + ) + + resp = client.put( + "/apis/auditor/v2/workspaces/default/targets/tgt-1", + json={"type": "nim", "model": "x"}, + ) + + assert resp.status_code == 409 + detail = resp.json()["detail"] + assert "AuditTarget 'tgt-1'" in detail + assert "Refresh the target" in detail + assert "Error code" not in detail + assert "db_version" not in detail + + def test_conflict_sanitizes_log_fields(self, client, mock_entity_client, caplog) -> None: + mock_entity_client.get = AsyncMock(return_value=_make_target("tgt-1")) + mock_entity_client.update = AsyncMock(side_effect=NemoEntityConflictError("conflict")) + + with caplog.at_level(logging.INFO, logger=targets_router_module.__name__): + resp = client.put( + "/apis/auditor/v2/workspaces/default%0Aforged/targets/tgt-1%0D%0Aforged", + json={"type": "nim", "model": "x"}, + ) + + assert resp.status_code == 409 + message = next( + record.getMessage() for record in caplog.records if "Conflict updating audit target" in record.msg + ) + assert "\r" not in message + assert "\n" not in message + class TestDeleteTarget: def test_returns_204(self, client, mock_entity_client) -> None: diff --git a/services/core/jobs/src/nmp/core/jobs/api/v2/jobs/endpoints.py b/services/core/jobs/src/nmp/core/jobs/api/v2/jobs/endpoints.py index 92f3c0c233..65f23c9b31 100644 --- a/services/core/jobs/src/nmp/core/jobs/api/v2/jobs/endpoints.py +++ b/services/core/jobs/src/nmp/core/jobs/api/v2/jobs/endpoints.py @@ -5,10 +5,12 @@ import logging import math -from typing import Optional +import re +from typing import Any, Optional from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, status from nemo_platform import AsyncNeMoPlatform +from nemo_platform_plugin.log_utils import sanitize_for_log from nmp.common.api.common import Page, PaginationData from nmp.common.api.parsed_filter import ParsedFilter, make_filter_dep from nmp.common.api.utils import generate_openapi_extra_params, parse_deep_object @@ -25,6 +27,7 @@ PlatformJobLogPage, PlatformJobResultCreateRequest, PlatformJobResultResponse, + PlatformJobStatus, PlatformJobStatusResponse, ) from nmp.common.observability import scoped_app_ctx @@ -45,7 +48,12 @@ PlatformJobTaskUpdate, ) from nmp.core.jobs.app.ctx import JobContext -from nmp.core.jobs.app.dispatcher import JobDispatcher, StateTransitionConflictError +from nmp.core.jobs.app.dispatcher import ( + JobAlreadyExistsError, + JobDispatcher, + JobSecretValidationError, + StateTransitionConflictError, +) from nmp.core.jobs.app.profiles import ExecutionProfileT from nmp.core.jobs.app.providers import CPUExecutionProvider, SubprocessExecutionProvider from nmp.core.jobs.app.schemas import ( @@ -61,6 +69,98 @@ router = APIRouter() platform_config = get_platform_config() +_JOB_STATUS_VALUES = ", ".join(job_status.value for job_status in PlatformJobStatus) + + +def _format_validation_location(loc: Any, *, prefix: str | None = None) -> str: + if isinstance(loc, (list, tuple)): + parts = [str(part) for part in loc if part not in ("body", "query")] + elif loc: + parts = [str(loc)] + else: + parts = [] + if prefix: + parts.insert(0, prefix) + return ".".join(parts) if parts else (prefix or "request") + + +def _safe_validation_message(message: object) -> str: + if not isinstance(message, str): + return "has an invalid value" + + normalized = message.lower() + if "required" in normalized or "missing" in normalized: + return "is required" + if "integer" in normalized: + return "must be an integer" + if "number" in normalized or "float" in normalized: + return "must be a number" + if "boolean" in normalized: + return "must be a boolean" + if "string" in normalized: + return "must be a string" + if "list" in normalized or "array" in normalized: + return "must be a list" + if "dictionary" in normalized or "object" in normalized: + return "must be an object" + return "has an invalid value" + + +def _format_pydantic_validation_error(exc: ValidationError, *, prefix: str | None = None) -> str: + messages: list[str] = [] + for error in exc.errors(): + location = _format_validation_location(error.get("loc"), prefix=prefix) + message = error.get("msg") or "Invalid value." + if location.endswith("status") or ".status." in location: + message = f"{message} Expected one or more of: {_JOB_STATUS_VALUES}." + messages.append(f"{location}: {message}") + if not messages: + return "The request contains invalid values. Update the request and try again." + return f"The request contains invalid values. {'; '.join(messages)}." + + +def _format_entity_validation_error(exc: EntityValidationError, *, resource: str) -> str: + detail = str(exc.args[0]) if exc.args else "" + field_match = re.search(r"field ['\"](?P[A-Za-z_][A-Za-z0-9_.\[\]-]*)['\"](?P.*)", detail) + if field_match: + field = field_match.group("field") + message = _safe_validation_message(field_match.group("message")) + return f"Invalid {resource} request. Field '{field}' {message}. Update the request and try again." + + return ( + f"Invalid {resource} request. Check that the name, project, platform_spec, and referenced resources are " + "valid for this workspace, then try again." + ) + + +def _format_job_compilation_error(exc: PlatformJobCompilationError) -> str: + detail = str(exc).lower() + if "gpu" in detail and "docker" in detail: + return ( + "Invalid job specification: this job requires GPU resources, but the platform is running on Docker with " + "no GPUs configured. Configure GPUs in the platform config or choose a non-GPU execution profile." + ) + return ( + "Invalid job specification. Check platform_spec.steps, executor profiles, and resource requirements, " + "then try again." + ) + + +def _format_create_job_conflict(exc: ValueError, *, job_name: str | None, workspace: str) -> str: + if isinstance(exc, JobAlreadyExistsError) and job_name: + return ( + f"Job '{job_name}' already exists in workspace '{workspace}'. Choose a different job name or delete the " + "existing job before creating a new one." + ) + if isinstance(exc, JobSecretValidationError): + return ( + "Unable to create job because one or more referenced secrets were not found or are not accessible. " + "Verify each platform_spec secret reference uses the expected workspace/name and that you have access." + ) + return ( + "Unable to create job because the request conflicts with existing platform state. Review the job name and " + "referenced resources, then try again." + ) def get_platform_jobs_steps_list_filter(request: Request) -> PlatformJobStepsListFilter: @@ -68,8 +168,12 @@ def get_platform_jobs_steps_list_filter(request: Request) -> PlatformJobStepsLis try: filters = parse_deep_object(name="filter", params=request.query_params) or {} return PlatformJobStepsListFilter(**filters) - except ValidationError as e: - raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(e)) + except ValidationError as exc: + logger.info("Invalid job steps filter parameters", exc_info=True) + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=_format_pydantic_validation_error(exc, prefix="filter"), + ) from exc def validate_job_spec( @@ -96,11 +200,12 @@ def validate_job_spec( ) try: validate_gpu_available_for_docker(job_spec.model_dump()) - except PlatformJobCompilationError as e: + except PlatformJobCompilationError as exc: + logger.info("Invalid platform job specification", exc_info=True) raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, - detail=str(e), - ) from e + detail=_format_job_compilation_error(exc), + ) from exc def translate_cpu_container_steps_to_subprocess( @@ -163,10 +268,23 @@ async def create_job( auth_context=AuthContext.from_principal(auth_client.principal), sdk=sdk, ) - except ValueError as e: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"Unable to create job: {str(e)}") - except EntityValidationError as e: - raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(e)) + except ValueError as exc: + logger.info( + "Failed to create job '%s' in workspace '%s'", + sanitize_for_log(request.name), + sanitize_for_log(workspace), + exc_info=True, + ) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=_format_create_job_conflict(exc, job_name=request.name, workspace=workspace), + ) from exc + except EntityValidationError as exc: + logger.info("Invalid job entity for workspace '%s'", sanitize_for_log(workspace), exc_info=True) + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=_format_entity_validation_error(exc, resource="job"), + ) from exc @router.get( @@ -226,7 +344,10 @@ async def get_job( with scoped_app_ctx(JobContext(id=name)): job = await dispatcher.get_job(name, workspace) if not job: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) return job @@ -246,10 +367,22 @@ async def cancel_job( with scoped_app_ctx(JobContext(id=name)): try: job = await dispatcher.cancel_job(name, workspace) - except StateTransitionConflictError as e: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) + except StateTransitionConflictError as exc: + logger.info( + "Cannot cancel job '%s' in workspace '%s'", + sanitize_for_log(name), + sanitize_for_log(workspace), + exc_info=True, + ) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=(f"Cannot cancel job '{name}' from its current state. Refresh the job status and try again."), + ) from exc if not job: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) return job @@ -269,7 +402,10 @@ async def pause_job( with scoped_app_ctx(JobContext(id=name)): job = await dispatcher.pause_job(name, workspace) if not job: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) return job @@ -289,7 +425,10 @@ async def resume_job( with scoped_app_ctx(JobContext(id=name)): job = await dispatcher.resume_job(name, workspace) if not job: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) return job @@ -310,7 +449,10 @@ async def delete_job( with scoped_app_ctx(JobContext(id=name)): deleted = await dispatcher.delete_job(name, workspace) if not deleted: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) @router.get( @@ -330,7 +472,10 @@ async def get_job_status( with scoped_app_ctx(JobContext(id=name)): job_status = await dispatcher.get_job_status(name, workspace) if not job_status: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) return job_status @@ -352,7 +497,10 @@ async def update_job_status_details( with scoped_app_ctx(JobContext(id=name)): result = await dispatcher.update_job_status_details(name, workspace, request) if not result: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) @router.get( @@ -378,7 +526,10 @@ async def page_job_logs( with scoped_app_ctx(JobContext(id=name)): job = await dispatcher.get_job(name, workspace) if not job: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) try: filters = { @@ -420,7 +571,10 @@ async def create_job_result( with scoped_app_ctx(JobContext(id=job, result_name=name)): job_entity = await dispatcher.get_job(job, workspace) if not job_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{job}' not found in workspace '{workspace}'.", + ) result = await dispatcher.create_result( job_id=job_entity.id, result_name=name, @@ -452,7 +606,10 @@ async def list_job_results( with scoped_app_ctx(JobContext(id=name)): job_entity = await dispatcher.get_job(name, workspace) if not job_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{name}' not found in workspace '{workspace}'.", + ) results, _ = await dispatcher.list_results(job_id=job_entity.id, workspace=workspace, sort=sort) return PlatformJobListResultResponse(data=[r.to_response() for r in results]) @@ -476,7 +633,10 @@ async def get_job_result( with scoped_app_ctx(JobContext(id=job, result_name=name)): result = await dispatcher.get_result(job, name, workspace) if not result: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job result not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Result '{name}' for job '{job}' not found in workspace '{workspace}'.", + ) return result.to_response() @@ -505,7 +665,10 @@ async def download_job_result( with scoped_app_ctx(JobContext(id=job, result_name=name)): result = await dispatcher.get_result(job, name, workspace) if not result: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job result not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Result '{name}' for job '{job}' not found in workspace '{workspace}'.", + ) filename, tmp_dir_path = await download_from_result_info( result_name=name, @@ -583,7 +746,10 @@ async def get_job_step( with scoped_app_ctx(JobContext(id=job, step_name=name)): step = await dispatcher.get_current_job_step_by_name(job, name, workspace) if not step: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job step not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Step '{name}' for job '{job}' not found in workspace '{workspace}'.", + ) return step @@ -607,7 +773,10 @@ async def update_job_step_status( with scoped_app_ctx(JobContext(id=job, step_name=name)): step_entity = await dispatcher.get_current_job_step_by_name(job, name, workspace) if not step_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job step not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Step '{name}' for job '{job}' not found in workspace '{workspace}'.", + ) try: step_entity, _ = await dispatcher.update_job_status_from_step( @@ -616,13 +785,32 @@ async def update_job_step_status( status_details=request.status_details, error_details=request.error_details, ) - except StateTransitionConflictError as e: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) - except EntityConflictError as e: + except StateTransitionConflictError as exc: + logger.info( + "Cannot update job step '%s' for job '%s' to status '%s'", + sanitize_for_log(name), + sanitize_for_log(job), + sanitize_for_log(request.status), + exc_info=True, + ) raise HTTPException( status_code=status.HTTP_409_CONFLICT, - detail=f"Conflict updating job step (entity was modified by another request): {e}", + detail=( + f"Cannot update job step '{name}' to status '{request.status}' from its current state. " + "Refresh the step status and try again." + ), + ) from exc + except EntityConflictError as exc: + logger.info( + "Conflict updating job step '%s' for job '%s'", + sanitize_for_log(name), + sanitize_for_log(job), + exc_info=True, ) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Conflict updating job step: it was modified by another request. Refresh the step and retry.", + ) from exc return step_entity @@ -646,11 +834,17 @@ async def list_job_step_tasks( with scoped_app_ctx(JobContext(id=job, step_name=name)): job_entity = await dispatcher.get_job(job, workspace) if not job_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Job '{job}' not found in workspace '{workspace}'.", + ) step_entity = await dispatcher.get_current_job_step_by_name(job, name, workspace) if not step_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job step not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Step '{name}' for job '{job}' not found in workspace '{workspace}'.", + ) return PlatformJobListTaskResponse(data=await dispatcher.list_tasks(step_entity.id, workspace=workspace)) @@ -675,7 +869,10 @@ async def update_job_step_task( with scoped_app_ctx(JobContext(id=job, step_name=step)): step_entity = await dispatcher.get_current_job_step_by_name(job, step, workspace) if not step_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job step not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Step '{step}' for job '{job}' not found in workspace '{workspace}'.", + ) return await dispatcher.create_or_update_task( job, @@ -705,9 +902,15 @@ async def get_job_step_task( with scoped_app_ctx(JobContext(id=job, step_name=step, task_id=name)): step_entity = await dispatcher.get_current_job_step_by_name(job, step, workspace) if not step_entity: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job step not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Step '{step}' for job '{job}' not found in workspace '{workspace}'.", + ) task = await dispatcher.get_task(step_entity.id, name, workspace) if not task: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job step task not found") + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=(f"Task '{name}' for step '{step}' of job '{job}' not found in workspace '{workspace}'."), + ) return task diff --git a/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py b/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py index fd12f0f918..591f166ac5 100644 --- a/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py +++ b/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py @@ -57,6 +57,14 @@ class StateTransitionConflictError(Exception): """Exception raised when a state transition is invalid.""" +class JobAlreadyExistsError(ValueError): + """Exception raised when creating a job whose name is already in use.""" + + +class JobSecretValidationError(ValueError): + """Exception raised when a job's secret references cannot be validated.""" + + operations_counter = create_counter( meter=meter, subsystem="jobs", @@ -223,15 +231,19 @@ async def validate_job_secrets( secrets = client_from_platform(sdk_to_use, AsyncSecretsClient) try: await secrets.get_secret(name=secret_name, workspace=workspace) - except ClientNotFoundError: - raise ValueError(f"Secret '{workspace}/{secret_name}' not found.") - except ClientPermissionDeniedError: - raise ValueError(f"User does not have access to secret '{workspace}/{secret_name}'.") - except Exception: + except ClientNotFoundError as exc: + raise JobSecretValidationError(f"Secret '{workspace}/{secret_name}' not found.") from exc + except ClientPermissionDeniedError as exc: + raise JobSecretValidationError( + f"User does not have access to secret '{workspace}/{secret_name}'." + ) from exc + except Exception as exc: logger.exception( "Error validating secret", extra={"secret_name": secret_name, "workspace": workspace} ) - raise ValueError(f"Unknown error when validating secret '{workspace}/{secret_name}'.") + raise JobSecretValidationError( + f"Unknown error when validating secret '{workspace}/{secret_name}'." + ) from exc async def create_job( self, @@ -247,7 +259,9 @@ async def create_job( try: existing_job = await self.store.get(PlatformJob, job_name, workspace=workspace) if existing_job: - raise ValueError(f"Job with name '{job_name}' already exists in workspace '{workspace}'.") + raise JobAlreadyExistsError( + f"Job with name '{job_name}' already exists in workspace '{workspace}'." + ) except EntityNotFoundError: pass # Job does not exist, proceed to create diff --git a/services/core/jobs/tests/api/test_cancel_rerun.py b/services/core/jobs/tests/api/test_cancel_rerun.py index d0369b92ef..8da7041cf0 100644 --- a/services/core/jobs/tests/api/test_cancel_rerun.py +++ b/services/core/jobs/tests/api/test_cancel_rerun.py @@ -248,7 +248,7 @@ async def test_job_cancel_nonexistent_job(test_client: AsyncClient): response = await test_client.post("/apis/jobs/v2/workspaces/default/jobs/nonexistent-job-id/cancel") assert response.status_code == 404 error_data = response.json() - assert error_data["detail"] == "Job not found" + assert error_data["detail"] == "Job 'nonexistent-job-id' not found in workspace 'default'." @pytest.mark.asyncio diff --git a/services/core/jobs/tests/api/test_pause_resume.py b/services/core/jobs/tests/api/test_pause_resume.py index 47df833493..a8f1e61b01 100644 --- a/services/core/jobs/tests/api/test_pause_resume.py +++ b/services/core/jobs/tests/api/test_pause_resume.py @@ -200,7 +200,7 @@ async def test_job_pause_nonexistent_job(test_client: AsyncClient): response = await test_client.post("/apis/jobs/v2/workspaces/default/jobs/nonexistent-job-id/pause") assert response.status_code == 404 error_data = response.json() - assert error_data["detail"] == "Job not found" + assert error_data["detail"] == "Job 'nonexistent-job-id' not found in workspace 'default'." @pytest.mark.asyncio @@ -209,7 +209,7 @@ async def test_job_resume_nonexistent_job(test_client: AsyncClient): response = await test_client.post("/apis/jobs/v2/workspaces/default/jobs/nonexistent-job-id/resume") assert response.status_code == 404 error_data = response.json() - assert error_data["detail"] == "Job not found" + assert error_data["detail"] == "Job 'nonexistent-job-id' not found in workspace 'default'." @pytest.mark.asyncio diff --git a/services/core/jobs/tests/test_job_logs.py b/services/core/jobs/tests/test_job_logs.py index 4d3c752df2..d77745492c 100644 --- a/services/core/jobs/tests/test_job_logs.py +++ b/services/core/jobs/tests/test_job_logs.py @@ -148,7 +148,7 @@ async def test_get_job_logs_job_not_found(self, test_client, dispatcher, mock_lo response = test_client.get(f"/v2/workspaces/{DEFAULT_WORKSPACE}/jobs/nonexistent-job/logs") assert response.status_code == 404 - assert response.json()["detail"] == "Job not found" + assert response.json()["detail"] == "Job 'nonexistent-job' not found in workspace 'default'." mock_logs_client.query_logs.assert_not_called() async def test_get_job_logs_invalid_page_cursor( diff --git a/services/core/jobs/tests/test_jobs_api.py b/services/core/jobs/tests/test_jobs_api.py index 0ed059d7c6..0a257140c5 100644 --- a/services/core/jobs/tests/test_jobs_api.py +++ b/services/core/jobs/tests/test_jobs_api.py @@ -2,20 +2,24 @@ # SPDX-License-Identifier: Apache-2.0 import asyncio +import logging import tarfile from datetime import datetime from io import BytesIO from pathlib import Path from typing import Any, Dict -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import HTTPException from httpx import AsyncClient from nemo_platform import AsyncNeMoPlatform from nmp.common.entities import ALL_WORKSPACES, DEFAULT_WORKSPACE +from nmp.common.entities.client import EntityValidationError from nmp.common.jobs.schemas import PlatformJobStatus from nmp.core.jobs.api.v2.jobs.endpoints import ( + _format_create_job_conflict, + _format_entity_validation_error, get_platform_jobs_steps_list_filter, ) from nmp.core.jobs.api.v2.jobs.schemas import ( @@ -24,7 +28,12 @@ PlatformJobSortField, PlatformJobStepsListFilter, ) -from nmp.core.jobs.app.dispatcher import JobDispatcher +from nmp.core.jobs.app.dispatcher import ( + JobAlreadyExistsError, + JobDispatcher, + JobSecretValidationError, + StateTransitionConflictError, +) from nmp.core.jobs.app.providers import ContainerSpec, GPUExecutionProvider, SubprocessExecutionProvider from nmp.core.jobs.app.schemas import ( PlatformJobSpec, @@ -233,6 +242,130 @@ async def test_create_job_with_invalid_project_name(test_client: AsyncClient): ) +@pytest.mark.asyncio +async def test_create_job_validation_error_returns_actionable_message( + test_client: AsyncClient, + mock_dispatcher: JobDispatcher, +): + raw_error = "EntityValidationError: field 'name' is required; internal=db_version" + with patch.object(mock_dispatcher, "create_job", new=AsyncMock(side_effect=EntityValidationError(raw_error))): + response = await test_client.post( + "/apis/jobs/v2/workspaces/default/jobs", + json={ + "name": "bad-job", + "source": "test-source", + "spec": {}, + "platform_spec": { + "steps": [ + { + "name": "valid-step", + "executor": TestConstants.TEST_EXECUTOR.model_dump(mode="json"), + "config": {}, + } + ] + }, + }, + ) + + assert response.status_code == 422 + detail = response.json()["detail"] + assert "Field 'name' is required" in detail + assert "EntityValidationError" not in detail + assert "db_version" not in detail + + +def test_format_entity_validation_error_unknown_type_message(): + raw_error = "field 'required_string_count' has an invalid value" + detail = _format_entity_validation_error(EntityValidationError(raw_error), resource="job") + + assert detail == ( + "Invalid job request. Field 'required_string_count' has an invalid value. Update the request and try again." + ) + + +@pytest.mark.asyncio +async def test_create_job_conflict_hides_raw_exception_message( + test_client: AsyncClient, + mock_dispatcher: JobDispatcher, +): + raw_error = "ValueError: Job with name 'dup-job' already exists in workspace 'default'. internal=db_version" + with patch.object(mock_dispatcher, "create_job", new=AsyncMock(side_effect=JobAlreadyExistsError(raw_error))): + response = await test_client.post( + "/apis/jobs/v2/workspaces/default/jobs", + json={ + "name": "dup-job", + "source": "test-source", + "spec": {}, + "platform_spec": { + "steps": [ + { + "name": "valid-step", + "executor": TestConstants.TEST_EXECUTOR.model_dump(mode="json"), + "config": {}, + } + ] + }, + }, + ) + + assert response.status_code == 409 + detail = response.json()["detail"] + assert "Job 'dup-job' already exists in workspace 'default'" in detail + assert "ValueError" not in detail + assert "db_version" not in detail + + +def test_create_job_conflict_does_not_classify_generic_message(): + raw_error = "unrelated secret already exists; internal=db_version" + detail = _format_create_job_conflict(ValueError(raw_error), job_name="test-job", workspace="default") + + assert "conflicts with existing platform state" in detail + assert "already exists" not in detail + assert "referenced secrets" not in detail + assert "db_version" not in detail + + +def test_create_job_secret_validation_error_returns_safe_guidance(): + raw_error = "Secret 'default/api-key' not found; internal=db_version" + detail = _format_create_job_conflict(JobSecretValidationError(raw_error), job_name="test-job", workspace="default") + + assert "referenced secrets were not found or are not accessible" in detail + assert "api-key" not in detail + assert "db_version" not in detail + + +@pytest.mark.asyncio +async def test_create_job_conflict_sanitizes_log_fields( + test_client: AsyncClient, + mock_dispatcher: JobDispatcher, + caplog, +): + with patch.object(mock_dispatcher, "create_job", new=AsyncMock(side_effect=ValueError("conflict"))): + with caplog.at_level(logging.INFO, logger="nmp.core.jobs.api.v2.jobs.endpoints"): + response = await test_client.post( + "/apis/jobs/v2/workspaces/default%0Aforged/jobs", + json={ + "name": "dup-job\r\nforged", + "source": "test-source", + "spec": {}, + "platform_spec": { + "steps": [ + { + "name": "valid-step", + "executor": TestConstants.TEST_EXECUTOR.model_dump(mode="json"), + "config": {}, + } + ] + }, + }, + ) + + assert response.status_code == 409 + message = next(record.getMessage() for record in caplog.records if "Failed to create job" in record.msg) + assert "\r" not in message + assert "\n" not in message + + @pytest.mark.asyncio async def test_create_job_gpu_fail_fast_when_docker_no_gpus(test_client: AsyncClient): """Direct Jobs API create with GPU step fails fast with 422 when platform is Docker with no GPUs.""" @@ -1143,8 +1276,84 @@ async def test_get_platform_jobs_steps_list_filter(query_string, expected_checks async def test_get_platform_jobs_steps_list_filter_invalid(): """Test that invalid status raises an error.""" request = MockRequest("filter[status]=INVALID_STATUS") - with pytest.raises(HTTPException): + with pytest.raises(HTTPException) as exc_info: get_platform_jobs_steps_list_filter(request) # type: ignore[arg-type] + assert exc_info.value.detail is not None + assert "filter.status" in str(exc_info.value.detail) + assert "Expected one or more of" in str(exc_info.value.detail) + assert "ValidationError" not in str(exc_info.value.detail) + + +@pytest.mark.asyncio +async def test_cancel_job_conflict_hides_internal_transition( + test_client: AsyncClient, + mock_dispatcher: JobDispatcher, +): + raw_error = ( + "Invalid status transition from PlatformJobStatus.RESUMING to PlatformJobStatus.CANCELLING for step step-id-123" + ) + with patch.object( + mock_dispatcher, "cancel_job", new=AsyncMock(side_effect=StateTransitionConflictError(raw_error)) + ): + response = await test_client.post("/apis/jobs/v2/workspaces/default/jobs/conflicted-job/cancel") + + assert response.status_code == 409 + detail = response.json()["detail"] + assert "Cannot cancel job 'conflicted-job'" in detail + assert "PlatformJobStatus" not in detail + assert "step-id-123" not in detail + + +@pytest.mark.asyncio +async def test_cancel_job_conflict_sanitizes_log_fields( + test_client: AsyncClient, + mock_dispatcher: JobDispatcher, + caplog, +): + with patch.object( + mock_dispatcher, + "cancel_job", + new=AsyncMock(side_effect=StateTransitionConflictError("invalid transition")), + ): + with caplog.at_level(logging.INFO, logger="nmp.core.jobs.api.v2.jobs.endpoints"): + response = await test_client.post( + "/apis/jobs/v2/workspaces/default%0Aforged/jobs/conflicted-job%0D%0Aforged/cancel" + ) + + assert response.status_code == 409 + message = next(record.getMessage() for record in caplog.records if "Cannot cancel job" in record.msg) + assert "\r" not in message + assert "\n" not in message + + +@pytest.mark.asyncio +async def test_update_job_step_conflict_sanitizes_log_fields( + test_client: AsyncClient, + mock_dispatcher: JobDispatcher, + caplog, +): + with ( + patch.object( + mock_dispatcher, + "get_current_job_step_by_name", + new=AsyncMock(return_value=MagicMock()), + ), + patch.object( + mock_dispatcher, + "update_job_status_from_step", + new=AsyncMock(side_effect=StateTransitionConflictError("invalid transition")), + ), + caplog.at_level(logging.INFO, logger="nmp.core.jobs.api.v2.jobs.endpoints"), + ): + response = await test_client.patch( + "/apis/jobs/v2/workspaces/default/jobs/job%0Aforged/steps/step%0D%0Aforged/status", + json={"status": "pending"}, + ) + + assert response.status_code == 409 + message = next(record.getMessage() for record in caplog.records if "Cannot update job step" in record.msg) + assert "\r" not in message + assert "\n" not in message @pytest.mark.asyncio