diff --git a/docs/cli/reference.mdx b/docs/cli/reference.mdx index 4024707eab..ea6696b9b1 100644 --- a/docs/cli/reference.mdx +++ b/docs/cli/reference.mdx @@ -1419,11 +1419,12 @@ Example: metadata = FilesetMetadata( dataset=DatasetMetadataContent( schema=`{"c Delete Fileset. -Permanently deletes a fileset from the platform. +Permanently deletes an unreferenced fileset from the platform. -Returns metadata about the -deleted fileset. For local storage backends, this also deletes the underlying -files. +Referencing model +or adapter entities must be relinked or deleted first. Returns metadata about +the deleted fileset. For local storage backends, this also deletes the +underlying files. **Usage:** @@ -1667,6 +1668,7 @@ nemo files otlp logs query [OPTIONS] NAME * `--filters`: Key-value filters to apply to the query * `--limit `: Maximum number of results to return * `--page-cursor`: Cursor for pagination +* `--tail `: Number of newest log lines to return **Help:** @@ -3926,6 +3928,7 @@ nemo jobs [OPTIONS] COMMAND [ARGS]... * `get`: Get a platform job by name. * `update-status-details`: Update the status details of a platform job. * `watch`: Watch a platform job until it reaches a terminal status. +* `tail`: Print the newest lines from a platform job log. * `results`: Manage results * `steps`: Manage steps * `tasks`: Manage tasks @@ -4056,6 +4059,7 @@ nemo jobs get-logs [OPTIONS] NAME * `--limit `: Maximum number of logs to return * `--page-cursor`: Page cursor * `--step-id`: Filter logs by step name +* `--tail `: Number of newest log lines to return * `--task-id`: Filter logs by task ID * `--all-pages`: Fetch all pages @@ -4315,6 +4319,32 @@ nemo jobs watch [OPTIONS] NAME * `--help, -h`: Show this message and exit. +#### nemo jobs tail + +Print the newest lines from a platform job log. + +**Usage:** + +```shell +nemo jobs tail [OPTIONS] NAME +``` + +**Arguments:** + +* ``: Name of the platform job whose logs to tail + +**Options:** + +* `-n, --lines `: Number of lines to show [default: 100] +* `--workspace`: Workspace containing the job +* `--attempt-id `: Filter logs to an attempt ID +* `--step-id`: Filter logs to a step ID +* `--task-id`: Filter logs to a task ID + +**Help:** + +* `--help, -h`: Show this message and exit. + #### nemo jobs results Manage results diff --git a/docs/set-up/manage-jobs.mdx b/docs/set-up/manage-jobs.mdx index 9240459275..f954fa78c5 100644 --- a/docs/set-up/manage-jobs.mdx +++ b/docs/set-up/manage-jobs.mdx @@ -9,6 +9,22 @@ description: "" This section describes how to configure jobs in NeMo Platform. The Jobs service is responsible for scheduling batch jobs, collecting telemetry and managing job results. +## Viewing Job Logs + +Use `nemo jobs get-logs` to inspect paginated job logs: + +```bash +nemo jobs get-logs --workspace default --limit 100 +``` + +Use `nemo jobs tail` when you only need the newest lines from a long or running job: + +```bash +nemo jobs tail --workspace default -n 500 +``` + +Both commands support `--attempt-id`, `--step-id`, and `--task-id` to narrow the log output. + ## Execution Profiles You can configure jobs to run on different hardware by specifying an execution profile. This allows you to define different compute resources and job execution backends for different job types and use cases. diff --git a/openapi/ga/individual/platform.openapi.yaml b/openapi/ga/individual/platform.openapi.yaml index 66f24c9bdf..e127df2435 100644 --- a/openapi/ga/individual/platform.openapi.yaml +++ b/openapi/ga/individual/platform.openapi.yaml @@ -5827,20 +5827,31 @@ paths: in: query required: false schema: - type: integer - exclusiveMinimum: 0 description: Maximum number of logs to return - default: 100 title: Limit + default: 100 + type: integer + maximum: 10000 + exclusiveMinimum: 0 description: Maximum number of logs to return - name: page_cursor in: query required: false schema: - type: string description: Page cursor title: Page Cursor + type: string description: Page cursor + - name: tail + in: query + required: false + schema: + description: Number of newest log lines to return + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 + description: Number of newest log lines to return - name: attempt_id in: query required: false @@ -14650,16 +14661,22 @@ components: title: Filters description: Key-value filters to apply to the query limit: - type: integer - maximum: 1000.0 - exclusiveMinimum: 0.0 title: Limit description: Maximum number of results to return default: 100 + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 page_cursor: title: Page Cursor description: Cursor for pagination type: string + tail: + title: Tail + description: Number of newest log lines to return + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 artifact_base_path: title: Artifact Base Path description: Folder inside the fileset the logs were nested under (must diff --git a/openapi/ga/openapi.yaml b/openapi/ga/openapi.yaml index 66f24c9bdf..e127df2435 100644 --- a/openapi/ga/openapi.yaml +++ b/openapi/ga/openapi.yaml @@ -5827,20 +5827,31 @@ paths: in: query required: false schema: - type: integer - exclusiveMinimum: 0 description: Maximum number of logs to return - default: 100 title: Limit + default: 100 + type: integer + maximum: 10000 + exclusiveMinimum: 0 description: Maximum number of logs to return - name: page_cursor in: query required: false schema: - type: string description: Page cursor title: Page Cursor + type: string description: Page cursor + - name: tail + in: query + required: false + schema: + description: Number of newest log lines to return + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 + description: Number of newest log lines to return - name: attempt_id in: query required: false @@ -14650,16 +14661,22 @@ components: title: Filters description: Key-value filters to apply to the query limit: - type: integer - maximum: 1000.0 - exclusiveMinimum: 0.0 title: Limit description: Maximum number of results to return default: 100 + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 page_cursor: title: Page Cursor description: Cursor for pagination type: string + tail: + title: Tail + description: Number of newest log lines to return + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 artifact_base_path: title: Artifact Base Path description: Folder inside the fileset the logs were nested under (must diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 66f24c9bdf..e127df2435 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -5827,20 +5827,31 @@ paths: in: query required: false schema: - type: integer - exclusiveMinimum: 0 description: Maximum number of logs to return - default: 100 title: Limit + default: 100 + type: integer + maximum: 10000 + exclusiveMinimum: 0 description: Maximum number of logs to return - name: page_cursor in: query required: false schema: - type: string description: Page cursor title: Page Cursor + type: string description: Page cursor + - name: tail + in: query + required: false + schema: + description: Number of newest log lines to return + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 + description: Number of newest log lines to return - name: attempt_id in: query required: false @@ -14650,16 +14661,22 @@ components: title: Filters description: Key-value filters to apply to the query limit: - type: integer - maximum: 1000.0 - exclusiveMinimum: 0.0 title: Limit description: Maximum number of results to return default: 100 + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 page_cursor: title: Page Cursor description: Cursor for pagination type: string + tail: + title: Tail + description: Number of newest log lines to return + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 artifact_base_path: title: Artifact Base Path description: Folder inside the fileset the logs were nested under (must diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/filesets.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/filesets.py index efdc346ee4..1b01488833 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/filesets.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/filesets.py @@ -167,11 +167,12 @@ def delete_filesets( ) -> None: """Delete Fileset. - Permanently deletes a fileset from the platform. + Permanently deletes an unreferenced fileset from the platform. - Returns metadata about the - deleted fileset. For local storage backends, this also deletes the underlying - files.""" + Referencing model + or adapter entities must be relinked or deleted first. Returns metadata about + the deleted fileset. For local storage backends, this also deletes the + underlying files.""" state: CLIContext = ctx.obj client = state.get_client() diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/otlp/logs.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/otlp/logs.py index 8c99eccb25..85d85d5556 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/otlp/logs.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/files/otlp/logs.py @@ -98,6 +98,7 @@ def query_logs( filters: Annotated[str | None, typer.Option("--filters", help="Key-value filters to apply to the query")] = None, limit: Annotated[int | None, typer.Option("--limit", help="Maximum number of results to return")] = None, page_cursor: Annotated[str | None, typer.Option("--page-cursor", help="Cursor for pagination")] = None, + tail: Annotated[int | None, typer.Option("--tail", help="Number of newest log lines to return")] = None, output_format: EntityOutputFormatOption = None, ) -> None: """Query logs from parquet files in a fileset. @@ -113,6 +114,7 @@ def query_logs( filters=filters, limit=limit, page_cursor=page_cursor, + tail=tail, ) if handle_code_generation(["files", "otlp", "logs"], "query", kwargs, output_format, state): return diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/jobs/__init__.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/jobs/__init__.py index da66c40f00..21ae14cfef 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/jobs/__init__.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/jobs/__init__.py @@ -10,6 +10,7 @@ import typer from nemo_platform_plugin.client.adapter import client_from_platform from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import JobLogsQueryParams from nemo_platform_ext.cli.core.api import build_kwargs, merge_filter_dict from nemo_platform_ext.cli.core.code_generator import handle_code_generation @@ -22,6 +23,7 @@ validate_stream_output_format, ) from nemo_platform_ext.cli.core.help_formatter import collect_warnings, create_typer_app +from nemo_platform_ext.cli.core.job_log_renderer import render_job_logs from nemo_platform_ext.cli.core.job_watch_renderer import JobWatchRenderResult, render_job_watch_events from nemo_platform_ext.cli.core.pagination import PaginationType, fetch_all_pages, warn_if_more_pages from nemo_platform_ext.cli.core.stdin_utils import read_data_input_with_flags, read_payload, validate_required_fields @@ -275,6 +277,7 @@ def get_logs_jobs( limit: Annotated[int | None, typer.Option("--limit", help="Maximum number of logs to return")] = None, page_cursor: Annotated[str | None, typer.Option("--page-cursor", help="Page cursor")] = None, step_id: Annotated[str | None, typer.Option("--step-id", help="Filter logs by step name")] = None, + tail: Annotated[int | None, typer.Option("--tail", help="Number of newest log lines to return")] = None, task_id: Annotated[str | None, typer.Option("--task-id", help="Filter logs by task ID")] = None, output_format: ListOutputFormatOption = None, no_truncate: NoTruncateOption = None, @@ -302,6 +305,7 @@ def get_logs_jobs( limit=limit, page_cursor=page_cursor, step_id=step_id, + tail=tail, task_id=task_id, ) @@ -720,3 +724,34 @@ def watch_platform_job( raise typer.Exit(130) if watch_result is not JobWatchRenderResult.SUCCEEDED: raise typer.Exit(1) + + +@app.command("tail") +@collect_warnings +@handle_errors +def tail_platform_job( + ctx: typer.Context, + name: Annotated[str, typer.Argument(help="Name of the platform job whose logs to tail")], + lines: Annotated[int, typer.Option("-n", "--lines", min=1, max=10_000, help="Number of lines to show")] = 100, + workspace: Annotated[str | None, typer.Option("--workspace", help="Workspace containing the job")] = None, + attempt_id: Annotated[int | None, typer.Option("--attempt-id", help="Filter logs to an attempt ID")] = None, + step_id: Annotated[str | None, typer.Option("--step-id", help="Filter logs to a step ID")] = None, + task_id: Annotated[str | None, typer.Option("--task-id", help="Filter logs to a task ID")] = None, +) -> None: + """Print the newest lines from a platform job log.""" + state: CLIContext = ctx.obj + client = state.get_client() + jobs_client = client_from_platform(client, JobsClient) + if workspace is None: + workspace = client._get_workspace_path_param() + + query_params: JobLogsQueryParams = {"tail": lines} + if attempt_id is not None: + query_params["attempt_id"] = attempt_id + if step_id is not None: + query_params["step_id"] = step_id + if task_id is not None: + query_params["task_id"] = task_id + + response = jobs_client.list_job_logs(workspace=workspace, name=name, query_params=query_params) + render_job_logs(response.page().items) diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_log_renderer.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_log_renderer.py new file mode 100644 index 0000000000..3ce1cd6ec9 --- /dev/null +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_log_renderer.py @@ -0,0 +1,55 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from collections.abc import Iterable +from datetime import datetime + +from nemo_platform_plugin.jobs.schemas import PlatformJobLog +from rich.console import Console +from rich.text import Text + + +def render_job_logs(logs: Iterable[PlatformJobLog], *, console: Console | None = None) -> None: + output = console or Console() + for log in logs: + render_log_line( + output, + timestamp=log.timestamp, + step_id=log.job_step, + task_id=log.job_task, + message=log.message, + ) + + +def render_log_line( + console: Console, + *, + timestamp: datetime | None, + step_id: str | None, + task_id: str | None, + message: str, +) -> None: + line = Text() + line.append(f"[{_time_label(timestamp)}] ", style="dim") + scope = _scope(step_id, task_id) + if scope: + line.append(f"{scope} | ", style="dim") + line.append(message) + console.print(line) + + +def _time_label(timestamp: datetime | None = None) -> str: + value = timestamp.astimezone() if timestamp is not None else datetime.now() + return value.strftime("%H:%M:%S") + + +def _scope(step_id: str | None, task_id: str | None) -> str: + if step_id and task_id: + return f"{step_id}/{task_id}" + if step_id: + return step_id + if task_id: + return task_id + return "" diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_watch_renderer.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_watch_renderer.py index e6bb9e6038..b3af03d7f6 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_watch_renderer.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_watch_renderer.py @@ -21,6 +21,8 @@ from rich.console import Console from rich.text import Text +from nemo_platform_ext.cli.core.job_log_renderer import render_log_line + class JobWatchRenderResult(str, Enum): SUCCEEDED = "succeeded" @@ -129,13 +131,13 @@ def _render_status(console: Console, event: JobStatusEvent) -> None: def _render_log(console: Console, event: JobLogEvent) -> None: - line = Text() - line.append(f"[{_time_label(event.timestamp)}] ", style="dim") - scope = _scope(event) - if scope: - line.append(f"{scope} | ", style="dim") - line.append(event.message) - console.print(line) + render_log_line( + console, + timestamp=event.timestamp, + step_id=event.step_id, + task_id=event.task_id, + message=event.message, + ) def _render_warning(console: Console, event: JobWarningEvent) -> None: @@ -168,13 +170,3 @@ def _status_details(details: Mapping[str, object]) -> str: if len(parts) >= 6: break return " ".join(parts) - - -def _scope(event: JobLogEvent) -> str: - if event.step_id and event.task_id: - return f"{event.step_id}/{event.task_id}" - if event.step_id: - return event.step_id - if event.task_id: - return event.task_id - return "" diff --git a/packages/nemo_platform_ext/tests/cli/commands/api/test_jobs_tail.py b/packages/nemo_platform_ext/tests/cli/commands/api/test_jobs_tail.py new file mode 100644 index 0000000000..166093f6b8 --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/commands/api/test_jobs_tail.py @@ -0,0 +1,81 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from nemo_platform_ext.cli.commands.api.jobs import tail_platform_job + + +class _PlatformClient: + _get_workspace_path_param: MagicMock + + def __init__(self) -> None: + self._get_workspace_path_param = MagicMock(return_value="default") + + +def _ctx(client: _PlatformClient) -> SimpleNamespace: + state = MagicMock() + state.get_client.return_value = client + return SimpleNamespace(obj=state) + + +def test_jobs_tail_command_fetches_tail_page() -> None: + client = _PlatformClient() + logs = [object(), object()] + response = MagicMock() + response.page.return_value = SimpleNamespace(items=logs) + jobs_client = MagicMock() + jobs_client.list_job_logs.return_value = response + + with ( + patch("nemo_platform_ext.cli.commands.api.jobs.client_from_platform", return_value=jobs_client), + patch("nemo_platform_ext.cli.commands.api.jobs.render_job_logs") as render_logs, + ): + tail_platform_job( + _ctx(client), + name="job-a", + lines=25, + workspace=None, + attempt_id=1, + step_id="step-a", + task_id="task-a", + ) + + jobs_client.list_job_logs.assert_called_once_with( + workspace="default", + name="job-a", + query_params={"tail": 25, "attempt_id": 1, "step_id": "step-a", "task_id": "task-a"}, + ) + render_logs.assert_called_once_with(logs) + + +def test_jobs_tail_command_uses_explicit_workspace() -> None: + client = _PlatformClient() + response = MagicMock() + response.page.return_value = SimpleNamespace(items=[]) + jobs_client = MagicMock() + jobs_client.list_job_logs.return_value = response + + with ( + patch("nemo_platform_ext.cli.commands.api.jobs.client_from_platform", return_value=jobs_client), + patch("nemo_platform_ext.cli.commands.api.jobs.render_job_logs"), + ): + tail_platform_job( + _ctx(client), + name="job-a", + lines=100, + workspace="custom", + attempt_id=None, + step_id=None, + task_id=None, + ) + + client._get_workspace_path_param.assert_not_called() + jobs_client.list_job_logs.assert_called_once_with( + workspace="custom", + name="job-a", + query_params={"tail": 100}, + ) diff --git a/packages/nemo_platform_ext/tests/cli/core/test_job_log_renderer.py b/packages/nemo_platform_ext/tests/cli/core/test_job_log_renderer.py new file mode 100644 index 0000000000..c4080c6122 --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/core/test_job_log_renderer.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from datetime import datetime, timezone +from io import StringIO + +from nemo_platform_ext.cli.core.job_log_renderer import render_job_logs +from nemo_platform_plugin.jobs.schemas import PlatformJobLog +from rich.console import Console + + +def _console_pair() -> tuple[Console, StringIO]: + output = StringIO() + return Console(file=output, force_terminal=False, color_system=None, width=120), output + + +def test_render_job_logs_prints_logs_in_order() -> None: + console, output = _console_pair() + timestamp = datetime(2026, 9, 2, 12, 0, tzinfo=timezone.utc) + + render_job_logs( + [ + PlatformJobLog(timestamp=timestamp, job="job-a", job_step="step-a", job_task="task-a", message="started"), + PlatformJobLog(timestamp=timestamp, job="job-a", job_step="step-a", job_task="", message="running"), + ], + console=console, + ) + + rendered = output.getvalue() + expected_time = timestamp.astimezone().strftime("%H:%M:%S") + assert f"[{expected_time}] step-a/task-a | started" in rendered + assert f"[{expected_time}] step-a | running" in rendered + + +def test_render_job_logs_prints_nothing_for_empty_logs() -> None: + console, output = _console_pair() + + render_job_logs([], console=console) + + assert output.getvalue() == "" diff --git a/packages/nemo_platform_ext/tests/cli/test_app.py b/packages/nemo_platform_ext/tests/cli/test_app.py index 21ddf79578..0931e98b43 100644 --- a/packages/nemo_platform_ext/tests/cli/test_app.py +++ b/packages/nemo_platform_ext/tests/cli/test_app.py @@ -307,6 +307,16 @@ def test_jobs_watch_command_is_registered(): assert "--no-history" in result.stdout +def test_jobs_tail_command_is_registered(): + runner = CliRunner() + result = runner.invoke(app, ["jobs", "tail", "--help"]) + + assert result.exit_code == 0 + assert "Print the newest lines from a platform job log." in result.stdout + assert "--lines" in result.stdout + assert "-n" in result.stdout + + def test_jobs_create_exposes_wait_and_watch_flags(): runner = CliRunner() result = runner.invoke(app, ["jobs", "create", "--help"]) diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/data_designer/types.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/data_designer/types.py index 327df28753..1cff171d0f 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/data_designer/types.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/data_designer/types.py @@ -111,3 +111,4 @@ class ListDataDesignerJobsQueryParams(TypedDict, total=False): class DataDesignerJobLogsQueryParams(TypedDict, total=False): limit: NotRequired[int] page_cursor: NotRequired[str] + tail: NotRequired[int] diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/files/types.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/files/types.py index 201c1511fe..c32bd1c890 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/files/types.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/files/types.py @@ -169,6 +169,7 @@ class OtlpLogQueryRequest(BaseModel): filters: dict[str, str] = Field(default_factory=dict) limit: int | None = None page_cursor: str | None = None + tail: int | None = None artifact_base_path: str | None = None diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py index be31bffa33..4ea6270beb 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py @@ -1142,6 +1142,7 @@ async def get_job_logs( sdk: AsyncNeMoPlatform = Depends(get_sdk_client), limit: int | None = Query(default=None), page_cursor: str | None = Query(default=None), + tail: int | None = Query(default=None, gt=0, le=10_000), ) -> PlatformJobLogPage: f"""Get the logs of a job by name for the {service_name} microservice.""" @@ -1150,6 +1151,8 @@ async def get_job_logs( logs_query["limit"] = limit if page_cursor is not None: logs_query["page_cursor"] = page_cursor + if tail is not None: + logs_query["tail"] = tail logs_page = ( await client_from_platform(sdk, AsyncJobsClient).list_job_logs( workspace=workspace, name=name, query_params=logs_query diff --git a/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/types.py b/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/types.py index b4a2d3114f..7dc648d687 100644 --- a/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/types.py +++ b/packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/types.py @@ -356,3 +356,4 @@ class JobLogsQueryParams(TypedDict, total=False): attempt_id: NotRequired[int] step_id: NotRequired[str] task_id: NotRequired[str] + tail: NotRequired[int] diff --git a/packages/nemo_platform_plugin/tests/jobs/test_endpoints.py b/packages/nemo_platform_plugin/tests/jobs/test_endpoints.py index bcb400be96..f1386bd6cc 100644 --- a/packages/nemo_platform_plugin/tests/jobs/test_endpoints.py +++ b/packages/nemo_platform_plugin/tests/jobs/test_endpoints.py @@ -10,11 +10,15 @@ from nemo_platform_plugin.client.types import BinaryContent, CursorPagination, Paginated, PreparedRequest from nemo_platform_plugin.jobs import endpoints +from nemo_platform_plugin.jobs.providers import ContainerSpec, CPUExecutionProvider from nemo_platform_plugin.jobs.schemas import ( + FileStorageType, PlatformJobLog, PlatformJobResultCreateRequest, + PlatformJobStatus, PlatformJobStatusResponse, ) +from nemo_platform_plugin.jobs.spec import PlatformJobSpec, PlatformJobStepSpec from nemo_platform_plugin.jobs.types import ( CreatePlatformJobRequest, JobStatusDetailsUpdate, @@ -31,7 +35,14 @@ def _create_request() -> CreatePlatformJobRequest: return CreatePlatformJobRequest( spec={}, source="test", - platform_spec={"steps": [{"name": "step-one", "executor": {"provider": "cpu", "container": {"image": "img"}}}]}, + platform_spec=PlatformJobSpec( + steps=[ + PlatformJobStepSpec( + name="step-one", + executor=CPUExecutionProvider(container=ContainerSpec(image="img")), + ) + ] + ), ) @@ -142,6 +153,7 @@ def test_update_job_status_details() -> None: assert prepared.path_template.endswith("/jobs/{name}/status-details") assert prepared.response_type is None # RootModel serialises to the bare JSON object + assert isinstance(prepared.content, bytes) assert json.loads(prepared.content) == {"note": "x"} @@ -152,11 +164,11 @@ def test_update_job_status_details() -> None: def test_list_job_logs() -> None: prepared = endpoints.list_job_logs( - workspace="default", name="j-1", query_params={"limit": 50, "page_cursor": "abc"} + workspace="default", name="j-1", query_params={"limit": 50, "page_cursor": "abc", "tail": 500} ) assert prepared.method == "GET" assert prepared.path_template.endswith("/jobs/{name}/logs") - assert prepared.query_params == {"limit": 50, "page_cursor": "abc"} + assert prepared.query_params == {"limit": 50, "page_cursor": "abc", "tail": 500} assert get_origin(prepared.response_type) is Paginated assert get_args(prepared.response_type) == (PlatformJobLog, CursorPagination) @@ -167,7 +179,7 @@ def test_list_job_logs() -> None: def test_create_job_result() -> None: - body = PlatformJobResultCreateRequest(artifact_url="s3://x", artifact_storage_type="fileset") + body = PlatformJobResultCreateRequest(artifact_url="s3://x", artifact_storage_type=FileStorageType.FILESET) prepared = endpoints.create_job_result(workspace="default", job="j-1", name="out", body=body) assert prepared.method == "POST" assert prepared.path_params == {"workspace": "default", "job": "j-1", "name": "out"} @@ -220,7 +232,7 @@ def test_get_job_step() -> None: def test_update_job_step_status() -> None: - body = PlatformJobStatusUpdateRequest(status="active") + body = PlatformJobStatusUpdateRequest(status=PlatformJobStatus.ACTIVE) prepared = endpoints.update_job_step_status(workspace="default", job="j-1", name="step-one", body=body) assert prepared.method == "PATCH" assert prepared.path_template.endswith("/steps/{name}/status") @@ -240,7 +252,7 @@ def test_list_job_step_tasks() -> None: def test_update_job_step_task() -> None: - body = PlatformJobTaskUpdate(status="completed") + body = PlatformJobTaskUpdate(status=PlatformJobStatus.COMPLETED) prepared = endpoints.update_job_step_task(workspace="default", job="j-1", step="step-one", name="task-1", body=body) assert prepared.method == "PUT" assert prepared.path_params == {"workspace": "default", "job": "j-1", "step": "step-one", "name": "task-1"} @@ -262,14 +274,16 @@ def test_get_job_step_task() -> None: def test_create_job_body_roundtrip() -> None: body = _create_request() prepared = endpoints.create_job(workspace="default", body=body) + assert isinstance(prepared.content, bytes) content = json.loads(prepared.content) assert content["source"] == "test" assert content["platform_spec"]["steps"][0]["name"] == "step-one" def test_step_status_update_excludes_unset() -> None: - body = PlatformJobStatusUpdateRequest(status="active") + body = PlatformJobStatusUpdateRequest(status=PlatformJobStatus.ACTIVE) prepared = endpoints.update_job_step_status(workspace="default", job="j-1", name="s", body=body) + assert isinstance(prepared.content, bytes) content = json.loads(prepared.content) assert content["status"] == "active" @@ -277,4 +291,4 @@ def test_step_status_update_excludes_unset() -> None: def test_step_with_context_response_type() -> None: prepared = endpoints.list_steps(workspace="default", name="j-1") # Paginated marker parametrised with the step-with-context model. - assert prepared.response_type.__args__[0] is PlatformJobStepWithContext # type: ignore[attr-defined] + assert get_args(prepared.response_type)[0] is PlatformJobStepWithContext diff --git a/packages/nmp_common/src/nmp/common/api/utils.py b/packages/nmp_common/src/nmp/common/api/utils.py index 945ecfe85c..3138fe0084 100644 --- a/packages/nmp_common/src/nmp/common/api/utils.py +++ b/packages/nmp_common/src/nmp/common/api/utils.py @@ -352,7 +352,20 @@ def _sort_schemas(spec: Dict) -> Dict: return spec +def _apply_schema_default_extensions(spec: Dict) -> Dict: + """Promote opt-in schema default metadata into the emitted OpenAPI schema.""" + + def _visitor(key: str, value: Any, parent: Dict): + if key == "x-schema-default": + parent.setdefault("default", value) + del parent[key] + + _walk_spec(spec, _visitor) + return spec + + def tweak_spec(spec: Dict) -> Dict: + spec = _apply_schema_default_extensions(spec) _walk_spec(spec, _anyof_null_visitor) spec = _normalize_refs_and_schema_keys(spec) spec = _split_input_output_schemas(spec) diff --git a/packages/nmp_common/src/nmp/common/jobs/log_client.py b/packages/nmp_common/src/nmp/common/jobs/log_client.py index 6024cb533d..919406cb4e 100644 --- a/packages/nmp_common/src/nmp/common/jobs/log_client.py +++ b/packages/nmp_common/src/nmp/common/jobs/log_client.py @@ -46,6 +46,7 @@ async def query_logs( page_size: int = 100, page_cursor: str | None = None, artifact_base_path: str | None = None, + tail: int | None = None, ) -> PlatformJobLogPage: """Query job logs via Files service OTLP endpoint. @@ -56,6 +57,7 @@ async def query_logs( page_size: Number of results per page page_cursor: Encoded cursor for pagination artifact_base_path: Folder inside the fileset the logs were nested under + tail: Number of newest log lines to return Returns: PlatformJobLogPage with data, total count, and pagination cursors @@ -63,8 +65,9 @@ async def query_logs( try: body = OtlpLogQueryRequest( filters=filters or {}, - limit=page_size, + limit=None if tail is not None else page_size, page_cursor=page_cursor, + tail=tail, artifact_base_path=artifact_base_path, ) resp = await self._files_client.query_otlp_logs( diff --git a/packages/nmp_common/src/nmp/common/jobs/schemas.py b/packages/nmp_common/src/nmp/common/jobs/schemas.py index 14655e13f3..c7493b2112 100644 --- a/packages/nmp_common/src/nmp/common/jobs/schemas.py +++ b/packages/nmp_common/src/nmp/common/jobs/schemas.py @@ -2,7 +2,9 @@ # SPDX-License-Identifier: Apache-2.0 import json +from datetime import datetime from enum import Enum +from typing import Literal import base58 from nemo_platform_plugin.jobs.schemas import FileStorageType as FileStorageType @@ -15,7 +17,7 @@ from nemo_platform_plugin.jobs.schemas import PlatformJobStatusResponse as PlatformJobStatusResponse from nemo_platform_plugin.jobs.schemas import PlatformJobStepStatusResponse as PlatformJobStepStatusResponse from nemo_platform_plugin.jobs.schemas import PlatformJobTaskStatusResponse as PlatformJobTaskStatusResponse -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field, ValidationError # ============================================================================= # Pagination (stays in nmp-common — depends on base58) @@ -50,7 +52,7 @@ def decode(page_cursor: str) -> "PageCursor": start_id, direction_int = json.loads(decoded) direction = PaginationDirection(direction_int) return PageCursor(start_id=start_id, direction=direction) - except (ValueError, TypeError, Exception): + except (ValueError, TypeError): raise ValueError("Invalid page cursor") @@ -58,3 +60,40 @@ class InvalidPageCursorError(Exception): """Custom exception for invalid page cursor errors.""" pass + + +class LogPageCursorV1(BaseModel): + """Versioned cursor for fetching the page preceding a returned boundary.""" + + model_config = ConfigDict(populate_by_name=True) + + version: Literal[1] = Field(default=1, validation_alias="v", serialization_alias="v") + boundary_timestamp: datetime = Field(validation_alias="t", serialization_alias="t") + boundary_row_hash: str = Field(validation_alias="r", serialization_alias="r", min_length=32, max_length=32) + query_scope_hash: str = Field(validation_alias="q", serialization_alias="q", min_length=32, max_length=32) + emitted_boundary_rows: int = Field(default=0, validation_alias="e", serialization_alias="e", ge=0) + + def encode(self) -> str: + """Encode the cursor using compact aliases and base58 JSON.""" + cursor_data = self.model_dump(mode="json", by_alias=True) + json_str = json.dumps(cursor_data, separators=(",", ":")) + return base58.b58encode(json_str.encode()).decode() + + +LogPageCursorV0 = PageCursor +LogPageCursor = LogPageCursorV0 | LogPageCursorV1 + + +def decode_log_page_cursor(page_cursor: str) -> LogPageCursor: + """Decode either a v0 page-number cursor or a v1 boundary cursor.""" + try: + decoded = base58.b58decode(page_cursor.encode()).decode() + payload = json.loads(decoded) + if isinstance(payload, list): + start_id, direction_int = payload + return LogPageCursorV0(start_id=start_id, direction=PaginationDirection(direction_int)) + if isinstance(payload, dict) and payload.get("v") == 1: + return LogPageCursorV1.model_validate(payload) + except (ValueError, TypeError, ValidationError): + pass + raise ValueError("Invalid page cursor") diff --git a/packages/nmp_common/tests/api/test_utils_openapi_spec.py b/packages/nmp_common/tests/api/test_utils_openapi_spec.py index 3d58a36f70..e33ca5a9f5 100644 --- a/packages/nmp_common/tests/api/test_utils_openapi_spec.py +++ b/packages/nmp_common/tests/api/test_utils_openapi_spec.py @@ -473,3 +473,71 @@ def test_anyof_null_collapse_hoists_optional_union(): ], "title": "Replace", } + + +def test_tweak_spec_promotes_schema_default_extension(): + spec = { + "components": { + "schemas": { + "LogQueryRequest": { + "type": "object", + "properties": { + "limit": { + "type": "integer", + "x-schema-default": 100, + } + }, + } + } + }, + "paths": { + "/logs": { + "get": { + "parameters": [ + { + "name": "limit", + "in": "query", + "schema": { + "type": "integer", + "x-schema-default": 100, + }, + } + ] + } + } + }, + } + + result = tweak_spec(spec) + + query_schema = result["paths"]["/logs"]["get"]["parameters"][0]["schema"] + model_schema = result["components"]["schemas"]["LogQueryRequest"]["properties"]["limit"] + assert query_schema["default"] == 100 + assert "x-schema-default" not in query_schema + assert model_schema["default"] == 100 + assert "x-schema-default" not in model_schema + + +def test_tweak_spec_removes_null_schema_default_extension_after_nullable_collapse(): + spec = { + "components": { + "schemas": { + "SearchParams": { + "type": "object", + "properties": { + "cursor": { + "anyOf": [{"type": "string"}, {"type": "null"}], + "x-schema-default": None, + "title": "Cursor", + } + }, + } + } + }, + "paths": {}, + } + + result = tweak_spec(spec) + + cursor_schema = result["components"]["schemas"]["SearchParams"]["properties"]["cursor"] + assert cursor_schema == {"type": "string", "title": "Cursor"} diff --git a/packages/nmp_common/tests/api_factory/test_api_factory.py b/packages/nmp_common/tests/api_factory/test_api_factory.py index a48f49ac35..02cd3bd9bf 100644 --- a/packages/nmp_common/tests/api_factory/test_api_factory.py +++ b/packages/nmp_common/tests/api_factory/test_api_factory.py @@ -528,6 +528,29 @@ def test_get_job_logs_with_page_cursor(mock_service_with_job_routes): ) +def test_get_job_logs_with_tail(mock_service_with_job_routes): + """Test get_job_logs with tail parameter.""" + app, mock_jobs = mock_service_with_job_routes + client = TestClient(app) + + mock_log_page = create_mock_log_page(num_logs=2, total=10, prev_page="prev_cursor_123") + mock_jobs.list_job_logs = AsyncMock(return_value=_cursor_page_resp(mock_log_page)) + + response = client.get("/v2/workspaces/default/test/jobs/test-job-123/logs?tail=2") + + assert response.status_code == 200 + response_data = response.json() + assert len(response_data["data"]) == 2 + assert response_data["total"] == 10 + assert response_data["prev_page"] == "prev_cursor_123" + + mock_jobs.list_job_logs.assert_called_once_with( + workspace="default", + name="test-job-123", + query_params={"tail": 2}, + ) + + def test_get_job_logs_with_both_parameters(mock_service_with_job_routes): """Test get_job_logs with both limit and page_cursor parameters.""" app, mock_jobs = mock_service_with_job_routes diff --git a/packages/nmp_common/tests/jobs/test_log_client.py b/packages/nmp_common/tests/jobs/test_log_client.py index 3e4fd695f0..54fc626fde 100644 --- a/packages/nmp_common/tests/jobs/test_log_client.py +++ b/packages/nmp_common/tests/jobs/test_log_client.py @@ -3,16 +3,22 @@ """Tests for JobLogsClient SDK wrapper and PageCursor.""" +import json +from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch +import base58 import httpx import pytest from nemo_platform_plugin.client.errors import NotFoundError from nmp.common.jobs.log_client import JobLogsClient from nmp.common.jobs.schemas import ( + LogPageCursorV1, PageCursor, PaginationDirection, + PlatformJobLog, PlatformJobLogPage, + decode_log_page_cursor, ) # ============================================================================= @@ -46,6 +52,46 @@ def test_decode_invalid_cursor(): PageCursor.decode("invalid_cursor_string") +def test_decode_log_page_cursor_accepts_existing_page_cursor(): + """Log cursor decoding preserves the legacy page-number cursor format.""" + cursor = PageCursor(start_id=2, direction=PaginationDirection.FORWARD).encode() + + decoded = decode_log_page_cursor(cursor) + + assert isinstance(decoded, PageCursor) + assert decoded.start_id == 2 + assert decoded.direction == PaginationDirection.FORWARD + + +def test_log_page_cursor_v1_round_trip(): + """V1 cursor encoding stays compact while validating typed fields.""" + cursor = LogPageCursorV1( + boundary_timestamp=datetime(2026, 9, 2, 12, 0, 1), + boundary_row_hash="a" * 32, + query_scope_hash="b" * 32, + emitted_boundary_rows=2, + ) + + decoded = decode_log_page_cursor(cursor.encode()) + + assert decoded == cursor + + +def test_decode_log_page_cursor_rejects_invalid_v1_hash_length(): + """V1 cursors reject malformed compact hash fields.""" + bad_payload = { + "v": 1, + "t": "2026-09-02T12:00:01", + "r": "short", + "q": "b" * 32, + "e": 0, + } + encoded = base58.b58encode(json.dumps(bad_payload).encode()).decode() + + with pytest.raises(ValueError, match="Invalid page cursor"): + decode_log_page_cursor(encoded) + + # ============================================================================= # JobLogsClient Tests # ============================================================================= @@ -78,13 +124,13 @@ async def test_query_logs_success(log_client): page = PlatformJobLogPage( data=[ - { - "timestamp": "2024-01-01T12:00:00", - "job": "job-123", - "job_step": "step1", - "job_task": "task1", - "message": "Test log message", - } + PlatformJobLog( + timestamp=datetime(2024, 1, 1, 12, 0, 0), + job="job-123", + job_step="step1", + job_task="task1", + message="Test log message", + ) ], total=1, next_page=None, @@ -129,6 +175,27 @@ async def test_query_logs_with_pagination_cursor(log_client): assert call_kwargs["body"].page_cursor == cursor +async def test_query_logs_with_tail(log_client): + """Test query_logs passes tail without also setting limit.""" + client, mock_files = log_client + + mock_files.query_otlp_logs.return_value = _make_response_mock( + PlatformJobLogPage(data=[], total=0, next_page=None, prev_page=None) + ) + + await client.query_logs( + fileset="logs", + workspace="test-workspace", + filters={"job": "job-123"}, + tail=500, + ) + + call_kwargs = mock_files.query_otlp_logs.call_args.kwargs + body = call_kwargs["body"] + assert body.tail == 500 + assert body.limit is None + + async def test_query_logs_404_returns_empty_page(log_client): """Test that NotFoundError returns an empty page.""" client, mock_files = log_client diff --git a/plugins/nemo-agents/openapi/openapi.yaml b/plugins/nemo-agents/openapi/openapi.yaml index 37ef79e398..93b1cd6e26 100644 --- a/plugins/nemo-agents/openapi/openapi.yaml +++ b/plugins/nemo-agents/openapi/openapi.yaml @@ -1112,6 +1112,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -1582,6 +1590,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -1860,6 +1876,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -2234,6 +2258,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -2704,6 +2736,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -2982,6 +3022,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -3356,6 +3404,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-anonymizer/openapi/openapi.yaml b/plugins/nemo-anonymizer/openapi/openapi.yaml index 6ea5a3fc07..163440a340 100644 --- a/plugins/nemo-anonymizer/openapi/openapi.yaml +++ b/plugins/nemo-anonymizer/openapi/openapi.yaml @@ -327,6 +327,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-auditor/openapi/openapi.yaml b/plugins/nemo-auditor/openapi/openapi.yaml index 50cddbcf89..a197fa0a5a 100644 --- a/plugins/nemo-auditor/openapi/openapi.yaml +++ b/plugins/nemo-auditor/openapi/openapi.yaml @@ -550,6 +550,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-customizer/openapi/openapi.yaml b/plugins/nemo-customizer/openapi/openapi.yaml index 994ac2f5d7..9055823633 100644 --- a/plugins/nemo-customizer/openapi/openapi.yaml +++ b/plugins/nemo-customizer/openapi/openapi.yaml @@ -315,6 +315,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -689,6 +697,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -1063,6 +1079,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-data-designer/openapi/openapi.yaml b/plugins/nemo-data-designer/openapi/openapi.yaml index e17ab278b5..e503e1cf29 100644 --- a/plugins/nemo-data-designer/openapi/openapi.yaml +++ b/plugins/nemo-data-designer/openapi/openapi.yaml @@ -300,6 +300,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -674,6 +682,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -1048,6 +1064,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -1422,6 +1446,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index a6cd1bdfc7..49b2eeeb35 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -472,6 +472,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -976,6 +984,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-iron-swarm/openapi/openapi.yaml b/plugins/nemo-iron-swarm/openapi/openapi.yaml index ba772b06ca..b583883743 100644 --- a/plugins/nemo-iron-swarm/openapi/openapi.yaml +++ b/plugins/nemo-iron-swarm/openapi/openapi.yaml @@ -315,6 +315,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response @@ -1375,6 +1383,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/plugins/nemo-safe-synthesizer/openapi/openapi.yaml b/plugins/nemo-safe-synthesizer/openapi/openapi.yaml index 8b9792ba59..8ab5aa49ac 100644 --- a/plugins/nemo-safe-synthesizer/openapi/openapi.yaml +++ b/plugins/nemo-safe-synthesizer/openapi/openapi.yaml @@ -439,6 +439,14 @@ paths: schema: title: Page Cursor type: string + - name: tail + in: query + required: false + schema: + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 responses: '200': description: Successful Response diff --git a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml index 3eee5f6d36..1d430636d5 100644 --- a/sdk/python/nemo-platform/.nmpcontext/openapi.yaml +++ b/sdk/python/nemo-platform/.nmpcontext/openapi.yaml @@ -5830,20 +5830,31 @@ paths: in: query required: false schema: - type: integer - exclusiveMinimum: 0 description: Maximum number of logs to return - default: 100 title: Limit + default: 100 + type: integer + maximum: 10000 + exclusiveMinimum: 0 description: Maximum number of logs to return - name: page_cursor in: query required: false schema: - type: string description: Page cursor title: Page Cursor + type: string description: Page cursor + - name: tail + in: query + required: false + schema: + description: Number of newest log lines to return + title: Tail + type: integer + maximum: 10000 + exclusiveMinimum: 0 + description: Number of newest log lines to return - name: attempt_id in: query required: false @@ -14653,16 +14664,22 @@ components: title: Filters description: Key-value filters to apply to the query limit: - type: integer - maximum: 1000.0 - exclusiveMinimum: 0.0 title: Limit description: Maximum number of results to return default: 100 + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 page_cursor: title: Page Cursor description: Cursor for pagination type: string + tail: + title: Tail + description: Number of newest log lines to return + type: integer + maximum: 10000.0 + exclusiveMinimum: 0.0 artifact_base_path: title: Artifact Base Path description: Folder inside the fileset the logs were nested under (must diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/files/otlp/logs.py b/sdk/python/nemo-platform/src/nemo_platform/resources/files/otlp/logs.py index 43d38b9959..a93acdcfa8 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/files/otlp/logs.py +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/files/otlp/logs.py @@ -117,6 +117,7 @@ def query( filters: Dict[str, str] | Omit = omit, limit: int | Omit = omit, page_cursor: str | Omit = omit, + tail: int | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, @@ -140,6 +141,8 @@ def query( page_cursor: Cursor for pagination + tail: Number of newest log lines to return + extra_headers: Send extra headers extra_query: Add additional query parameters to the request @@ -166,6 +169,7 @@ def query( "filters": filters, "limit": limit, "page_cursor": page_cursor, + "tail": tail, }, log_query_params.LogQueryParams, ), @@ -256,6 +260,7 @@ async def query( filters: Dict[str, str] | Omit = omit, limit: int | Omit = omit, page_cursor: str | Omit = omit, + tail: int | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. extra_headers: Headers | None = None, @@ -279,6 +284,8 @@ async def query( page_cursor: Cursor for pagination + tail: Number of newest log lines to return + extra_headers: Send extra headers extra_query: Add additional query parameters to the request @@ -305,6 +312,7 @@ async def query( "filters": filters, "limit": limit, "page_cursor": page_cursor, + "tail": tail, }, log_query_params.LogQueryParams, ), diff --git a/sdk/python/nemo-platform/src/nemo_platform/resources/jobs/jobs.py b/sdk/python/nemo-platform/src/nemo_platform/resources/jobs/jobs.py index 3fb1c4e410..9c3d71c392 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/resources/jobs/jobs.py +++ b/sdk/python/nemo-platform/src/nemo_platform/resources/jobs/jobs.py @@ -353,6 +353,7 @@ def get_logs( limit: int | Omit = omit, page_cursor: str | Omit = omit, step_id: str | Omit = omit, + tail: int | Omit = omit, task_id: str | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. @@ -373,6 +374,8 @@ def get_logs( step_id: Filter logs by step name + tail: Number of newest log lines to return + task_id: Filter logs by task ID extra_headers: Send extra headers @@ -403,6 +406,7 @@ def get_logs( "limit": limit, "page_cursor": page_cursor, "step_id": step_id, + "tail": tail, "task_id": task_id, }, job_get_logs_params.JobGetLogsParams, @@ -872,6 +876,7 @@ def get_logs( limit: int | Omit = omit, page_cursor: str | Omit = omit, step_id: str | Omit = omit, + tail: int | Omit = omit, task_id: str | Omit = omit, # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. # The extra values given here take precedence over values defined on the client or passed to this method. @@ -892,6 +897,8 @@ def get_logs( step_id: Filter logs by step name + tail: Number of newest log lines to return + task_id: Filter logs by task ID extra_headers: Send extra headers @@ -922,6 +929,7 @@ def get_logs( "limit": limit, "page_cursor": page_cursor, "step_id": step_id, + "tail": tail, "task_id": task_id, }, job_get_logs_params.JobGetLogsParams, diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/files/otlp/log_query_params.py b/sdk/python/nemo-platform/src/nemo_platform/types/files/otlp/log_query_params.py index c00908bf93..2058871651 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/files/otlp/log_query_params.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/files/otlp/log_query_params.py @@ -40,3 +40,6 @@ class LogQueryParams(TypedDict, total=False): page_cursor: str """Cursor for pagination""" + + tail: int + """Number of newest log lines to return""" diff --git a/sdk/python/nemo-platform/src/nemo_platform/types/jobs/job_get_logs_params.py b/sdk/python/nemo-platform/src/nemo_platform/types/jobs/job_get_logs_params.py index fea50736a5..02a4115ca3 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/types/jobs/job_get_logs_params.py +++ b/sdk/python/nemo-platform/src/nemo_platform/types/jobs/job_get_logs_params.py @@ -37,5 +37,8 @@ class JobGetLogsParams(TypedDict, total=False): step_id: str """Filter logs by step name""" + tail: int + """Number of newest log lines to return""" + task_id: str """Filter logs by task ID""" diff --git a/sdk/python/nemo-platform/tests/api_resources/files/otlp/test_logs.py b/sdk/python/nemo-platform/tests/api_resources/files/otlp/test_logs.py index b6ef76b21c..f46caaaf5b 100644 --- a/sdk/python/nemo-platform/tests/api_resources/files/otlp/test_logs.py +++ b/sdk/python/nemo-platform/tests/api_resources/files/otlp/test_logs.py @@ -114,6 +114,7 @@ def test_method_query_with_all_params(self, client: NeMoPlatform) -> None: filters={"foo": "string"}, limit=1, page_cursor="page_cursor", + tail=1, ) assert_matches_type(PlatformJobLogPage, log, path=["response"]) @@ -247,6 +248,7 @@ async def test_method_query_with_all_params(self, async_client: AsyncNeMoPlatfor filters={"foo": "string"}, limit=1, page_cursor="page_cursor", + tail=1, ) assert_matches_type(PlatformJobLogPage, log, path=["response"]) diff --git a/sdk/python/nemo-platform/tests/api_resources/test_jobs.py b/sdk/python/nemo-platform/tests/api_resources/test_jobs.py index 979ce4e038..f1f9cc63f1 100644 --- a/sdk/python/nemo-platform/tests/api_resources/test_jobs.py +++ b/sdk/python/nemo-platform/tests/api_resources/test_jobs.py @@ -451,6 +451,7 @@ def test_method_get_logs_with_all_params(self, client: NeMoPlatform) -> None: limit=1, page_cursor="page_cursor", step_id="step_id", + tail=1, task_id="task_id", ) assert_matches_type(SyncLogsPagination[PlatformJobLog], job, path=["response"]) @@ -1153,6 +1154,7 @@ async def test_method_get_logs_with_all_params(self, async_client: AsyncNeMoPlat limit=1, page_cursor="page_cursor", step_id="step_id", + tail=1, task_id="task_id", ) assert_matches_type(AsyncLogsPagination[PlatformJobLog], job, path=["response"]) diff --git a/services/core/files/src/nmp/core/files/api/v2/otlp/endpoints.py b/services/core/files/src/nmp/core/files/api/v2/otlp/endpoints.py index 76776a8ae8..fcb15d702c 100644 --- a/services/core/files/src/nmp/core/files/api/v2/otlp/endpoints.py +++ b/services/core/files/src/nmp/core/files/api/v2/otlp/endpoints.py @@ -12,7 +12,7 @@ from nmp.common.entities.client import ( EntityClient, ) -from nmp.common.jobs.schemas import PlatformJobLogPage +from nmp.common.jobs.schemas import InvalidPageCursorError, PlatformJobLogPage from nmp.common.service.dependencies import get_entity_client, get_sdk_client from nmp.core.files.api.endpoint_helpers import ( get_fileset, @@ -37,6 +37,9 @@ logger = logging.getLogger(__name__) +MAX_LOG_QUERY_LINES = 10_000 +DEFAULT_LOG_QUERY_LIMIT = 100 + class LogQueryRequest(BaseModel): """Request body for querying logs from a fileset.""" @@ -45,22 +48,46 @@ class LogQueryRequest(BaseModel): default_factory=dict, description="Key-value filters to apply to the query", ) - limit: int = Field( - default=100, + limit: int | None = Field( + default=None, gt=0, - le=1000, + le=MAX_LOG_QUERY_LINES, description="Maximum number of results to return", + json_schema_extra={"default": DEFAULT_LOG_QUERY_LIMIT}, ) page_cursor: str | None = Field( default=None, description="Cursor for pagination", ) + tail: int | None = Field( + default=None, + gt=0, + le=MAX_LOG_QUERY_LINES, + description="Number of newest log lines to return", + ) artifact_base_path: str | None = Field( default=None, description="Folder inside the fileset the logs were nested under (must match the value used on write)", ) +def _validate_log_query_request(request: LogQueryRequest) -> int: + if request.tail is not None: + if request.limit is not None: + raise HTTPException( + status_code=400, + detail="tail cannot be combined with limit; tail controls the returned log window size.", + ) + if request.page_cursor is not None: + raise HTTPException( + status_code=400, + detail="tail cannot be combined with page_cursor; pass the returned prev_page as page_cursor without tail.", + ) + return request.tail + + return request.limit if request.limit is not None else DEFAULT_LOG_QUERY_LIMIT + + router = APIRouter() @@ -92,15 +119,20 @@ async def query_otlp_logs( fileset = await get_fileset(workspace, name, entity_store) secrets = await resolve_storage_secrets_for_user(fileset.storage, workspace, sdk, auth_client) storage = storage_impl_factory(fileset.storage, secrets) + effective_limit = _validate_log_query_request(request) try: return await log_storage.query_logs( storage=storage, filters=request.filters, - page_size=request.limit, + page_size=effective_limit, page_cursor=request.page_cursor, + tail=request.tail, artifact_base_path=request.artifact_base_path, ) + except InvalidPageCursorError as e: + logger.error(f"Invalid page cursor: {str(e)}") + raise HTTPException(status_code=400, detail=str(e)) except InvalidFilterError as e: logger.error(f"Invalid filter: {str(e)}") raise HTTPException(status_code=400, detail=f"Invalid filter: {str(e)}") diff --git a/services/core/files/src/nmp/core/files/app/log_db.py b/services/core/files/src/nmp/core/files/app/log_db.py new file mode 100644 index 0000000000..b75e7fd57f --- /dev/null +++ b/services/core/files/src/nmp/core/files/app/log_db.py @@ -0,0 +1,617 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""DuckDB repository for persisted job logs. + +This module provides local DuckDB-based query and insert operations for job logs, +avoiding the cross-service HTTP overhead of the previous FilesetFileSystem approach. + +NOTE: pandas and S3StorageImpl are intentionally imported inside methods +for startup performance. Do not hoist them to module level. + +TODO: Right now, this is very Jobs logs specific; in a future MR we should try to make it more generic. +""" + +import hashlib +import json +import logging +import os +import re +import uuid +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol +from urllib.parse import urlparse + +import duckdb +from nmp.common.files.storage_config import S3StorageConfig +from nmp.common.jobs.schemas import ( + InvalidPageCursorError, + LogPageCursor, + LogPageCursorV0, + LogPageCursorV1, + PageCursor, + PaginationDirection, + PlatformJobLog, + PlatformJobLogPage, +) +from nmp.core.files.app.backends.base import StorageImpl +from nmp.core.files.exceptions import InvalidFilterError +from pydantic import BaseModel, Field + +logger = logging.getLogger(__name__) + +DuckDBQueryParam = str | int | datetime + + +class LogEntry(BaseModel): + """Internal representation of a log entry for storage.""" + + workspace: str + job: str + job_attempt: str + job_step: str + job_task: str + log_message: str + timestamp: datetime + + +class QueriedLogEntry(LogEntry): + """Internal representation of a queried log row.""" + + total_count: int = Field(alias="_total_count") + row_hash: str | None = None + + +@dataclass(frozen=True) +class LogQuery: + """Repository-level log query input.""" + + base_path: str + filters: dict[str, str] + page_size: int + cursor: LogPageCursor | None = None + tail: int | None = None + artifact_base_path: str | None = None + + +class LogRepository(Protocol): + """Persistence boundary for job logs.""" + + def query_logs(self, query: LogQuery, storage: StorageImpl) -> PlatformJobLogPage: + """Read a page of logs from the repository.""" + ... + + def insert_logs(self, log_entries: list[LogEntry], base_path: str, storage: StorageImpl) -> int: + """Persist log entries into the repository.""" + ... + + +class DuckDBLogRepository: + """DuckDB-backed repository for Hive-partitioned parquet log storage. + + This class is stateless - each operation creates its own DuckDB connection. + This ensures thread safety when operations run concurrently in the thread pool. + + For local storage: Uses direct path access with DuckDB. + For S3 storage: Uses s3:// URIs with DuckDB's httpfs extension. + """ + + # Partition columns in Hive directory order + PARTITION_COLUMNS = ("job", "job_attempt", "job_step", "job_task") + SAFE_PARTITION_VALUE = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") + SELECT_LOG_COLUMNS_SQL = ( + "workspace, CAST(job AS VARCHAR) AS job, CAST(job_attempt AS VARCHAR) AS job_attempt, " + "CAST(job_step AS VARCHAR) AS job_step, CAST(job_task AS VARCHAR) AS job_task, log_message, timestamp" + ) + ROW_HASH_SQL = ( + "left(sha256(concat_ws(chr(31), CAST(timestamp AS VARCHAR), CAST(workspace AS VARCHAR), " + "CAST(job AS VARCHAR), CAST(job_attempt AS VARCHAR), CAST(job_step AS VARCHAR), " + "CAST(job_task AS VARCHAR), log_message)), 32)" + ) + + def query_logs(self, query: LogQuery, storage: StorageImpl) -> PlatformJobLogPage: + """Query logs from parquet files using direct storage access. + + Supports both legacy page-number cursors and v1 boundary cursors used by + tail pagination. This method is synchronous; ``LogStorage`` runs it in a + thread pool to avoid blocking the event loop. + """ + try: + if query.tail is not None: + return self._query_tail_window( + base_path=query.base_path, + filters=query.filters, + page_size=query.tail, + storage=storage, + artifact_base_path=query.artifact_base_path, + ) + + if isinstance(query.cursor, LogPageCursorV1): + return self._query_cursor_v1_window( + base_path=query.base_path, + filters=query.filters, + page_size=query.page_size, + cursor=query.cursor, + storage=storage, + artifact_base_path=query.artifact_base_path, + ) + + current_page = query.cursor.start_id if isinstance(query.cursor, LogPageCursorV0) else 1 + direction = ( + query.cursor.direction if isinstance(query.cursor, LogPageCursorV0) else PaginationDirection.FORWARD + ) + return self._query_offset_page( + base_path=query.base_path, + filters=query.filters, + page_size=query.page_size, + current_page=current_page, + direction=direction, + storage=storage, + ) + except duckdb.IOException as e: + if "No files found that match the pattern" in str(e): + return PlatformJobLogPage(data=[], total=0, next_page=None, prev_page=None) + logger.exception("IO error when querying logs") + raise + + @classmethod + def _connect(cls, storage: StorageImpl) -> duckdb.DuckDBPyConnection: + conn = duckdb.connect( + ":memory:", + config={ + "autoload_known_extensions": "false", + "autoinstall_known_extensions": "false", + }, + ) + from nmp.core.files.app.backends.s3 import S3StorageImpl + + if isinstance(storage, S3StorageImpl): + cls._load_s3_extensions(conn) + cls._configure_s3_secret(conn, storage.config) + return conn + + @staticmethod + def _load_s3_extensions(conn: duckdb.DuckDBPyConnection) -> None: + """Load DuckDB extensions required for S3 (aws, httpfs). + + Load explicitly so DuckDB does not try to auto-install at runtime, which + can fail in environments without outbound network. Extensions should be + pre-installed at image build time (see Dockerfile.nmp-core). + """ + conn.execute("LOAD aws") + conn.execute("LOAD httpfs") + + @staticmethod + def _configure_s3_secret(conn: duckdb.DuckDBPyConnection, config: S3StorageConfig) -> None: + """Configure DuckDB S3 secret for accessing the storage backend. + + Only supports credential_chain provider (use_sdk_auth=True) since log storage + is only used with the platform's default_storage_config. + """ + # Build secret parameters - always use credential_chain + params = ["TYPE s3", "PROVIDER credential_chain"] + + if config.region: + # Escape single quotes to prevent SQL syntax issues + region = config.region.replace("'", "''") + params.append(f"REGION '{region}'") + + if config.endpoint_url: + # Extract just the host for DuckDB ENDPOINT + parsed = urlparse(config.endpoint_url) + endpoint = parsed.netloc.replace("'", "''") + params.append(f"ENDPOINT '{endpoint}'") + # OCI and some S3-compatible services need path-style URLs + params.append("URL_STYLE 'path'") + # Disable SSL for HTTP endpoints + if parsed.scheme == "http": + params.append("USE_SSL 'false'") + + secret_sql = f"CREATE OR REPLACE SECRET nmp_s3 ({', '.join(params)})" + conn.execute(secret_sql) + + @classmethod + def _build_query_path(cls, base_path: str, filters: dict[str, str]) -> str: + """Build an optimized parquet path pattern using partition filters. + + Hive partitioning uses directory structure: job=X/job_attempt=Y/job_step=Z/job_task=W/ + By pushing filters into the path, we avoid scanning unrelated partitions. + + Args: + base_path: Base logs directory path. + filters: Query filters, which may include partition and non-partition keys. + + Returns: + Optimized glob pattern for read_parquet. + """ + path_parts = [base_path] + + # Add partition filters to path in order (must be contiguous from root) + for col in cls.PARTITION_COLUMNS: + if col in filters: + path_parts.append(f"{col}={filters[col]}") + else: + # Can't skip partition levels in Hive, stop here + break + + # Add glob for remaining levels + path_parts.append("**/*.parquet") + return "/".join(path_parts) + + @staticmethod + def _ensure_single_statement(query: str) -> None: + sql = query.strip() + if ";" in sql.rstrip(" ;\n\t"): + logger.error(f"Multiple SQL statements detected in query: {sql}") + raise ValueError("Multiple SQL statements detected in query") + + @classmethod + def _prepare_query( + cls, + base_path: str, + filters: dict[str, str], + ) -> tuple[str, str, list[str]]: + if filters: + for key, value in filters.items(): + if key in cls.PARTITION_COLUMNS and not cls.SAFE_PARTITION_VALUE.fullmatch(str(value)): + raise InvalidFilterError(f"Invalid partition value: {value} for key: {key}.") + query_path = cls._build_query_path(base_path, filters) + base_table = f"read_parquet('{query_path}', hive_partitioning=1)" + + # Determine which filters were pushed into path vs need WHERE clause. + # Filters are pushed contiguously from the start of PARTITION_COLUMNS. + path_filters: set[str] = set() + for col in cls.PARTITION_COLUMNS: + if col in filters: + path_filters.add(col) + else: + break + + # Build WHERE clause for remaining filters not in path. + where_clauses: list[str] = [] + params: list[str] = [] + for key, value in filters.items(): + if key not in path_filters: + # Only allow filters that are in the LogEntry model and nothing else. + if key not in LogEntry.model_fields.keys(): + raise InvalidFilterError( + f"Invalid filter key: {key}. Allowed keys are: {LogEntry.model_fields.keys()}" + ) + where_clauses.append(f"{key} = ?") + params.append(value) + where_clause = " AND ".join(where_clauses) if where_clauses else "1=1" + return base_table, where_clause, params + + @staticmethod + def _fetch_queried_logs(result: duckdb.DuckDBPyConnection) -> list[QueriedLogEntry]: + if result.description is None: + raise RuntimeError("Cannot read description from DuckDB result") + + columns = [desc[0] for desc in result.description] + return [QueriedLogEntry.model_validate(dict(zip(columns, row))) for row in result.fetchall()] + + @staticmethod + def _platform_logs(rows: list[QueriedLogEntry]) -> list[PlatformJobLog]: + return [ + PlatformJobLog( + timestamp=row.timestamp, + job=row.job, + job_step=row.job_step, + job_task=row.job_task, + message=row.log_message, + ) + for row in rows + ] + + @staticmethod + def _query_scope_hash(filters: dict[str, str], artifact_base_path: str | None) -> str: + scope = {"artifact_base_path": artifact_base_path, "filters": filters} + encoded = json.dumps(scope, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest()[:32] + + @classmethod + def _validate_cursor_scope( + cls, cursor: LogPageCursorV1, filters: dict[str, str], artifact_base_path: str | None + ) -> None: + if cursor.query_scope_hash != cls._query_scope_hash(filters, artifact_base_path): + raise InvalidPageCursorError("page_cursor does not match the current log filters.") + + @staticmethod + def _same_boundary(row: QueriedLogEntry, cursor: LogPageCursorV1) -> bool: + return row.timestamp == cursor.boundary_timestamp and row.row_hash == cursor.boundary_row_hash + + @classmethod + def _drop_emitted_boundary_rows(cls, rows: list[QueriedLogEntry], cursor: LogPageCursorV1) -> list[QueriedLogEntry]: + remaining = cursor.emitted_boundary_rows + kept: list[QueriedLogEntry] = [] + for row in rows: + if remaining > 0 and cls._same_boundary(row, cursor): + remaining -= 1 + continue + kept.append(row) + return kept + + @classmethod + def _emitted_count_for_boundary( + cls, + rows: list[QueriedLogEntry], + prior_cursor: LogPageCursorV1 | None = None, + ) -> int: + if not rows: + return 0 + boundary = rows[0] + count = sum(1 for row in rows if row.timestamp == boundary.timestamp and row.row_hash == boundary.row_hash) + if prior_cursor is not None and cls._same_boundary(boundary, prior_cursor): + count += prior_cursor.emitted_boundary_rows + return count + + @classmethod + def _cursor_v1_from_boundary( + cls, + boundary_row: QueriedLogEntry, + filters: dict[str, str], + artifact_base_path: str | None, + emitted_boundary_rows: int, + ) -> str: + if boundary_row.row_hash is None: + raise RuntimeError("Cannot create a log cursor without a row hash.") + return LogPageCursorV1( + boundary_timestamp=boundary_row.timestamp, + boundary_row_hash=boundary_row.row_hash, + query_scope_hash=cls._query_scope_hash(filters, artifact_base_path), + emitted_boundary_rows=emitted_boundary_rows, + ).encode() + + @classmethod + def _build_tail_query(cls, base_table: str, where_clause: str) -> str: + return f""" + SELECT + {cls.SELECT_LOG_COLUMNS_SQL}, + {cls.ROW_HASH_SQL} AS row_hash, + COUNT(*) OVER() as _total_count + FROM {base_table} + WHERE {where_clause} + ORDER BY timestamp DESC, row_hash DESC + LIMIT ? + """ + + @classmethod + def _build_cursor_v1_query(cls, base_table: str, where_clause: str) -> str: + return f""" + WITH filtered AS ( + SELECT + {cls.SELECT_LOG_COLUMNS_SQL}, + {cls.ROW_HASH_SQL} AS row_hash, + COUNT(*) OVER() as _total_count + FROM {base_table} + WHERE {where_clause} + ), + eligible AS ( + SELECT * + FROM filtered + WHERE timestamp < ? OR (timestamp = ? AND row_hash <= ?) + ) + SELECT * + FROM eligible + ORDER BY timestamp DESC, row_hash DESC + LIMIT ? + """ + + @classmethod + def _query_offset_page( + cls, + base_path: str, + filters: dict[str, str], + page_size: int, + current_page: int, + direction: PaginationDirection, + storage: StorageImpl, + ) -> PlatformJobLogPage: + """Query the legacy page-number cursor mode. + + Uses a window function (COUNT(*) OVER()) to get total count in a single + query pass, avoiding the overhead of a separate COUNT query. + + Optimizes glob pattern by pushing partition filters into the path, + reducing filesystem scanning when filtering by job/attempt/step/task. + """ + conn = cls._connect(storage) + try: + base_table, where_clause, params = cls._prepare_query(base_path, filters) + + query_direction = "ASC" if direction == PaginationDirection.FORWARD else "DESC" + offset = (current_page - 1) * page_size + + # Single query with window function for total count. + # Fetch page_size + 1 to check if there are more results. + query = f""" + SELECT + {cls.SELECT_LOG_COLUMNS_SQL}, + COUNT(*) OVER() as _total_count + FROM {base_table} + WHERE {where_clause} + ORDER BY timestamp {query_direction} + LIMIT ? OFFSET ? + """ + cls._ensure_single_statement(query) + query_params: list[DuckDBQueryParam] = [*params, page_size + 1, offset] + logs = cls._fetch_queried_logs(conn.execute(query, query_params)) + + # Extract total count from first row (all rows have same _total_count). + total_count = logs[0].total_count if logs else 0 + + has_more = len(logs) > page_size + if has_more: + logs = logs[:page_size] + + log_lines = cls._platform_logs(logs) + + # Calculate pagination cursors. + next_page: str | None = None + prev_page: str | None = None + + if direction == PaginationDirection.FORWARD: + if has_more: + next_page = PageCursor(start_id=current_page + 1, direction=PaginationDirection.FORWARD).encode() + if current_page > 1: + prev_page = PageCursor(start_id=current_page - 1, direction=PaginationDirection.FORWARD).encode() + else: + if has_more: + prev_page = PageCursor( + start_id=current_page + 1, + direction=PaginationDirection.BACKWARD, + ).encode() + if current_page > 1: + next_page = PageCursor( + start_id=current_page - 1, + direction=PaginationDirection.BACKWARD, + ).encode() + + return PlatformJobLogPage( + data=log_lines, + total=total_count, + next_page=next_page, + prev_page=prev_page, + ) + finally: + conn.close() + + @classmethod + def _query_tail_window( + cls, + base_path: str, + filters: dict[str, str], + page_size: int, + storage: StorageImpl, + artifact_base_path: str | None, + ) -> PlatformJobLogPage: + conn = cls._connect(storage) + try: + base_table, where_clause, params = cls._prepare_query(base_path, filters) + query = cls._build_tail_query(base_table, where_clause) + cls._ensure_single_statement(query) + query_params: list[DuckDBQueryParam] = [*params, page_size + 1] + rows = cls._fetch_queried_logs(conn.execute(query, query_params)) + + total_count = rows[0].total_count if rows else 0 + has_previous_window = len(rows) > page_size + returned_rows = list(reversed(rows[:page_size])) + prev_page = None + if has_previous_window and returned_rows: + prev_page = cls._cursor_v1_from_boundary( + returned_rows[0], + filters, + artifact_base_path, + cls._emitted_count_for_boundary(returned_rows), + ) + + return PlatformJobLogPage( + data=cls._platform_logs(returned_rows), + total=total_count, + next_page=None, + prev_page=prev_page, + ) + finally: + conn.close() + + @classmethod + def _query_cursor_v1_window( + cls, + base_path: str, + filters: dict[str, str], + page_size: int, + cursor: LogPageCursorV1, + storage: StorageImpl, + artifact_base_path: str | None, + ) -> PlatformJobLogPage: + cls._validate_cursor_scope(cursor, filters, artifact_base_path) + conn = cls._connect(storage) + try: + base_table, where_clause, params = cls._prepare_query(base_path, filters) + query = cls._build_cursor_v1_query(base_table, where_clause) + cls._ensure_single_statement(query) + rows = cls._fetch_queried_logs( + conn.execute( + query, + [ + *params, + cursor.boundary_timestamp, + cursor.boundary_timestamp, + cursor.boundary_row_hash, + page_size + cursor.emitted_boundary_rows + 1, + ], + ) + ) + + total_count = rows[0].total_count if rows else 0 + usable_rows = cls._drop_emitted_boundary_rows(rows, cursor) + has_previous_window = len(usable_rows) > page_size + returned_rows = list(reversed(usable_rows[:page_size])) + prev_page = None + if has_previous_window and returned_rows: + prev_page = cls._cursor_v1_from_boundary( + returned_rows[0], + filters, + artifact_base_path, + cls._emitted_count_for_boundary(returned_rows, prior_cursor=cursor), + ) + + return PlatformJobLogPage( + data=cls._platform_logs(returned_rows), + total=total_count, + next_page=None, + prev_page=prev_page, + ) + finally: + conn.close() + + @classmethod + def insert_logs(cls, log_entries: list[LogEntry], base_path: str, storage: StorageImpl) -> int: + """Insert logs into Hive-partitioned parquet files. + + Uses DuckDB to write directly to the resolved storage path. ``base_path`` is + resolved by the async facade and may point at root logs or an artifact-scoped + nested logs directory. This method is synchronous; ``LogStorage`` runs it in + a thread pool to avoid blocking the event loop. + """ + conn = duckdb.connect(":memory:") + table_name = f"temp_logs_{uuid.uuid4().hex[:8]}" + + try: + from nmp.core.files.app.backends.local import LocalStorageImpl + from nmp.core.files.app.backends.s3 import S3StorageImpl + + # DuckDB COPY creates the leaf + partition dirs but not intermediate parents, so a + # nested base path (e.g. /jobs//logs) needs its parents created first. + if isinstance(storage, LocalStorageImpl): + os.makedirs(base_path, exist_ok=True) + + if isinstance(storage, S3StorageImpl): + cls._load_s3_extensions(conn) + cls._configure_s3_secret(conn, storage.config) + + import pandas as pd + + df = pd.DataFrame([entry.model_dump() for entry in log_entries]) + conn.register(table_name, df) + + insert_query = f""" + COPY ( + SELECT workspace, job, job_attempt, job_step, job_task, log_message, timestamp + FROM {table_name} + ORDER BY timestamp + ) TO '{base_path}' ( + FORMAT PARQUET, + PARTITION_BY (job, job_attempt, job_step, job_task), + APPEND + ) + """ + conn.execute(insert_query) + logger.debug(f"Successfully inserted {len(log_entries)} log entries") + return len(log_entries) + except Exception: + logger.exception("Failed to insert log entries") + raise + finally: + conn.close() diff --git a/services/core/files/src/nmp/core/files/app/log_storage.py b/services/core/files/src/nmp/core/files/app/log_storage.py index 8cae3769d2..375c3fa625 100644 --- a/services/core/files/src/nmp/core/files/app/log_storage.py +++ b/services/core/files/src/nmp/core/files/app/log_storage.py @@ -1,40 +1,15 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -""" -Log storage operations using DuckDB and StorageImpl. +"""Async log storage facade over the log repository.""" -This module provides local DuckDB-based query and insert operations for job logs, -avoiding the cross-service HTTP overhead of the previous FilesetFileSystem approach. - -NOTE: pandas and S3StorageImpl are intentionally imported inside methods -for startup performance. Do not hoist them to module level. - -TODO: Right now, this is very Jobs logs specific; in a future MR we should try to make it more generic -""" - -import logging -import os import re -import uuid -from datetime import datetime -from urllib.parse import urlparse -import duckdb from anyio import to_thread -from nmp.common.files.storage_config import S3StorageConfig -from nmp.common.jobs.schemas import ( - InvalidPageCursorError, - PageCursor, - PaginationDirection, - PlatformJobLog, - PlatformJobLogPage, -) +from nmp.common.jobs.schemas import InvalidPageCursorError, PlatformJobLogPage, decode_log_page_cursor from nmp.core.files.app.backends.base import StorageImpl -from nmp.core.files.exceptions import InvalidFilterError, InvalidPathError -from pydantic import BaseModel - -logger = logging.getLogger(__name__) +from nmp.core.files.app.log_db import DuckDBLogRepository, LogEntry, LogQuery, LogRepository +from nmp.core.files.exceptions import InvalidPathError _PATH_SEGMENT_RE = re.compile(r"^[\w\-.]+$") @@ -56,99 +31,11 @@ def logs_base_path(storage: StorageImpl, artifact_base_path: str | None = None) return storage.get_duckdb_path(f"{artifact_base_path}/logs") -class LogEntry(BaseModel): - """Internal representation of a log entry for storage.""" - - workspace: str - job: str - job_attempt: str - job_step: str - job_task: str - log_message: str - timestamp: datetime - - class LogStorage: - """Handles log storage operations using DuckDB and StorageImpl. - - This class is stateless - each operation creates its own DuckDB connection. - This ensures thread safety when operations run concurrently in the thread pool. - - For local storage: Uses direct path access with DuckDB - For S3 storage: Uses s3:// URIs with DuckDB's httpfs extension - """ - - # Partition columns in Hive directory order - PARTITION_COLUMNS = ("job", "job_attempt", "job_step", "job_task") - SAFE_PARTITION_VALUE = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") - - @staticmethod - def _load_s3_extensions(conn: duckdb.DuckDBPyConnection) -> None: - """Load DuckDB extensions required for S3 (aws, httpfs). - - Load explicitly so DuckDB does not try to auto-install at runtime, which - can fail in environments without outbound network. Extensions should be - pre-installed at image build time (see Dockerfile.nmp-core). - """ - conn.execute("LOAD aws") - conn.execute("LOAD httpfs") - - @staticmethod - def _configure_s3_secret(conn: duckdb.DuckDBPyConnection, config: S3StorageConfig) -> None: - """Configure DuckDB S3 secret for accessing the storage backend. + """Coordinates async log reads and writes through a repository implementation.""" - Only supports credential_chain provider (use_sdk_auth=True) since log storage - is only used with the platform's default_storage_config. - """ - # Build secret parameters - always use credential_chain - params = ["TYPE s3", "PROVIDER credential_chain"] - - if config.region: - # Escape single quotes to prevent SQL syntax issues - region = config.region.replace("'", "''") - params.append(f"REGION '{region}'") - - if config.endpoint_url: - # Extract just the host for DuckDB ENDPOINT - parsed = urlparse(config.endpoint_url) - endpoint = parsed.netloc.replace("'", "''") - params.append(f"ENDPOINT '{endpoint}'") - # OCI and some S3-compatible services need path-style URLs - params.append("URL_STYLE 'path'") - # Disable SSL for HTTP endpoints - if parsed.scheme == "http": - params.append("USE_SSL 'false'") - - secret_sql = f"CREATE OR REPLACE SECRET nmp_s3 ({', '.join(params)})" - conn.execute(secret_sql) - - @classmethod - def _build_query_path(cls, base_path: str, filters: dict[str, str]) -> str: - """Build an optimized parquet path pattern using partition filters. - - Hive partitioning uses directory structure: job=X/job_attempt=Y/job_step=Z/job_task=W/ - By pushing filters into the path, we avoid scanning unrelated partitions. - - Args: - base_path: Base logs directory path - filters: Query filters (may include partition and non-partition keys) - - Returns: - Optimized glob pattern for read_parquet - """ - path_parts = [base_path] - - # Add partition filters to path in order (must be contiguous from root) - for col in cls.PARTITION_COLUMNS: - if col in filters: - path_parts.append(f"{col}={filters[col]}") - else: - # Can't skip partition levels in Hive, stop here - break - - # Add glob for remaining levels - path_parts.append("**/*.parquet") - return "/".join(path_parts) + def __init__(self, repository: LogRepository | None = None) -> None: + self._repository = repository or DuckDBLogRepository() async def query_logs( self, @@ -157,242 +44,44 @@ async def query_logs( page_size: int = 100, page_cursor: str | None = None, artifact_base_path: str | None = None, + tail: int | None = None, ) -> PlatformJobLogPage: - """Query logs from parquet files using direct storage access. + """Query logs from storage. - Runs DuckDB queries in a thread pool to avoid blocking the event loop. + Runs repository work in a thread pool because the DuckDB implementation is synchronous. When ``artifact_base_path`` is set, logs are read from ``/logs`` (must match the value used at insert time). """ - direction = PaginationDirection.FORWARD - current_page = 1 + cursor = None if page_cursor: try: - cursor_obj = PageCursor.decode(page_cursor) - current_page = cursor_obj.start_id - direction = cursor_obj.direction + cursor = decode_log_page_cursor(page_cursor) except ValueError: raise InvalidPageCursorError("Invalid page cursor") - base_path = logs_base_path(storage, artifact_base_path) - - return await to_thread.run_sync( - self._query_logs_sync, - base_path, - filters or {}, - page_size, - current_page, - direction, - storage, + query = LogQuery( + base_path=logs_base_path(storage, artifact_base_path), + filters=filters or {}, + page_size=page_size, + cursor=cursor, + tail=tail, + artifact_base_path=artifact_base_path, ) - - @classmethod - def _query_logs_sync( - cls, - base_path: str, - filters: dict[str, str], - page_size: int, - current_page: int, - direction: PaginationDirection, - storage: StorageImpl, - ) -> PlatformJobLogPage: - """Synchronous log query implementation (runs in thread pool). - - Uses a window function (COUNT(*) OVER()) to get total count in a single - query pass, avoiding the overhead of a separate COUNT query. - - Optimizes glob pattern by pushing partition filters into the path, - reducing filesystem scanning when filtering by job/attempt/step/task. - """ - conn = duckdb.connect( - ":memory:", - config={ - "autoload_known_extensions": "false", - "autoinstall_known_extensions": "false", - }, - ) - try: - from nmp.core.files.app.backends.s3 import S3StorageImpl - - if isinstance(storage, S3StorageImpl): - cls._load_s3_extensions(conn) - cls._configure_s3_secret(conn, storage.config) - - if filters: - for key, value in filters.items(): - if key in cls.PARTITION_COLUMNS and not cls.SAFE_PARTITION_VALUE.fullmatch(value): - raise InvalidFilterError(f"Invalid partition value: {value} for key: {key}.") - query_path = cls._build_query_path(base_path, filters) - base_table = f"read_parquet('{query_path}', hive_partitioning=1)" - - # Determine which filters were pushed into path vs need WHERE clause - # Filters are pushed contiguously from the start of PARTITION_COLUMNS - path_filters: set[str] = set() - for col in cls.PARTITION_COLUMNS: - if col in filters: - path_filters.add(col) - else: - break - - # Build WHERE clause for remaining filters not in path - where_clauses = [] - params = [] - for key, value in filters.items(): - if key not in path_filters: - # Only allow filters that are in the LogEntry model and nothing else - if key not in LogEntry.model_fields.keys(): - raise InvalidFilterError( - f"Invalid filter key: {key}. Allowed keys are: {LogEntry.model_fields.keys()}" - ) - where_clauses.append(f"{key} = ?") - params.append(value) - where_clause = " AND ".join(where_clauses) if where_clauses else "1=1" - - query_direction = "ASC" if direction == PaginationDirection.FORWARD else "DESC" - offset = (current_page - 1) * page_size - - # Single query with window function for total count - # Fetch page_size + 1 to check if there are more results - query = f""" - SELECT - workspace, job, job_attempt, job_step, job_task, - log_message, timestamp, - COUNT(*) OVER() as _total_count - FROM {base_table} - WHERE {where_clause} - ORDER BY timestamp {query_direction} - LIMIT ? OFFSET ? - """ - sql = query.strip() - if ";" in sql.rstrip(" ;\n\t"): - logger.error(f"Multiple SQL statements detected in query: {sql}") - raise ValueError("Multiple SQL statements detected in query") - result = conn.execute(query, params + [page_size + 1, offset]) - if result.description is None: - raise RuntimeError("Cannot read description from DuckDB result") - - columns = [desc[0] for desc in result.description] - rows = result.fetchall() - logs = [dict(zip(columns, row)) for row in rows] - - # Extract total count from first row (all rows have same _total_count) - total_count = logs[0]["_total_count"] if logs else 0 - - has_more = len(logs) > page_size - if has_more: - logs = logs[:page_size] - - log_lines = [ - PlatformJobLog( - timestamp=entry["timestamp"], - job=entry["job"], - job_step=entry["job_step"], - job_task=entry["job_task"], - message=entry["log_message"], - ) - for entry in logs - ] - - # Calculate pagination cursors - next_page: str | None = None - prev_page: str | None = None - - if direction == PaginationDirection.FORWARD: - if has_more: - next_page = PageCursor(start_id=current_page + 1, direction=PaginationDirection.FORWARD).encode() - if current_page > 1: - prev_page = PageCursor(start_id=current_page - 1, direction=PaginationDirection.FORWARD).encode() - else: - if has_more: - prev_page = PageCursor( - start_id=current_page + 1, - direction=PaginationDirection.BACKWARD, - ).encode() - if current_page > 1: - next_page = PageCursor( - start_id=current_page - 1, - direction=PaginationDirection.BACKWARD, - ).encode() - - return PlatformJobLogPage( - data=log_lines, - total=total_count, - next_page=next_page, - prev_page=prev_page, - ) - - except duckdb.IOException as e: - if "No files found that match the pattern" in str(e): - return PlatformJobLogPage(data=[], total=0, next_page=None, prev_page=None) - logger.exception("IO error when querying logs") - raise - finally: - conn.close() + return await to_thread.run_sync(self._repository.query_logs, query, storage) async def insert_logs( self, storage: StorageImpl, log_entries: list[LogEntry], artifact_base_path: str | None = None ) -> int: - """Insert log entries into the Parquet storage. + """Insert log entries into storage. - Uses DuckDB to write Hive-partitioned parquet files directly to the - storage path. When ``artifact_base_path`` is set, logs nest under ``/logs``. + Runs repository work in a thread pool because the DuckDB implementation is synchronous. + When ``artifact_base_path`` is set, logs nest under ``/logs``. """ if not log_entries: return 0 base_path = logs_base_path(storage, artifact_base_path) - - return await to_thread.run_sync( - self._insert_logs_sync, - log_entries, - base_path, - storage, - ) - - @classmethod - def _insert_logs_sync(cls, log_entries: list[LogEntry], base_path: str, storage: StorageImpl) -> int: - """Synchronous log insert implementation (runs in thread pool).""" - conn = duckdb.connect(":memory:") - table_name = f"temp_logs_{uuid.uuid4().hex[:8]}" - - try: - from nmp.core.files.app.backends.local import LocalStorageImpl - from nmp.core.files.app.backends.s3 import S3StorageImpl - - # DuckDB COPY creates the leaf + partition dirs but not intermediate parents, so a - # nested base path (e.g. /jobs//logs) needs its parents created first. - if isinstance(storage, LocalStorageImpl): - os.makedirs(base_path, exist_ok=True) - - if isinstance(storage, S3StorageImpl): - cls._load_s3_extensions(conn) - cls._configure_s3_secret(conn, storage.config) - - import pandas as pd - - df = pd.DataFrame([entry.model_dump() for entry in log_entries]) - conn.register(table_name, df) - - insert_query = f""" - COPY ( - SELECT workspace, job, job_attempt, job_step, job_task, log_message, timestamp - FROM {table_name} - ORDER BY timestamp - ) TO '{base_path}' ( - FORMAT PARQUET, - PARTITION_BY (job, job_attempt, job_step, job_task), - APPEND - ) - """ - conn.execute(insert_query) - logger.debug(f"Successfully inserted {len(log_entries)} log entries") - return len(log_entries) - - except Exception: - logger.exception("Failed to insert log entries") - raise - finally: - conn.close() + return await to_thread.run_sync(self._repository.insert_logs, log_entries, base_path, storage) def dep_log_storage() -> LogStorage: diff --git a/services/core/files/tests/integration/test_otlp_endpoints.py b/services/core/files/tests/integration/test_otlp_endpoints.py index 2252bc5934..30a187372b 100644 --- a/services/core/files/tests/integration/test_otlp_endpoints.py +++ b/services/core/files/tests/integration/test_otlp_endpoints.py @@ -378,6 +378,61 @@ def test_query_empty_fileset( assert result["prev_page"] is None +def test_query_logs_accepts_tail( + client: httpx.Client, + fileset: FilesetOutput, +): + """The Files log query endpoint accepts tail without limit.""" + workspace = fileset.workspace + fileset_name = fileset.name + + response = client.post( + f"/apis/files/v2/workspaces/{workspace}/filesets/{fileset_name}/otlp/v1/logs/query", + json={"filters": {"job": "tail-test-job"}, "tail": 3}, + ) + + assert response.status_code == 200 + + +def test_query_logs_rejects_tail_with_limit( + client: httpx.Client, + fileset: FilesetOutput, +): + """Tail controls window size and cannot be combined with limit.""" + workspace = fileset.workspace + fileset_name = fileset.name + + response = client.post( + f"/apis/files/v2/workspaces/{workspace}/filesets/{fileset_name}/otlp/v1/logs/query", + json={"filters": {"job": "tail-test-job"}, "tail": 3, "limit": 3}, + ) + + assert response.status_code == 400 + assert ( + response.json()["detail"] == "tail cannot be combined with limit; tail controls the returned log window size." + ) + + +def test_query_logs_rejects_tail_with_page_cursor( + client: httpx.Client, + fileset: FilesetOutput, +): + """Tail starts at the end, so previous-page cursors must be requested without tail.""" + workspace = fileset.workspace + fileset_name = fileset.name + + response = client.post( + f"/apis/files/v2/workspaces/{workspace}/filesets/{fileset_name}/otlp/v1/logs/query", + json={"filters": {"job": "tail-test-job"}, "tail": 3, "page_cursor": "abc"}, + ) + + assert response.status_code == 400 + assert ( + response.json()["detail"] + == "tail cannot be combined with page_cursor; pass the returned prev_page as page_cursor without tail." + ) + + def test_upload_logs_missing_attributes_partial_success( client: httpx.Client, fileset: FilesetOutput, diff --git a/services/core/files/tests/test_log_storage.py b/services/core/files/tests/test_log_storage.py index 369eed5749..c697be42ae 100644 --- a/services/core/files/tests/test_log_storage.py +++ b/services/core/files/tests/test_log_storage.py @@ -7,9 +7,10 @@ import pytest from nmp.common.files.storage_config import LocalStorageConfig, S3StorageConfig -from nmp.common.jobs.schemas import PaginationDirection +from nmp.common.jobs.schemas import InvalidPageCursorError, LogPageCursorV1, PaginationDirection, decode_log_page_cursor from nmp.core.files.app.backends.local import LocalStorageImpl from nmp.core.files.app.backends.s3 import S3StorageImpl +from nmp.core.files.app.log_db import DuckDBLogRepository from nmp.core.files.app.log_storage import LogEntry, LogStorage from nmp.core.files.exceptions import InvalidFilterError, InvalidPathError @@ -250,6 +251,178 @@ async def test_query_logs_pagination(log_storage, local_storage, sample_log_entr assert page3.next_page is None +async def test_query_logs_tail_returns_latest_rows_oldest_to_newest(log_storage, local_storage, sample_log_entries): + """Tail returns the latest window in normal oldest-to-newest display order.""" + await log_storage.insert_logs(local_storage, sample_log_entries) + + result = await log_storage.query_logs( + local_storage, + filters={"job": "job-123", "job_attempt": "attempt-1"}, + page_size=3, + tail=3, + ) + + assert [log.message for log in result.data] == ["Log message 22", "Log message 23", "Log message 24"] + assert result.total == 25 + assert result.next_page is None + assert result.prev_page is not None + + +async def test_query_logs_tail_larger_than_log_returns_everything(log_storage, local_storage, sample_log_entries): + """Tail larger than the available rows returns the complete log without a previous cursor.""" + await log_storage.insert_logs(local_storage, sample_log_entries) + + result = await log_storage.query_logs( + local_storage, + filters={"job": "job-123", "job_attempt": "attempt-1"}, + page_size=100, + tail=100, + ) + + assert len(result.data) == 25 + assert result.total == 25 + assert result.next_page is None + assert result.prev_page is None + + +async def test_query_logs_prev_page_deduplicates_tail_boundary(log_storage, local_storage): + """The previous-page cursor avoids re-serving a non-unique timestamp boundary row.""" + timestamp = datetime(2026, 9, 2, 12, 0, 0) + entries = [ + LogEntry( + workspace="default", + job="j", + job_attempt="1", + job_step="s", + job_task="t", + log_message=message, + timestamp=timestamp, + ) + for message in ["a", "b", "c", "d"] + ] + await log_storage.insert_logs(local_storage, entries) + + tail_page = await log_storage.query_logs( + local_storage, + filters={"job": "j", "job_attempt": "1"}, + page_size=2, + tail=2, + ) + previous_page = await log_storage.query_logs( + local_storage, + filters={"job": "j", "job_attempt": "1"}, + page_size=2, + page_cursor=tail_page.prev_page, + ) + + combined_messages = [log.message for log in previous_page.data + tail_page.data] + assert sorted(combined_messages) == ["a", "b", "c", "d"] + assert len(combined_messages) == len(set(combined_messages)) + + +async def test_query_logs_prev_page_preserves_identical_boundary_row_count(log_storage, local_storage): + """The emitted-boundary count preserves identical rows without returning the same count twice.""" + timestamp = datetime(2026, 9, 2, 12, 0, 0) + entries = [ + LogEntry( + workspace="default", + job="j", + job_attempt="1", + job_step="s", + job_task="t", + log_message="same", + timestamp=timestamp, + ) + for _ in range(3) + ] + await log_storage.insert_logs(local_storage, entries) + + tail_page = await log_storage.query_logs( + local_storage, + filters={"job": "j", "job_attempt": "1"}, + page_size=1, + tail=1, + ) + previous_page = await log_storage.query_logs( + local_storage, + filters={"job": "j", "job_attempt": "1"}, + page_size=2, + page_cursor=tail_page.prev_page, + ) + + assert [log.message for log in previous_page.data + tail_page.data] == ["same", "same", "same"] + assert previous_page.prev_page is None + + +async def test_query_logs_prev_page_rejects_different_filters(log_storage, local_storage): + """A cursor is only valid for the effective query scope that produced it.""" + base_time = datetime(2026, 9, 2, 12, 0, 0) + entries = [ + LogEntry( + workspace="default", + job="j", + job_attempt="1", + job_step="s", + job_task="t", + log_message=f"message {i}", + timestamp=base_time + timedelta(seconds=i), + ) + for i in range(3) + ] + await log_storage.insert_logs(local_storage, entries) + + tail_page = await log_storage.query_logs( + local_storage, + filters={"job": "j", "job_attempt": "1"}, + page_size=1, + tail=1, + ) + + with pytest.raises(InvalidPageCursorError, match="page_cursor does not match the current log filters"): + await log_storage.query_logs( + local_storage, + filters={"job": "j"}, + page_size=1, + page_cursor=tail_page.prev_page, + ) + + +def test_tail_sql_does_not_use_offset(): + """Tail queries avoid the current page-number OFFSET path.""" + query = DuckDBLogRepository._build_tail_query("read_parquet('/tmp/logs/**/*.parquet')", "job = ?") + + assert "OFFSET" not in query.upper() + assert "SHA256" in query.upper() + + +def test_cursor_v1_sql_does_not_use_offset(): + """V1 cursor queries avoid the current page-number OFFSET path.""" + query = DuckDBLogRepository._build_cursor_v1_query("read_parquet('/tmp/logs/**/*.parquet')", "job = ?") + + assert "OFFSET" not in query.upper() + assert "SHA256" in query.upper() + + +async def test_tail_prev_page_cursor_uses_compact_anchor_payload(log_storage, local_storage, sample_log_entries): + """Tail previous-page cursors carry compact typed anchor fields.""" + await log_storage.insert_logs(local_storage, sample_log_entries) + + page = await log_storage.query_logs( + local_storage, + filters={"job": "job-123"}, + page_size=1, + tail=1, + ) + + assert page.prev_page is not None + cursor = decode_log_page_cursor(page.prev_page) + assert isinstance(cursor, LogPageCursorV1) + assert cursor.version == 1 + assert len(cursor.boundary_row_hash) == 32 + assert len(cursor.query_scope_hash) == 32 + assert cursor.emitted_boundary_rows == 1 + + async def test_query_logs_no_files(log_storage, local_storage): """Test querying when no log files exist.""" result = await log_storage.query_logs(local_storage) @@ -363,16 +536,16 @@ async def test_query_logs_allows_apostrophe_in_log_message_filter(log_storage, l assert result.data[0].message == "I'm testing apostrophes" -def test_query_logs_sync_rejects_multi_statement_query(local_storage, monkeypatch): +def test_log_repository_rejects_multi_statement_query(local_storage, monkeypatch): """Test that query SQL containing injected multi-statements is rejected.""" def _injected_path(cls, base_path, filters): # noqa: ARG001 return "logs'; SELECT 1; --" - monkeypatch.setattr(LogStorage, "_build_query_path", classmethod(_injected_path)) + monkeypatch.setattr(DuckDBLogRepository, "_build_query_path", classmethod(_injected_path)) with pytest.raises(ValueError, match="Multiple SQL statements detected in query"): - LogStorage._query_logs_sync( + DuckDBLogRepository._query_offset_page( base_path=local_storage.get_duckdb_path("logs"), filters={}, page_size=10, @@ -382,19 +555,21 @@ def _injected_path(cls, base_path, filters): # noqa: ARG001 ) -def test_query_logs_sync_uses_hardened_duckdb_config(local_storage, monkeypatch): +def test_log_repository_uses_hardened_duckdb_config(local_storage, monkeypatch): """Test that query connection is created with hardened DuckDB settings.""" - captured: dict[str, object] = {} + captured_database: str | None = None + captured_config: dict[str, str] | None = None - def _fake_connect(*args, **kwargs): - captured["database"] = args[0] if args else kwargs.get("database") - captured["config"] = kwargs.get("config") + def _fake_connect(database: str, *, config: dict[str, str]): + nonlocal captured_database, captured_config + captured_database = database + captured_config = config raise RuntimeError("connect intercepted") - monkeypatch.setattr("nmp.core.files.app.log_storage.duckdb.connect", _fake_connect) + monkeypatch.setattr("nmp.core.files.app.log_db.duckdb.connect", _fake_connect) with pytest.raises(RuntimeError, match="connect intercepted"): - LogStorage._query_logs_sync( + DuckDBLogRepository._query_offset_page( base_path=local_storage.get_duckdb_path("logs"), filters={}, page_size=10, @@ -403,8 +578,8 @@ def _fake_connect(*args, **kwargs): storage=local_storage, ) - assert captured["database"] == ":memory:" - assert captured["config"] == { + assert captured_database == ":memory:" + assert captured_config == { "autoload_known_extensions": "false", "autoinstall_known_extensions": "false", } @@ -459,7 +634,7 @@ def test_configure_s3_secret(self, region, endpoint_url, expected_fragments): ) mock_conn = MagicMock() - LogStorage._configure_s3_secret(mock_conn, config) + DuckDBLogRepository._configure_s3_secret(mock_conn, config) call_args = mock_conn.execute.call_args[0][0] assert "TYPE s3" in call_args 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 2fad1befba..0cb9145ef3 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 @@ -72,6 +72,8 @@ platform_config = get_platform_config() _JOB_STATUS_VALUES = ", ".join(job_status.value for job_status in PlatformJobStatus) +MAX_LOG_QUERY_LINES = 10_000 +DEFAULT_LOG_QUERY_LIMIT = 100 def _format_validation_location(loc: Any, *, prefix: str | None = None) -> str: @@ -530,14 +532,34 @@ async def page_job_logs( workspace: str, dispatcher: JobDispatcher = Depends(dep_dispatcher), logs_client: JobLogsClient = Depends(dep_job_logs_client), - limit: int = Query(default=100, description="Maximum number of logs to return", gt=0), - page_cursor: str = Query(default=None, description="Page cursor"), + limit: int | None = Query( + default=None, + description="Maximum number of logs to return", + gt=0, + le=MAX_LOG_QUERY_LINES, + json_schema_extra={"x-schema-default": DEFAULT_LOG_QUERY_LIMIT}, + ), + page_cursor: str | None = Query(default=None, description="Page cursor"), + tail: int | None = Query( + default=None, description="Number of newest log lines to return", gt=0, le=MAX_LOG_QUERY_LINES + ), attempt_id: Optional[int] = Query(default=None, description="Filter logs by job attempt ID"), step_id: Optional[str] = Query(default=None, description="Filter logs by step name"), task_id: Optional[str] = Query(default=None, description="Filter logs by task ID"), ) -> PlatformJobLogPage: """Get paginated logs for a platform job.""" with scoped_app_ctx(JobContext(id=name)): + if tail is not None and limit is not None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="tail cannot be combined with limit; tail controls the returned log window size.", + ) + if tail is not None and page_cursor is not None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="tail cannot be combined with page_cursor; pass the returned prev_page as page_cursor without tail.", + ) + job = await dispatcher.get_job(name, workspace) if not job: raise HTTPException( @@ -548,7 +570,7 @@ async def page_job_logs( try: filters = { "job": name, - "job_attempt": attempt_id if attempt_id is not None else job.attempt_id, + "job_attempt": str(attempt_id if attempt_id is not None else job.attempt_id), } if step_id: filters["job_step"] = step_id @@ -558,13 +580,14 @@ async def page_job_logs( job.fileset, workspace=workspace, filters=filters, - page_size=limit, + page_size=limit or DEFAULT_LOG_QUERY_LIMIT, page_cursor=page_cursor, + tail=tail, artifact_base_path=job_artifact_base_path(name, job.output_location), ) except InvalidPageCursorError as e: logger.error(f"Invalid page cursor: {str(e)}") - raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Invalid page cursor") + raise HTTPException(status_code=422, detail=str(e)) except Exception as e: logger.error(f"Unexpected error when querying logs: {str(e)}") raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to query job logs") diff --git a/services/core/jobs/tests/test_job_logs.py b/services/core/jobs/tests/test_job_logs.py index dadf768d14..a8f176e55c 100644 --- a/services/core/jobs/tests/test_job_logs.py +++ b/services/core/jobs/tests/test_job_logs.py @@ -108,6 +108,7 @@ async def test_get_job_logs_success( filters={"job": job.name, "job_attempt": job.attempt_id}, page_size=100, page_cursor=None, + tail=None, artifact_base_path=None, ) @@ -142,9 +143,72 @@ async def test_get_job_logs_with_pagination_params( filters={"job": job.name, "job_attempt": job.attempt_id}, page_size=2, page_cursor="some_cursor", + tail=None, artifact_base_path=None, ) + async def test_get_job_logs_with_tail( + self, test_client, dispatcher, mock_logs_client, sample_logs, sample_platform_job_request + ): + """Test job logs retrieval with tail parameter.""" + job = await dispatcher.create_job(sample_platform_job_request, DEFAULT_WORKSPACE) + + mock_logs_client.query_logs.return_value = PlatformJobLogPage( + data=sample_logs[-2:], + total=3, + next_page=None, + prev_page="previous_cursor", + ) + + response = test_client.get(f"/v2/workspaces/{DEFAULT_WORKSPACE}/jobs/{job.name}/logs?tail=2") + + assert response.status_code == 200 + data = response.json() + assert data["total"] == 3 + assert len(data["data"]) == 2 + assert data["next_page"] is None + assert data["prev_page"] == "previous_cursor" + + mock_logs_client.query_logs.assert_called_once_with( + job.fileset, + workspace=DEFAULT_WORKSPACE, + filters={"job": job.name, "job_attempt": job.attempt_id}, + page_size=100, + page_cursor=None, + tail=2, + artifact_base_path=None, + ) + + async def test_get_job_logs_rejects_tail_with_limit( + self, test_client, dispatcher, mock_logs_client, sample_platform_job_request + ): + """Tail controls returned window size and cannot be combined with limit.""" + job = await dispatcher.create_job(sample_platform_job_request, DEFAULT_WORKSPACE) + + response = test_client.get(f"/v2/workspaces/{DEFAULT_WORKSPACE}/jobs/{job.name}/logs?tail=2&limit=2") + + assert response.status_code == 400 + assert ( + response.json()["detail"] + == "tail cannot be combined with limit; tail controls the returned log window size." + ) + mock_logs_client.query_logs.assert_not_called() + + async def test_get_job_logs_rejects_tail_with_page_cursor( + self, test_client, dispatcher, mock_logs_client, sample_platform_job_request + ): + """Tail starts at the end and cannot be combined with an existing cursor.""" + job = await dispatcher.create_job(sample_platform_job_request, DEFAULT_WORKSPACE) + + response = test_client.get(f"/v2/workspaces/{DEFAULT_WORKSPACE}/jobs/{job.name}/logs?tail=2&page_cursor=abc") + + assert response.status_code == 400 + assert ( + response.json()["detail"] + == "tail cannot be combined with page_cursor; pass the returned prev_page as page_cursor without tail." + ) + mock_logs_client.query_logs.assert_not_called() + async def test_get_job_logs_scopes_to_artifact_base_path_when_output_location_set( self, test_client, dispatcher, mock_logs_client, sample_logs, sample_platform_job_request ): @@ -185,6 +249,21 @@ async def test_get_job_logs_invalid_page_cursor( assert response.status_code == 422 assert response.json()["detail"] == "Invalid page cursor" + async def test_get_job_logs_page_cursor_filter_mismatch_message( + self, test_client, dispatcher, mock_logs_client, sample_platform_job_request + ): + """Cursor mismatch errors preserve their detail for callers.""" + job = await dispatcher.create_job(sample_platform_job_request, DEFAULT_WORKSPACE) + + mock_logs_client.query_logs.side_effect = InvalidPageCursorError( + "page_cursor does not match the current log filters." + ) + + response = test_client.get(f"/v2/workspaces/{DEFAULT_WORKSPACE}/jobs/{job.name}/logs?page_cursor=old") + + assert response.status_code == 422 + assert response.json()["detail"] == "page_cursor does not match the current log filters." + def test_get_job_logs_invalid_limit(self, test_client): """Test job logs retrieval with invalid limit parameter.""" response = test_client.get(f"/v2/workspaces/{DEFAULT_WORKSPACE}/jobs/test-job/logs?limit=0") diff --git a/services/core/jobs/tests/test_jobs_client.py b/services/core/jobs/tests/test_jobs_client.py index 00661dfa64..2651d195f4 100644 --- a/services/core/jobs/tests/test_jobs_client.py +++ b/services/core/jobs/tests/test_jobs_client.py @@ -267,13 +267,22 @@ async def test_logs_round_trip( "job_task": "task-1", } assert logs_client.query_logs.await_args_list == [ - call(job.fileset, workspace="default", filters=filters, page_size=5, page_cursor=None, artifact_base_path=None), + call( + job.fileset, + workspace="default", + filters=filters, + page_size=5, + page_cursor=None, + tail=None, + artifact_base_path=None, + ), call( job.fileset, workspace="default", filters=filters, page_size=5, page_cursor="cursor-2", + tail=None, artifact_base_path=None, ), ] diff --git a/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/cli_config.yaml b/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/cli_config.yaml index a179660d57..4cd843126b 100644 --- a/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/cli_config.yaml +++ b/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/cli_config.yaml @@ -172,6 +172,8 @@ config: additional_methods: watch: override: jobs/watch.py + tail: + override: jobs/tail.py methods: create: wait: diff --git a/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/overrides/jobs/tail.py b/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/overrides/jobs/tail.py new file mode 100644 index 0000000000..d00060feb7 --- /dev/null +++ b/tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/overrides/jobs/tail.py @@ -0,0 +1,48 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import Annotated + +import typer +from nemo_platform_ext.cli.core.context import CLIContext +from nemo_platform_ext.cli.core.errors import handle_errors +from nemo_platform_ext.cli.core.help_formatter import collect_warnings +from nemo_platform_ext.cli.core.job_log_renderer import render_job_logs +from nemo_platform_plugin.client.adapter import client_from_platform +from nemo_platform_plugin.jobs.client import JobsClient +from nemo_platform_plugin.jobs.types import JobLogsQueryParams + +app = typer.Typer() # override-skip: provided by generated file + + +@app.command("tail") +@collect_warnings +@handle_errors +def tail_platform_job( + ctx: typer.Context, + name: Annotated[str, typer.Argument(help="Name of the platform job whose logs to tail")], + lines: Annotated[int, typer.Option("-n", "--lines", min=1, max=10_000, help="Number of lines to show")] = 100, + workspace: Annotated[str | None, typer.Option("--workspace", help="Workspace containing the job")] = None, + attempt_id: Annotated[int | None, typer.Option("--attempt-id", help="Filter logs to an attempt ID")] = None, + step_id: Annotated[str | None, typer.Option("--step-id", help="Filter logs to a step ID")] = None, + task_id: Annotated[str | None, typer.Option("--task-id", help="Filter logs to a task ID")] = None, +) -> None: + """Print the newest lines from a platform job log.""" + state: CLIContext = ctx.obj + client = state.get_client() + jobs_client = client_from_platform(client, JobsClient) + if workspace is None: + workspace = client._get_workspace_path_param() + + query_params: JobLogsQueryParams = {"tail": lines} + if attempt_id is not None: + query_params["attempt_id"] = attempt_id + if step_id is not None: + query_params["step_id"] = step_id + if task_id is not None: + query_params["task_id"] = task_id + + response = jobs_client.list_job_logs(workspace=workspace, name=name, query_params=query_params) + render_job_logs(response.page().items)