[SDK] Consolidate deployment log reading into fetch_logs (#152)

Deployment log reading was spread over three stateful readers, none of
which could stream one pod's logs lazily with bounded memory. Tailing a
running deployment meant re-creating a reader over overlapping windows and
keeping a seen dict in user code to drop the server's late-arrival
re-deliveries.

fetch_logs(deployment_id, revision_number, pod, start_time=None,
end_time=None, chunk_size=DEFAULT_LOG_CHUNK_LINES) replaces all three. pod
is required; discover names with get_deployment_pods(). start_time defaults
to the current time, resolved when fetch_logs is called rather than at the
first next(), so lines logged while the generator sits unstarted are not
skipped. chunk_size is the number of lines requested per round trip, so it
inherits the server's 1..5000 ceiling and is validated at the call.

With end_time set the iterator terminates once the window is delivered or
the store has no more lines to give. Without it the iterator never
terminates: once caught up it yields an empty chunk each time nothing new
is stored yet, and the caller decides when to sleep or break. The dedup
anchor lives inside the generator for its whole lifetime, holding only ids
and timestamps within the dedup window, so every stored line is delivered
at most once and callers need no cross-call bookkeeping.

Each server page that carries window lines becomes one chunk, ascending by
timestamp and id. A page the store answers as busy is retried in place with
exponential backoff and jitter: there is no server-issued cursor, so a page
request is a pure function of its anchor and re-issuing it can neither
duplicate nor skip lines.

get_deployment_logs, get_deployment_logs_range, deployment_log_session and
the DeploymentLogSession class carry typing_extensions.deprecated and keep
working unchanged. typing-extensions becomes a declared dependency.

The README is one paragraph plus one example each for a window read and a
tail, and examples/sdk/get_deployment_logs.py shows the same two shapes over
a single generator.

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

Author: V2arK

README.md

@@ -55,20 +55,45 @@ 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:
+`fetch_logs()` is the one way to read deployment logs: it fetches one pod's stored
+log lines within a time window (`start_time`/`end_time`, epoch ms, inclusive) and
+yields them lazily, oldest first, as chunks of `DeploymentLogEvent` — each line at
+most once, holding only a short dedup window in memory however long the stream.
+`chunk_size` (1 to 5000) is the number of lines requested from the server per
+round trip and each server page that carries window lines becomes one chunk, so
+a bulk read of history wants a large `chunk_size` — the log read path is
+rate-limited upstream, and a small `chunk_size` over a large window multiplies
+requests. Discover pod names with `get_deployment_pods()`
+(terminated pods still within log retention are included). `start_time` defaults
+to the moment of the call; with `end_time` set the iterator terminates once the
+window is delivered or the store has no more lines to give, whichever comes
+first:
+
+```python
+for chunk in cclient.fetch_logs(DEPLOYMENT_ID, REVISION, pod, start_time=t1_ms, end_time=t2_ms):
+    for event in chunk:
+        print(event.message)
+```
 
-```bash
-python examples/sdk/get_deployment_logs.py
+Without `end_time` the same generator tails: it never terminates, and once caught
+up it yields an empty chunk each time nothing new is stored yet — the caller
+decides when to sleep or break:
+
+```python
+import time
+
+for chunk in cclient.fetch_logs(DEPLOYMENT_ID, REVISION, pod):
+    if not chunk:
+        time.sleep(2)
+        continue
+    for event in chunk:
+        print(event.message)
 ```
 
+`python examples/sdk/get_deployment_logs.py` runs both. `get_deployment_logs()`,
+`get_deployment_logs_range()` and `deployment_log_session()` still work but are
+deprecated in favor of `fetch_logs()` and emit a `DeprecationWarning` on use.
+
 ### Un-installation
 
 To uninstall `centml`, simply do:

centml/sdk/api.py

@@ -1,7 +1,10 @@
+import random
+import time
 from bisect import insort
 from contextlib import contextmanager
 from dataclasses import dataclass
-from typing import List, Optional, Union
+from functools import partial
+from typing import Callable, Iterator, List, Optional, Union
 
 import platform_api_python_client
 from platform_api_python_client import (
@@ -18,6 +21,7 @@
     InviteUserRequest,
     Metric,
 )
+from typing_extensions import deprecated
 
 from centml.sdk import auth
 from centml.sdk.config import settings
@@ -26,9 +30,39 @@
 
 DEFAULT_LOG_PAGE_LINES = 100  # server-side default for max_lines
 MAX_LOG_PAGE_LINES = 5000  # server-side ceiling for max_lines
+# fetch_logs asks for less than the server's own default because the common call tails a
+# running deployment, where the store has only a handful of new lines to hand over per poll
+# whatever the page size is. Reading a backlog wants a far larger chunk_size.
+DEFAULT_LOG_CHUNK_LINES = 10
 # 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
+# The log read path is rate limited upstream on a bucket shared by every caller, and the
+# API reports a saturated bucket the same way it reports a sick store: HTTP 503.
+LOG_BUSY_STATUS = 503
+# Doubling from half a second spends about seven seconds over these attempts — long enough
+# to outlast a bucket refill without looking hung.
+LOG_RETRY_ATTEMPTS = 5
+LOG_RETRY_BASE_SECONDS = 0.5
+# Fraction of each backoff to vary it by: the bucket is shared, so clients backing off in
+# lockstep would re-collide on it every round.
+LOG_RETRY_JITTER = 0.25
+
+
+def _with_busy_retry(fetch_page: Callable[[], list]) -> list:
+    """Call fetch_page, retrying while the store reports itself busy. There is no
+    server-issued cursor, so a page request is a pure function of its anchor and
+    re-issuing it can neither duplicate nor skip lines."""
+    for attempt in range(LOG_RETRY_ATTEMPTS - 1):
+        try:
+            return fetch_page()
+        except ApiException as exc:
+            if exc.status != LOG_BUSY_STATUS:
+                raise
+            backoff = LOG_RETRY_BASE_SECONDS * 2**attempt
+            time.sleep(backoff * (1 + random.uniform(-LOG_RETRY_JITTER, LOG_RETRY_JITTER)))
+    # The last attempt propagates whatever it raises.
+    return fetch_page()
 
 
 def _recent_anchor(events: list) -> list:
@@ -42,6 +76,16 @@ def _recent_anchor(events: list) -> list:
     return events[first_recent:]
 
 
+@dataclass(frozen=True)
+class _LogAnchor:
+    """The two fields an after anchor uses: the id it dedupes by and the timestamp it
+    takes its boundary and its retention cutoff from. Holding these instead of whole
+    events keeps a long tail off the message text it has already handed out."""
+
+    id: str
+    timestamp: int
+
+
 @dataclass(frozen=True)
 class DeploymentLogEvent:
     """One log line with its pod attached — logs_v4 events carry no pod name, so
@@ -230,6 +274,7 @@ def get_deployment_pods(self, deployment_id: int, revision_number: int) -> List[
         ).pods
 
     # pylint: disable=R0917
+    @deprecated("get_deployment_logs() is deprecated; use fetch_logs() instead")
     def get_deployment_logs(
         self,
         deployment_id: int,
@@ -239,7 +284,9 @@ def get_deployment_logs(
         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
+        """Deprecated: use fetch_logs() instead.
+
+        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
@@ -257,6 +304,22 @@ def get_deployment_logs(
         through undeduplicated. An empty anchor list raises ValueError. Pages never
         split a millisecond, so a delivered boundary millisecond is always complete.
         """
+        return self._fetch_log_page(
+            deployment_id, revision_number, pod, before=before, after=after, max_lines=max_lines
+        )
+
+    # pylint: disable=R0917
+    def _fetch_log_page(
+        self,
+        deployment_id: int,
+        revision_number: int,
+        pod: str,
+        before: Optional[Union[list, int]] = None,
+        after: Optional[Union[list, int]] = None,
+        max_lines: int = DEFAULT_LOG_PAGE_LINES,
+    ) -> list:
+        """The page primitive behind fetch_logs and the deprecated readers — the
+        contract get_deployment_logs() documents, without the deprecation warning."""
         if before is not None and after is not None:
             raise ValueError("before and after are mutually exclusive")
 
@@ -294,6 +357,7 @@ def get_deployment_logs(
         return [event for event in response.events if event.id not in held_event_ids]
 
     # pylint: disable=R0917
+    @deprecated("get_deployment_logs_range() is deprecated; use fetch_logs() instead")
     def get_deployment_logs_range(
         self,
         deployment_id: int,
@@ -302,7 +366,9 @@ def get_deployment_logs_range(
         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
+        """Deprecated: use fetch_logs() instead.
+
+        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."""
@@ -317,7 +383,7 @@ def get_deployment_logs_range(
                 # 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] = _recent_anchor(events) if events else (start_time - 1 if start_time else 0)
-                page = self.get_deployment_logs(
+                page = self._fetch_log_page(
                     deployment_id, revision_number, pod_name, after=anchor, max_lines=MAX_LOG_PAGE_LINES
                 )
                 if not page:
@@ -334,18 +400,156 @@ def get_deployment_logs_range(
         merged.sort(key=lambda event: event.id)
         return merged
 
+    # pylint: disable=R0917
+    def fetch_logs(
+        self,
+        deployment_id: int,
+        revision_number: int,
+        pod: str,
+        start_time: Optional[int] = None,
+        end_time: Optional[int] = None,
+        chunk_size: int = DEFAULT_LOG_CHUNK_LINES,
+    ) -> Iterator[List[DeploymentLogEvent]]:
+        """Fetch one pod's stored log lines within [start_time, end_time] (epoch ms,
+        inclusive), yielded lazily oldest first as chunks of DeploymentLogEvent,
+        each stored line at most once. Discover pod names with
+        get_deployment_pods().
+
+        chunk_size is the number of lines requested from the server per round
+        trip (1 to MAX_LOG_PAGE_LINES), and each server page that carries window
+        lines becomes one yielded chunk, so a bulk read of history wants a large
+        chunk_size — the log read path is rate-limited upstream, and a small
+        chunk_size over a large window multiplies requests. A chunk usually holds
+        up to chunk_size lines but can be smaller (lines below start_time or
+        already delivered are filtered out of the page) or larger (the server
+        never splits one millisecond across pages, so a millisecond holding more
+        than chunk_size lines arrives whole). A page filtered away entirely
+        yields nothing at all rather than an empty chunk, which means only that
+        the stream is caught up.
+
+        start_time defaults to the current time, resolved once when fetch_logs is
+        called (not at the first next()), so lines logged while the generator sits
+        unstarted are not skipped; pass an earlier start_time to read history.
+
+        With end_time set the iterator terminates once the window is delivered or
+        the store has no more lines to give, whichever comes first — an end_time
+        in the future does not keep it polling until then.
+        Without end_time it never terminates — once caught up it yields an empty
+        chunk each time nothing new is stored yet, and the caller decides when to
+        sleep or break:
+
+            for chunk in cclient.fetch_logs(dep, rev, pod):
+                if not chunk:
+                    time.sleep(2)
+                    continue
+                ...
+
+        Nothing is fetched before the first next(), and memory is bounded by the
+        dedup window rather than by the length of the stream: the anchor holds
+        only the ids and timestamps of the last LOG_DEDUP_RETENTION_MS, a
+        generous margin over the server's re-delivery span, so no line is
+        delivered twice — across empty chunks too.
+
+        A line the log store received late lands in a later chunk than its
+        timestamp position, never duplicated, as long as it lands inside that
+        re-delivery span (~15s). This reader only ever pages forward, and the
+        server re-delivers the span behind the boundary only; a line whose
+        timestamp falls further than the span behind the newest line already
+        delivered is never returned at all. Lines inside each chunk are always
+        in ascending (timestamp, id) order.
+
+        A page request the store answers as busy (HTTP 503) is retried with
+        exponential backoff and jitter; the anchor survives the retry, so an
+        upstream rate limit costs a pause rather than the stream.
+
+        If a page request fails for any other reason, or the retries run out,
+        the iterator raises and, like any generator, cannot be resumed — but
+        every chunk already yielded is complete and none is left half-built.
+        Resume with a new fetch_logs whose start_time is the last delivered
+        event's timestamp: bounds are inclusive, so the only lines delivered
+        again are the ones sharing that millisecond, which the caller already
+        holds.
+        """
+        if not 1 <= chunk_size <= MAX_LOG_PAGE_LINES:
+            raise ValueError(
+                f"chunk_size must be between 1 and {MAX_LOG_PAGE_LINES} lines "
+                "(chunk_size is also the per-request page size, and the server "
+                f"caps max_lines at {MAX_LOG_PAGE_LINES})"
+            )
+        if (start_time is not None and start_time < 0) or (end_time is not None and end_time < 0):
+            raise ValueError("start_time and end_time are epoch milliseconds and must not be negative")
+        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")
+        start_ms = int(time.time() * 1000) if start_time is None else start_time
+
+        def chunks() -> Iterator[List[DeploymentLogEvent]]:
+            held: list = []
+            # after is exclusive, so start_ms - 1 admits lines at start_ms itself.
+            initial_boundary = max(start_ms - 1, 0)
+            while True:
+                anchor: Union[list, int] = _recent_anchor(held) if held else initial_boundary
+                page = _with_busy_retry(
+                    partial(
+                        self._fetch_log_page, deployment_id, revision_number, pod, after=anchor, max_lines=chunk_size
+                    )
+                )
+                past_end = False
+                chunk: List[DeploymentLogEvent] = []
+                for raw in page:
+                    anchor_event = _LogAnchor(id=raw.id, timestamp=raw.timestamp)
+                    if held and raw.id <= held[-1].id:
+                        # Late arrival inside the look-behind span: keep the held window
+                        # id-ordered (id order == time order) so trimming stays correct.
+                        insort(held, anchor_event, key=lambda held_event: held_event.id)
+                    else:
+                        held.append(anchor_event)
+                    # Anchoring at start_ms - 1 re-delivers the look-behind span below
+                    # start_ms; those ids must be held for dedup but never emitted.
+                    if raw.timestamp < start_ms:
+                        continue
+                    if end_time is not None and raw.timestamp > end_time:
+                        past_end = True
+                        break
+                    event = DeploymentLogEvent(id=raw.id, timestamp=raw.timestamp, message=raw.message, pod=pod)
+                    # The server orders a page by nanosecond timestamp only, never by
+                    # the id's hash suffix, so lines sharing one nanosecond can arrive
+                    # in either id order; insort keeps every chunk ascending.
+                    if chunk and (event.timestamp, event.id) < (chunk[-1].timestamp, chunk[-1].id):
+                        insort(chunk, event, key=lambda chunk_event: (chunk_event.timestamp, chunk_event.id))
+                    else:
+                        chunk.append(event)
+                if page:
+                    held = _recent_anchor(held)
+                if chunk:
+                    yield chunk
+                if past_end or (not page and end_time is not None):
+                    break
+                if not page:
+                    # Caught up with no end bound: signal "nothing new yet" until
+                    # new lines are stored.
+                    yield []
+
+        # The nested generator closes over the validated arguments, so the
+        # ValueErrors above raise at the call rather than at the first next().
+        return chunks()
+
+    @deprecated("deployment_log_session() is deprecated; use fetch_logs() instead")
     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
+        """Deprecated: use fetch_logs() instead.
+
+        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)
 
 
+@deprecated("DeploymentLogSession is deprecated; use CentMLClient.fetch_logs() instead")
 class DeploymentLogSession:
-    """Maintains a contiguous, ordered window of one pod's logs across fetches.
+    """Deprecated: use CentMLClient.fetch_logs() instead.
 
+    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).
@@ -372,7 +576,7 @@ 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(
+        page = self._client._fetch_log_page(
             self._deployment_id,
             self._revision_number,
             self._pod,
@@ -390,7 +594,7 @@ def fetch_newer(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list:
         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(
+        delta = self._client._fetch_log_page(
             self._deployment_id,
             self._revision_number,
             self._pod,