feat(sdk): migrate deployment logs to logs_v4 and add pod listing

Signed-off-by: Honglin Cao <hocao@nvidia.com>

Author: V2arK

README.md

@@ -53,6 +53,17 @@ Use `python examples/sdk/get_clusters.py` and
 Creating the example reserves GPU capacity and may incur usage charges. It does not
 delete the deployment automatically.
 
+### Deployment logs SDK example
+
+Logs are read per pod. Discover pod names with `get_deployment_pods()` (terminated
+pods still within log retention are included), then fetch that pod's logs oldest-first
+with `get_deployment_logs()` — as a flat list, a lazy generator (`stream=True`), or a
+live tail that keeps polling for new lines (`follow=True`):
+
+```bash
+python examples/sdk/get_deployment_logs.py
+```
+
 ### Un-installation
 
 To uninstall `centml`, simply do:

centml/sdk/api.py

@@ -1,4 +1,6 @@
+import time
 from contextlib import contextmanager
+from typing import Dict, List, Optional
 
 import platform_api_python_client
 from platform_api_python_client import (
@@ -20,6 +22,12 @@
 
 STATUS_V3_DEPLOYMENT_TYPES = {DeploymentType.INFERENCE_V3, DeploymentType.CSERVE_V3}
 
+DEFAULT_LOG_PAGE_LINES = 100  # server-side default for max_lines
+DEFAULT_LOG_POLL_INTERVAL_SECONDS = 2.0
+# The server re-delivers a ~15s look-behind window on follow polls; retain seen event
+# ids well past that so long-running follows dedupe correctly without unbounded memory.
+LOG_DEDUP_RETENTION_MS = 300_000
+
 
 class CentMLClient:
     def __init__(self, api):
@@ -187,43 +195,69 @@ def get_deployment_revisions(self, deployment_id: int):
             deployment_id=deployment_id
         ).results
 
+    def get_deployment_pods(self, deployment_id: int, revision_number: int) -> List[str]:
+        """List pods that have logged for a deployment revision, including terminated
+        pods still within log retention. A fresh deployment may return an empty list."""
+        return self._api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get(
+            deployment_id=deployment_id, revision_number=revision_number
+        ).pods
+
+    # pylint: disable=R0917
     def get_deployment_logs(
         self,
         deployment_id: int,
         revision_number: int,
-        start_time: int,
-        end_time: int,
-        line_count: int = 100,
-        start_from_head: bool = True,
+        pod: str,
+        max_lines: int = DEFAULT_LOG_PAGE_LINES,
         stream: bool = False,
+        follow: bool = False,
+        poll_interval: float = DEFAULT_LOG_POLL_INTERVAL_SECONDS,
     ):
-        """Fetch logs for a deployment within a time window, handling pagination automatically.
+        """Fetch one pod's logs oldest-first, handling pagination and deduplication automatically.
 
-        start_time and end_time are Unix timestamps in milliseconds.
-        Use get_deployment_revisions() to find the current revision number.
+        Use get_deployment_pods() to discover pod names and get_deployment_revisions()
+        to find the current revision number.
 
         If stream=True, returns a generator that yields events as each page is fetched.
