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

Adds get_deployment_pods, a page-based get_deployment_logs anchored on
previously fetched events or an epoch-ms boundary (before/after), a
get_deployment_logs_range time-window read that can merge all pods of a
revision, and a stateful DeploymentLogSession for backfill, tailing, and
cross-process resume.

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

Author: V2arK

README.md

@@ -53,6 +53,22 @@ 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 read with a
+`deployment_log_session()`: `fetch_older()` pages toward the beginning of history and
+`fetch_newer()` returns only new lines, while the session keeps the merged, ordered
+log in `.events`. `get_deployment_logs_range()` fetches a specific time window
+(epoch-millisecond bounds, both optional) and, with `pod=None`, merges every pod's
+stream chronologically. The same paging is available statelessly through
+`get_deployment_logs(before=..., after=...)`, anchored on events you already hold
+or on a bare epoch-millisecond boundary:
+
+```bash
+python examples/sdk/get_deployment_logs.py
+```
+
 ### Un-installation
 
 To uninstall `centml`, simply do:

centml/sdk/api.py

@@ -1,4 +1,7 @@
+from bisect import insort
 from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import List, Optional, Union
 
 import platform_api_python_client
 from platform_api_python_client import (
@@ -21,6 +24,23 @@
 
 STATUS_V3_DEPLOYMENT_TYPES = {DeploymentType.INFERENCE_V3, DeploymentType.CSERVE_V3}
 
+DEFAULT_LOG_PAGE_LINES = 100  # server-side default for max_lines
+MAX_LOG_PAGE_LINES = 5000  # server-side ceiling for max_lines
+# The server re-delivers a ~15s look-behind window on fetch-newer requests; only the
+# caller's events within this generous margin of the boundary can be re-delivered.
+LOG_DEDUP_RETENTION_MS = 300_000
+
+
+@dataclass(frozen=True)
+class DeploymentLogEvent:
+    """One log line with its pod attached — logs_v4 events carry no pod name, so
+    merged multi-pod views need the SDK to attribute each line itself."""
+
+    id: str
+    timestamp: int
+    message: str
+    pod: str
+
 
 class CentMLClient:
     def __init__(self, api):
@@ -191,46 +211,175 @@ 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,
-        stream: bool = False,
-    ):
-        """Fetch logs for a deployment within a time window, handling pagination automatically.
-
-        start_time and end_time are Unix timestamps in milliseconds.
-        Use 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.
+        pod: str,
+        before: Optional[Union[list, int]] = None,
+        after: Optional[Union[list, int]] = None,
+        max_lines: int = DEFAULT_LOG_PAGE_LINES,
+    ) -> list:
+        """Fetch one page of a pod's logs, oldest-first. Use get_deployment_pods() to
+        discover pod names and get_deployment_revisions() for the revision number.
+
+        before and after anchor the page to events a previous call returned for the
+        same pod (pass your accumulated list; only the relevant boundary is used):
+          - neither: the newest page (tail).
+          - before=<events>: the page strictly older than the oldest of them;
+            an empty result means the beginning of history is reached.
+          - after=<events>: lines strictly newer than the newest of them; an empty
+            result means nothing new yet — call again later to keep tailing. Late
+            lines still landing near that boundary are included on top of max_lines
+            and may sort below events you already hold (order by id if that matters).
+        Either anchor also accepts a bare epoch-millisecond int as the (exclusive)
+        boundary itself; an int after anchor holds no event ids, so the re-delivered
+        span at the boundary comes through undeduplicated.
         """
+        if before is not None and after is not None:
+            raise ValueError("before and after are mutually exclusive")
+
+        fetch_newer = after is not None
+        anchor = after if fetch_newer else before
+        anchor_events = None
+        boundary_timestamp = None
+        if isinstance(anchor, int):
+            boundary_timestamp = anchor
+        elif anchor:
+            anchor_events = anchor
+            timestamps = [event.timestamp for event in anchor_events]
+            boundary_timestamp = max(timestamps) if fetch_newer else min(timestamps)
+
+        response = self._api.get_deployment_logs_v4_logs_deployment_id_revision_number_get(
+            deployment_id=deployment_id,
+            revision_number=revision_number,
+            pod=pod,
+            fetch_newer=fetch_newer,
+            timestamp=boundary_timestamp,
+            max_lines=max_lines,
+        )
+        if not fetch_newer or not anchor_events:
+            return response.events
+
+        # fetch_newer re-delivers a look-behind window at and before the boundary
+        # (late-arrival protection); drop the lines the caller already holds by id.
+        cutoff = max(event.timestamp for event in anchor_events) - LOG_DEDUP_RETENTION_MS
+        held_event_ids = {event.id for event in anchor_events if event.timestamp >= cutoff}
+        return [event for event in response.events if event.id not in held_event_ids]
 
-        def _iter_events():
-            next_page_token = None
+    # pylint: disable=R0917
+    def get_deployment_logs_range(
+        self,
+        deployment_id: int,
+        revision_number: int,
+        pod: Optional[str] = None,
+        start_time: Optional[int] = None,
+        end_time: Optional[int] = None,
+    ) -> List[DeploymentLogEvent]:
+        """Fetch every log line in [start_time, end_time] (epoch ms, inclusive; both
+        optional — an open end reads to the beginning or the present), oldest first.
+        pod=None reads all pods of the revision and merges the streams
+        chronologically; each returned event carries its pod name."""
+        if start_time is not None and end_time is not None and start_time > end_time:
+            raise ValueError("start_time must not exceed end_time")
+
+        pods = [pod] if pod is not None else self.get_deployment_pods(deployment_id, revision_number)
+        merged: List[DeploymentLogEvent] = []
+        for pod_name in pods:
+            events: list = []
             while True:
-                response = self._api.get_deployment_logs_v3_deployments_logs_v3_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,
+                # after is exclusive, so start_time - 1 admits lines at start_time itself;
+                # start_time 0 (or None) means the whole window — scan from the head.
+                anchor: Union[list, int] = events if events else (start_time - 1 if start_time else [])
+                page = self.get_deployment_logs(
+                    deployment_id, revision_number, pod_name, after=anchor, max_lines=MAX_LOG_PAGE_LINES
                 )
-                yield from response.events
-                next_page_token = response.next_page_token
-                if not next_page_token:
+                if not page:
                     break
+                events += page
+                if end_time is not None and page[-1].timestamp > end_time:
+                    break
+            merged += [
+                DeploymentLogEvent(id=event.id, timestamp=event.timestamp, message=event.message, pod=pod_name)
+                for event in events
+                if (start_time is None or event.timestamp >= start_time)
+                and (end_time is None or event.timestamp <= end_time)
+            ]
+        merged.sort(key=lambda event: event.id)
+        return merged
+
+    def deployment_log_session(
+        self, deployment_id: int, revision_number: int, pod: str, events: Optional[list] = None
+    ) -> "DeploymentLogSession":
+        """Stateful reader for one pod's logs that tracks fetched pages and anchors
+        every request itself — see DeploymentLogSession. Seed events with logs a
+        previous session (or get_deployment_logs) returned for the same pod."""
+        return DeploymentLogSession(self, deployment_id, revision_number, pod, events)
+
+
+class DeploymentLogSession:
+    """Maintains a contiguous, ordered window of one pod's logs across fetches.
+
+    Every fetch is anchored on the window itself, so pages can never overlap or
+    leave gaps inside it (within log retention; an undetectable gap forms if the
+    session idles past retention before fetching newer lines).
+    """
 
-        if stream:
-            return _iter_events()
-
-        return list(_iter_events())
+    # pylint: disable=R0917
+    def __init__(self, client: CentMLClient, deployment_id: int, revision_number: int, pod: str, events=None):
+        self._client = client
+        self._deployment_id = deployment_id
+        self._revision_number = revision_number
+        self._pod = pod
+        # Seeded events come from outside the session: canonicalize to unique ids in
+        # chronological order (id order == time order at nanosecond precision).
+        unique_events = {event.id: event for event in events or []}
+        self._events = [unique_events[event_id] for event_id in sorted(unique_events)]
+
+    @property
+    def events(self) -> list:
+        """Copy of the window fetched so far, oldest first. Complete from the beginning
+        of history only once fetch_older() has returned an empty list."""
+        return list(self._events)
+
+    def fetch_older(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list:
+        """Fetch the page older than the window and prepend it; on an empty session
+        fetches the newest page (tail). Returns the page; empty list = no older
+        lines exist (yet)."""
+        page = self._client.get_deployment_logs(
+            self._deployment_id, self._revision_number, self._pod, before=self._events, max_lines=max_lines
+        )
+        self._events[:0] = page
+        return page
+
+    def fetch_newer(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list:
+        """Fetch lines newer than the window and merge them in; on an empty session
+        fetches the newest page (tail) — to read from the beginning of history
+        instead, loop fetch_older() until it returns an empty list. Returns only
+        the new lines; empty list = nothing new yet, call again later to keep
+        tailing. Rare late arrivals sort into the window below its newest lines."""
+        if not self._events:
+            return self.fetch_older(max_lines=max_lines)
+        delta = self._client.get_deployment_logs(
+            self._deployment_id, self._revision_number, self._pod, after=self._events, max_lines=max_lines
+        )
+        for event in delta:
+            if event.id > self._events[-1].id:
+                self._events.append(event)
+            else:
+                # A late arrival may even precede the window's oldest line (tail page
+                # cut inside the look-behind span); the server delivers that span
+                # completely on top of max_lines, so the window stays contiguous.
+                insort(self._events, event, key=lambda held: held.id)
+        return delta
 
 
 @contextmanager

examples/sdk/get_deployment_logs.py

@@ -1,74 +1,58 @@
-from datetime import datetime, timezone, timedelta
+import time
+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
+TAIL_SECONDS = 30  # How long to keep polling for new lines after reading history
 
 
-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
-            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,
-            ):
-                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,
-            )
-
-            if not events:
-                print("No logs found in the given time window.")
-                return
-
-            print(f"Found {len(events)} log entries:\n")
-            for event in events:
+        # 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"Reading logs for deployment {DEPLOYMENT_ID} revision {REVISION_NUMBER}, pod {pod}\n")
+
+        # The session tracks what it has fetched and anchors every request itself.
+        session = cclient.deployment_log_session(DEPLOYMENT_ID, REVISION_NUMBER, pod)
+
+        # Read the full history: newest page first, then page back to the beginning.
+        while session.fetch_older():
+            pass
+        print(f"Found {len(session.events)} log entries:\n")
+        for event in session.events:
+            print(format_event(event))
+
+        # Keep tailing: each call returns only the lines the session does not hold yet.
+        print(f"\nPolling for new lines for {TAIL_SECONDS}s...")
+        deadline = time.monotonic() + TAIL_SECONDS
+        while time.monotonic() < deadline:
+            for event in session.fetch_newer():
                 print(format_event(event))
+            time.sleep(2)
+
+        # The same paging is available statelessly via get_deployment_logs, anchored
+        # on events you already hold — useful when you manage storage yourself:
+        #   page  = cclient.get_deployment_logs(DEPLOYMENT_ID, REVISION_NUMBER, pod=pod)  # tail
+        #   older = cclient.get_deployment_logs(..., pod=pod, before=page)  # [] = beginning
+        #   newer = cclient.get_deployment_logs(..., pod=pod, after=page)   # [] = nothing new
+        # A specific time window (all pods merged, oldest first, pod on each event):
+        #   window = cclient.get_deployment_logs_range(
+        #       DEPLOYMENT_ID, REVISION_NUMBER, start_time=t1_ms, end_time=t2_ms
+        #   )
 
 
 if __name__ == "__main__":