[iris] Replace heartbeat-based logging with push-based LogService (#4274)

Add a LogService (PushLogs/QueryLogs RPCs) co-hosted on the controller.
Workers push task and process logs directly via RPC instead of
piggybacking
on heartbeat responses. Removes worker-local LogStore, FetchLogs RPC
from
WorkerService, and all heartbeat log_entries plumbing. K8s provider
pushes
logs to the controller LogStore directly. Controller process logs keyed
as
/system/controller, worker process logs as /system/worker/<id>. Requires
full cluster restart.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: rjpower

lib/iris/dashboard/src/components/shared/LogViewer.vue

@@ -1,6 +1,6 @@
 <script setup lang="ts">
 import { ref, computed, onMounted, watch } from 'vue'
-import { useControllerRpc, useWorkerRpc } from '@/composables/useRpc'
+import { useLogServiceRpc } from '@/composables/useRpc'
 import { useAutoRefresh } from '@/composables/useAutoRefresh'
 import type { FetchLogsResponse, LogEntry, TaskAttempt } from '@/types/rpc'
 import { timestampMs, logLevelClass, formatLogTime } from '@/utils/formatting'
@@ -34,8 +34,8 @@ function levelPriority(lvl: string | undefined): number {
   return LOG_LEVEL_PRIORITY[lvl.toLowerCase()] ?? 1
 }
 
-// Choose the right RPC based on what we're viewing
-const useRpc = props.source === 'worker' ? useWorkerRpc : useControllerRpc
+// FetchLogs is served by the LogService (co-hosted on the controller)
+const useRpc = useLogServiceRpc
 
 // Task IDs end with a numeric segment (e.g. /alice/job/0), job IDs don't.
 const isTask = props.taskId ? /\/\d+$/.test(props.taskId) : false
@@ -54,7 +54,7 @@ const taskLogState = props.taskId
 
 const processLogState = !props.taskId
   ? useRpc<FetchLogsResponse>('FetchLogs', () => ({
-      source: props.workerId ? `/worker/${props.workerId}` : '/system/process',
+      source: props.workerId ? `/system/worker/${props.workerId}` : '/system/controller',
       maxLines: tailLines.value || undefined,
       tail: true,
     }))

lib/iris/dashboard/src/composables/useRpc.ts

@@ -92,6 +92,26 @@ export async function controllerRpcCall<T>(method: string, body?: Record<string,
   return resp.json() as Promise<T>
 }
 
+/** RPC composable for LogService endpoints. */
+export function useLogServiceRpc<T>(
+  method: string,
+  body?: RpcBody,
+): RpcState<T> {
+  return useRpc<T>('iris.logging.LogService', method, body)
+}
+
+/** One-shot RPC call for LogService. */
+export async function logServiceRpcCall<T>(method: string, body?: Record<string, unknown>): Promise<T> {
+  const resp = await fetch(`/iris.logging.LogService/${method}`, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(body ?? {}),
+  })
+  handleUnauthorized(resp)
+  if (!resp.ok) throw new Error(`${method}: ${resp.status} ${resp.statusText}`)
+  return resp.json() as Promise<T>
+}
+
 export async function workerRpcCall<T>(method: string, body?: Record<string, unknown>): Promise<T> {
   const resp = await fetch(`/iris.cluster.WorkerService/${method}`, {
     method: 'POST',

lib/iris/src/iris/cli/bug_report.py

@@ -13,9 +13,10 @@
 
 from iris.cluster.log_store import build_log_source
 from iris.cluster.types import JobName
-from iris.rpc import cluster_pb2
+from iris.rpc import cluster_pb2, logging_pb2
 from iris.rpc.auth import AuthTokenInjector, TokenProvider
 from iris.rpc.cluster_connect import ControllerServiceClientSync
+from iris.rpc.logging_connect import LogServiceClientSync
 from iris.time_proto import timestamp_from_proto
 
 logger = logging.getLogger(__name__)
@@ -105,14 +106,17 @@ def gather_bug_report(
     """Gather all diagnostic data for a job into a BugReport."""
     interceptors = [AuthTokenInjector(token_provider)] if token_provider else []
     client = ControllerServiceClientSync(controller_url, timeout_ms=30000, interceptors=interceptors)
+    log_client = LogServiceClientSync(controller_url, timeout_ms=30000, interceptors=interceptors)
     try:
-        return _gather(client, job_id, tail=tail)
+        return _gather(client, log_client, job_id, tail=tail)
     finally:
+        log_client.close()
         client.close()
 
 
 def _gather(
     client: ControllerServiceClientSync,
+    log_client: LogServiceClientSync,
     job_id: JobName,
     *,
     tail: int,
@@ -141,8 +145,8 @@ def _gather(
         try:
             # Fetch all attempts for this task, taking only the last `tail` lines.
             source = build_log_source(JobName.from_wire(task.task_id))
-            log_resp = client.fetch_logs(
-                cluster_pb2.FetchLogsRequest(
+            log_resp = log_client.fetch_logs(
+                logging_pb2.FetchLogsRequest(
                     source=source,
                     max_lines=tail,
                     tail=True,

lib/iris/src/iris/cli/process_status.py

@@ -13,9 +13,10 @@
 import humanfriendly
 
 from iris.cli.main import require_controller_url, rpc_client
-from iris.rpc import cluster_pb2
+from iris.rpc import cluster_pb2, logging_pb2
+from iris.rpc.logging_connect import LogServiceClientSync
 
-_CONTROLLER_TARGET = "/system/process"
+_CONTROLLER_LOG_TARGET = "/system/controller"
 
 
 def _print_status(resp: cluster_pb2.GetProcessStatusResponse, label: str) -> None:
@@ -81,36 +82,36 @@ def logs(ctx, target: str | None, level: str, follow: bool, max_lines: int, subs
     from datetime import datetime, timezone
 
     url = require_controller_url(ctx)
-    source = target or _CONTROLLER_TARGET
-
-    with rpc_client(url) as client:
-        cursor = 0
-        first = True
-        while True:
-            req = cluster_pb2.FetchLogsRequest(
-                source=source,
-                max_lines=max_lines if first else 100,
-                tail=first,
-                min_level=level,
-                cursor=cursor if not first else 0,
-            )
-            if substring:
-                req.substring = substring
+    log_client = LogServiceClientSync(url)
+    source = target or _CONTROLLER_LOG_TARGET
+
+    cursor = 0
+    first = True
+    while True:
+        req = logging_pb2.FetchLogsRequest(
+            source=source,
+            max_lines=max_lines if first else 100,
+            tail=first,
+            min_level=level,
+            cursor=cursor if not first else 0,
+        )
+        if substring:
+            req.substring = substring
 
-            resp = client.fetch_logs(req)
-            for entry in resp.entries:
-                ts = ""
-                if entry.timestamp and entry.timestamp.epoch_ms:
-                    dt = datetime.fromtimestamp(entry.timestamp.epoch_ms / 1000, tz=timezone.utc)
-                    ts = dt.strftime("%H:%M:%S")
-                click.echo(f"[{ts}] {entry.data}")
+        resp = log_client.fetch_logs(req)
+        for entry in resp.entries:
+            ts = ""
+            if entry.timestamp and entry.timestamp.epoch_ms:
+                dt = datetime.fromtimestamp(entry.timestamp.epoch_ms / 1000, tz=timezone.utc)
+                ts = dt.strftime("%H:%M:%S")
+            click.echo(f"[{ts}] {entry.data}")
 
-            cursor = resp.cursor
-            first = False
+        cursor = resp.cursor
+        first = False
 
-            if not follow:
-                break
-            time.sleep(2)
+        if not follow:
+            break
+        time.sleep(2)
 
 
 @process_group.command()
@@ -139,7 +140,7 @@ def profile(
     /system/worker/<id> for a worker, /alice/job/0 for a task container.
     """
     url = require_controller_url(ctx)
-    rpc_target = target or _CONTROLLER_TARGET
+    rpc_target = target or ""
     label = target or "Controller"
 
     if profiler == "threads":

lib/iris/src/iris/cluster/client/protocol.py

@@ -6,7 +6,7 @@
 from typing import Protocol
 
 from iris.cluster.types import Entrypoint, JobName, TaskAttempt
-from iris.rpc import cluster_pb2
+from iris.rpc import cluster_pb2, logging_pb2
 from rigging.timing import Duration
 
 
@@ -91,7 +91,7 @@ def fetch_logs(
         substring: str = "",
         min_level: str = "",
         tail: bool = False,
-    ) -> cluster_pb2.FetchLogsResponse: ...
+    ) -> logging_pb2.FetchLogsResponse: ...
 
     def get_autoscaler_status(self) -> cluster_pb2.Controller.GetAutoscalerStatusResponse: ...
 

lib/iris/src/iris/cluster/client/remote_client.py

@@ -18,8 +18,9 @@
 from iris.cluster.log_store._types import build_log_source
 from iris.cluster.runtime.entrypoint import build_runtime_entrypoint
 from iris.cluster.types import Entrypoint, EnvironmentSpec, JobName, TaskAttempt, adjust_tpu_replicas, is_job_finished
-from iris.rpc import cluster_pb2
+from iris.rpc import cluster_pb2, logging_pb2
 from iris.rpc.cluster_connect import ControllerServiceClientSync
+from iris.rpc.logging_connect import LogServiceClientSync
 from iris.rpc.errors import call_with_retry, format_connect_error, poll_with_retries
 from iris.time_proto import duration_to_proto
 from rigging.timing import Deadline, Duration, ExponentialBackoff
@@ -65,6 +66,11 @@ def __init__(
             timeout_ms=timeout_ms,
             interceptors=interceptors,
         )
+        self._log_client = LogServiceClientSync(
+            address=controller_address,
+            timeout_ms=timeout_ms,
+            interceptors=interceptors,
+        )
 
     def submit_job(
         self,
@@ -343,6 +349,7 @@ def _call():
 
     def shutdown(self, wait: bool = True) -> None:
         del wait
+        self._log_client.close()
         self._client.close()
 
     def get_task_status(self, task_name: JobName) -> cluster_pb2.TaskStatus:
@@ -390,8 +397,8 @@ def fetch_logs(
         substring: str = "",
         min_level: str = "",
         tail: bool = False,
-    ) -> cluster_pb2.FetchLogsResponse:
-        request = cluster_pb2.FetchLogsRequest(
+    ) -> logging_pb2.FetchLogsResponse:
+        request = logging_pb2.FetchLogsRequest(
             source=source,
             since_ms=since_ms,
             cursor=cursor,
@@ -402,7 +409,7 @@ def fetch_logs(
         )
 
         def _call():
-            return self._client.fetch_logs(request)
+            return self._log_client.fetch_logs(request)
 
         return call_with_retry(f"fetch_logs({source})", _call)
 

lib/iris/src/iris/cluster/controller/controller.py

@@ -73,6 +73,7 @@
 from iris.cluster.controller.dashboard import ControllerDashboard
 from iris.cluster.providers.k8s.tasks import K8sTaskProvider
 from iris.cluster.controller.provider import TaskProvider
+from iris.cluster.controller.worker_provider import WorkerProvider
 from iris.cluster.controller.scheduler import (
     JobRequirements,
     Scheduler,
@@ -94,7 +95,8 @@
     ReservationClaim,
     SchedulingEvent,
 )
-from iris.cluster.log_store import PROCESS_LOG_KEY, LogStore, LogStoreHandler
+from iris.cluster.log_store import CONTROLLER_LOG_KEY, LogStoreHandler
+from iris.log_server.server import LogServiceImpl
 from iris.cluster.types import (
     JobName,
     WorkerStatus,
@@ -105,7 +107,7 @@
 )
 from rigging.log_setup import slow_log
 from iris.managed_thread import ManagedThread, ThreadContainer, get_thread_container
-from iris.rpc import cluster_pb2
+from iris.rpc import cluster_pb2, logging_pb2
 from iris.rpc.auth import TokenVerifier
 from rigging.timing import Duration, ExponentialBackoff, RateLimiter, Timer, Timestamp, TokenBucket
 
@@ -123,6 +125,21 @@
 _HISTORY_CLEANUP_INTERVAL_S = 60.0
 
 
+class _InProcessLogPusher:
+    """Adapts LogServiceImpl to the LogPusherProtocol for in-process use.
+
+    Avoids a network round-trip when the K8s provider is co-hosted with
+    the controller: calls push_logs() directly on the service impl.
+    """
+
+    def __init__(self, log_service: LogServiceImpl) -> None:
+        self._log_service = log_service
+
+    def push(self, key: str, entries: list[logging_pb2.LogEntry]) -> None:
+        if entries:
+            self._log_service.push_logs(logging_pb2.PushLogsRequest(key=key, entries=entries), ctx=None)
+
+
 class SchedulingOutcome(enum.Enum):
     """Result of a scheduling cycle, used to drive adaptive backoff."""
 
@@ -998,19 +1015,25 @@ def __init__(
             self._db = db
         else:
             self._db = ControllerDB(db_dir=config.local_state_dir / "db")
-        self._log_store = LogStore(
+
+        self._log_service = LogServiceImpl(
             log_dir=config.local_state_dir / "logs",
             remote_log_dir=f"{config.remote_state_dir.rstrip('/')}/logs",
         )
 
-        # Wire log store into the K8s provider so its LogCollector can write logs directly.
-        # Collectors are created lazily on first sync(), so just setting the field is enough.
+        # Wire an in-process log pusher into providers so log entries are
+        # forwarded through the LogService without a network hop.
+        # - K8sTaskProvider: its LogCollector pushes logs directly.
+        # - WorkerProvider: forwards log_entries piggybacked on heartbeat
+        #   responses from old workers that predate push-based logging.
+        in_process_log_pusher = _InProcessLogPusher(self._log_service)
         if isinstance(self._provider, K8sTaskProvider):
-            self._provider.log_store = self._log_store
+            self._provider.log_pusher = in_process_log_pusher
+        elif isinstance(self._provider, WorkerProvider):
+            self._provider.log_pusher = in_process_log_pusher
 
         self._transitions = ControllerTransitions(
             db=self._db,
-            log_store=self._log_store,
             heartbeat_failure_threshold=config.heartbeat_failure_threshold,
             user_budget_defaults=config.user_budget_defaults,
         )
@@ -1023,20 +1046,23 @@ def __init__(
             self._db,
             controller=self,
             bundle_store=self._bundle_store,
-            log_store=self._log_store,
+            log_service=self._log_service,
             auth=config.auth,
+            system_endpoints={},
         )
         self._dashboard = ControllerDashboard(
             self._service,
+            log_service=self._log_service,
             host=config.host,
             port=config.port,
             auth_verifier=config.auth_verifier,
             auth_provider=config.auth_provider,
             auth_optional=config.auth.optional if config.auth else False,
         )
 
-        # Ingest process logs into the LogStore so they are available via FetchLogs.
-        self._log_store_handler = LogStoreHandler(self._log_store, key=PROCESS_LOG_KEY)
+        # Ingest controller process logs into the LogStore via LogStoreHandler.
+        # This writes directly to the co-hosted LogStore (no RPC round-trip).
+        self._log_store_handler = LogStoreHandler(self._log_service.log_store, key=CONTROLLER_LOG_KEY)
         self._log_store_handler.setLevel(logging.DEBUG)
         self._log_store_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(message)s"))
         logging.getLogger("iris").addHandler(self._log_store_handler)
@@ -1144,6 +1170,12 @@ def start(self) -> None:
             timeout=Duration.from_seconds(5.0),
         )
 
+        # Register system endpoints. The address here is used for in-process
+        # and local-network callers (e.g. CLI, tests). Remote workers fall back
+        # to their configured controller_address when the resolved endpoint
+        # is unreachable, since the log service is co-hosted on the controller.
+        self._service._system_endpoints["/system/log-server"] = self.url
+
     def stop(self) -> None:
         """Stop all background components gracefully.
 
@@ -1183,7 +1215,7 @@ def stop(self) -> None:
         # sqlite3.ProgrammingError spam from late log records.
         logging.getLogger("iris").removeHandler(self._log_store_handler)
         self._log_store_handler.close()
-        self._log_store.close()
+        self._log_service.close()
         self._db.close()
         self._bundle_store.close()