-        If stream=False (default), returns a flat list of all events.
+        If follow=True, returns an endless generator that keeps polling for new lines
+        every poll_interval seconds (implies streaming); rare late-arriving lines may
+        then be yielded slightly out of timestamp order, as with `kubectl logs -f`.
+        If neither (default), returns a flat list of all events logged so far.
         """
 
         def _iter_events():
-            next_page_token = None
+            seen_event_ids: Dict[str, int] = {}
+            newest_timestamp: Optional[int] = None
             while True:
-                response = self._api.get_deployment_logs_v3_deployments_logs_v3_deployment_id_revision_number_get(
+                response = self._api.get_deployment_logs_v4_logs_deployment_id_revision_number_get(
                     deployment_id=deployment_id,
                     revision_number=revision_number,
-                    start_time=start_time,
-                    end_time=end_time,
-                    next_page_token=next_page_token,
-                    start_from_head=start_from_head,
-                    line_count=line_count,
+                    pod=pod,
+                    fetch_newer=True,
+                    timestamp=newest_timestamp,
+                    max_lines=max_lines,
                 )
-                yield from response.events
-                next_page_token = response.next_page_token
-                if not next_page_token:
+                # Follow polls re-deliver a look-behind window of already-seen lines
+                # (late-arrival protection); the event id is the deduplication key.
+                has_newer_events = False
+                for event in response.events:
+                    if event.id in seen_event_ids:
+                        continue
+                    seen_event_ids[event.id] = event.timestamp
+                    if newest_timestamp is None or event.timestamp > newest_timestamp:
+                        has_newer_events = True
+                    yield event
+                if has_newer_events:
+                    newest_timestamp = max(seen_event_ids.values())
+                    cutoff = newest_timestamp - LOG_DEDUP_RETENTION_MS
+                    seen_event_ids = {i: t for i, t in seen_event_ids.items() if t >= cutoff}
+                elif follow:
+                    time.sleep(poll_interval)
+                else:
+                    # No event newer than the boundary: history is exhausted.
                     break
 
-        if stream:
+        if stream or follow:
             return _iter_events()
 
         return list(_iter_events())

examples/sdk/get_deployment_logs.py

@@ -1,69 +1,42 @@
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timezone
 
 from centml.sdk.api import get_centml_client
 
 # --- Configuration ---
 DEPLOYMENT_ID = 1234  # Replace with your deployment ID
 REVISION_NUMBER = 10
-HOURS_BACK = 1  # Fetch logs from the last N hours
+FOLLOW = False  # Set True to keep polling for new log lines (like `kubectl logs -f`)
 
 
-def format_event(event: dict) -> str:
-    timestamp_ms = (
-        event.get("timestamp")
-        or event.get("time")
-        or event.get("ts")
-        or ""
-    )
-    message = (
-        event.get("message")
-        or event.get("msg")
-        or event.get("log")
-        or str(event)
-    )
-    if timestamp_ms:
-        ts = datetime.fromtimestamp(int(timestamp_ms) / 1000, tz=timezone.utc).isoformat()
-        return f"[{ts}] {message}"
-    return message
+def format_event(event) -> str:
+    ts = datetime.fromtimestamp(event.timestamp / 1000, tz=timezone.utc).isoformat()
+    return f"[{ts}] {event.message}"
 
 
 def main():
-    stream = True
-    end_time = int(datetime.now(timezone.utc).timestamp() * 1000)
-    start_time = end_time - int(timedelta(hours=HOURS_BACK).total_seconds() * 1000)
-
-    print(f"Fetching logs for deployment {DEPLOYMENT_ID}")
-    print(
-        f"Time window: "
-        f"{datetime.fromtimestamp(start_time / 1000, tz=timezone.utc).isoformat()} → "
-        f"{datetime.fromtimestamp(end_time / 1000, tz=timezone.utc).isoformat()}"
-    )
-    print()
-
     with get_centml_client() as cclient:
-        if stream:
-            # Streaming: print events as each page arrives
+        # Logs are read per pod: discover the pods that have logged for this revision
+        # (terminated pods within log retention are included).
+        pods = cclient.get_deployment_pods(DEPLOYMENT_ID, REVISION_NUMBER)
+        if not pods:
+            print("No pods have logged for this revision yet.")
+            return
+
+        pod = pods[0]
+        print(f"Fetching logs for deployment {DEPLOYMENT_ID} revision {REVISION_NUMBER}, pod {pod}\n")
+
+        if FOLLOW:
+            # Endless generator: yields history oldest-first, then keeps polling for new lines
             for event in cclient.get_deployment_logs(
-                deployment_id=DEPLOYMENT_ID,
-                revision_number=REVISION_NUMBER,
-                start_time=start_time,
-                end_time=end_time,
-                start_from_head=False,
-                stream=stream,
+                deployment_id=DEPLOYMENT_ID, revision_number=REVISION_NUMBER, pod=pod, follow=True
             ):
                 print(format_event(event))
         else:
-            # Batch: collect all events then process
-            events = cclient.get_deployment_logs(
-                deployment_id=DEPLOYMENT_ID,
-                revision_number=REVISION_NUMBER,
-                start_time=start_time,
-                end_time=end_time,
-                start_from_head=False,
-            )
+            # Batch: collect all events logged so far, oldest first
+            events = cclient.get_deployment_logs(deployment_id=DEPLOYMENT_ID, revision_number=REVISION_NUMBER, pod=pod)
 
             if not events:
-                print("No logs found in the given time window.")
+                print("No logs found for this pod.")
                 return
 
             print(f"Found {len(events)} log entries:\n")

tests/test_sdk_api.py

@@ -1,4 +1,6 @@
+from itertools import islice
 from types import SimpleNamespace
+from typing import Generator
 from unittest.mock import MagicMock, patch
 
 import platform_api_python_client
@@ -209,3 +211,150 @@ def test_delete_hardware_instance_delegates_to_platform_client():
 
     assert response is expected_response
     api.delete_hardware_instance_hardware_instances_hardware_instance_id_delete.assert_called_once_with(123)
+
+
+def _log_event(event_id, timestamp, message="line"):
+    return SimpleNamespace(id=event_id, timestamp=timestamp, message=message)
+
+
+def _log_page(*events):
+    return SimpleNamespace(events=list(events))
+
+
+def test_generated_client_exposes_logs_v4_contract():
+    assert hasattr(
+        platform_api_python_client.EXTERNALApi, "get_deployment_logs_v4_logs_deployment_id_revision_number_get"
+    )
+    assert hasattr(
+        platform_api_python_client.EXTERNALApi, "get_deployment_pods_deployments_pods_deployment_id_revision_number_get"
+    )
+
+
+def test_get_deployment_pods_returns_pod_names():
+    api = MagicMock()
+    api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get.return_value = SimpleNamespace(
+        pods=["pod-a", "pod-b"]
+    )
+    client = CentMLClient(api)
+
+    assert client.get_deployment_pods(123, 2) == ["pod-a", "pod-b"]
+
+    api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get.assert_called_once_with(
+        deployment_id=123, revision_number=2
+    )
+
+
+def test_get_deployment_logs_paginates_oldest_first_until_no_newer_events():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [
+        _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)),
+        _log_page(_log_event("3-c", 3000)),
+        _log_page(),
+    ]
+    client = CentMLClient(api)
+
+    events = client.get_deployment_logs(123, 2, pod="pod-a")
+
+    assert [e.id for e in events] == ["1-a", "2-b", "3-c"]
+    calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list
+    assert len(calls) == 3
+    # First page scans from the head of the log window; later pages pass the newest
+    # seen timestamp verbatim (the boundary is exclusive server-side).
+    assert [c.kwargs["timestamp"] for c in calls] == [None, 2000, 3000]
+    assert all(c.kwargs["fetch_newer"] is True for c in calls)
+    assert all(c.kwargs["deployment_id"] == 123 and c.kwargs["revision_number"] == 2 for c in calls)
+    assert all(c.kwargs["pod"] == "pod-a" for c in calls)
+
+
+def test_get_deployment_logs_returns_empty_list_when_pod_has_no_logs():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page()
+    client = CentMLClient(api)
+
+    assert client.get_deployment_logs(123, 2, pod="pod-a") == []
+
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_called_once()
+
+
+def test_get_deployment_logs_deduplicates_look_behind_redelivery():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [
+        _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)),
+        # Look-behind re-delivers 2000 alongside fresh lines; a late arrival shows up at 1500.
+        _log_page(_log_event("15-l", 1500), _log_event("2-b", 2000), _log_event("3-c", 3000)),
+        # Re-delivery-only page: no event newer than the boundary means history is done.
+        _log_page(_log_event("3-c", 3000)),
+    ]
+    client = CentMLClient(api)
+
+    events = client.get_deployment_logs(123, 2, pod="pod-a")
+
+    assert [e.id for e in events] == ["1-a", "2-b", "15-l", "3-c"]
+    assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 3
+
+
+def test_get_deployment_logs_passes_max_lines_through():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page()
+    client = CentMLClient(api)
+
+    client.get_deployment_logs(123, 2, pod="pod-a", max_lines=7)
+
+    call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args
+    assert call.kwargs["max_lines"] == 7
+
+
+def test_get_deployment_logs_stream_fetches_pages_lazily():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [
+        _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)),
+        _log_page(_log_event("3-c", 3000)),
+        _log_page(),
+    ]
+    client = CentMLClient(api)
+
+    events = client.get_deployment_logs(123, 2, pod="pod-a", stream=True)
+
+    assert isinstance(events, Generator)
+    assert [e.id for e in islice(events, 2)] == ["1-a", "2-b"]
+    assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 1
+    assert [e.id for e in events] == ["3-c"]
+    assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 3
+
+
+def test_get_deployment_logs_follow_polls_for_new_events_and_dedupes():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [
+        _log_page(_log_event("1-a", 1000)),
+        _log_page(),  # nothing new yet -> sleep, poll again with the same boundary
+        _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)),
+    ]
+    client = CentMLClient(api)
+
+    with patch("centml.sdk.api.time.sleep") as mock_sleep:
+        events = client.get_deployment_logs(123, 2, pod="pod-a", follow=True, poll_interval=0.5)
+        assert [e.id for e in islice(events, 2)] == ["1-a", "2-b"]
+
+    calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list
+    assert [c.kwargs["timestamp"] for c in calls] == [None, 1000, 1000]
+    mock_sleep.assert_called_once_with(0.5)
+
+
+def test_get_deployment_logs_follow_sleeps_on_redelivery_only_page():
+    api = MagicMock()
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [
+        _log_page(_log_event("1-a", 1000)),
+        # Look-behind re-delivery with nothing new: the boundary must not advance
+        # and the poll must sleep instead of spinning.
+        _log_page(_log_event("1-a", 1000)),
+        _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)),
+    ]
+    client = CentMLClient(api)
+
+    with patch("centml.sdk.api.time.sleep") as mock_sleep:
+        events = client.get_deployment_logs(123, 2, pod="pod-a", follow=True, poll_interval=0.5)
+        assert [e.id for e in islice(events, 2)] == ["1-a", "2-b"]
+
+    calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list
+    assert [c.kwargs["timestamp"] for c in calls] == [None, 1000, 1000]
+    mock_sleep.assert_called_once_with(0.5)