[1/N] [History Server Beta] [Feat] Lazy loading

historyserver/README.md

@@ -11,7 +11,7 @@ It provides a web interface to explore the history of Ray jobs, tasks, actors, a
 The History Server consists of two main components:
 
 1. **Collector**: Runs as a sidecar container in Ray clusters to collect logs and metadata
-2. **History Server**: Central service that aggregates data from collectors and provides a web UI
+2. **History Server**: Serves a Ray Dashboard-shaped HTTP API and ingests cluster sessions' events on demand
 
 ## Building
 
@@ -139,6 +139,87 @@ To run lint checks:
 make alllint
 ```
 
+## Smoke Tests
+
+### 1. Deploy History Server
+
+Apply MinIO and the History Server manifests:
+
+```bash
+kubectl apply -f historyserver/config/minio.yaml
+kubectl apply -f historyserver/config/service_account.yaml
+kubectl apply -f historyserver/config/historyserver.yaml
+```
+
+Port-forward the HS service:
+
+```bash
+kubectl port-forward svc/historyserver 8080:30080
+```
+
+### 2. Generate a Dead Session
+
+Deploy the sample RayCluster, run a deterministic workload, then delete the CR:
+
+```bash
+kubectl apply -f historyserver/config/raycluster.yaml
+kubectl wait pod -l ray.io/node-type=head --for=condition=Ready --timeout=180s
+
+# Run a workload so events are written to MinIO
+kubectl exec $(kubectl get pod -l ray.io/node-type=head -o name) \
+  -c ray-head -- python -c "
+import ray
+ray.init(address='auto')
+
+@ray.remote
+def add(x, y): return x + y
+
+print('tasks:', ray.get([add.remote(i, i) for i in range(5)]))
+"
+
+# Delete the cluster — produces a 'dead' session
+kubectl delete -f historyserver/config/raycluster.yaml
+
+# Discover the session name. /clusters lists both live and dead sessions;
+# dead sessions carry the `session_*` name you'll feed into /enter_cluster.
+curl -sS http://localhost:8080/clusters
+```
+
+### 3. Cold Path (first visit)
+
+Trigger the lazy load synchronously. Replace `<session>` with the session name from §2:
+
+```bash
+time curl -s -o /dev/null \
+  http://localhost:8080/enter_cluster/default/raycluster-historyserver/<session>
+```
+
+> [!NOTE]
+> Cold path runs synchronously: K8s probe + event parse. Expect this to take seconds. Subsequent endpoint calls
+> (`/api/v0/jobs`, `/api/v0/tasks/...`) read from in-memory state populated by this call.
+
+### 4. Warm Path (subsequent visits on same replica)
+
+Re-enter the same cluster:
+
+```bash
+time curl -s -o /dev/null \
+  http://localhost:8080/enter_cluster/default/raycluster-historyserver/<session>
+```
+
+> [!NOTE]
+> Warm path returns immediately — SessionLoader's loaded-session set fast-paths the request without re-parsing. If the HS
+> process restarts, the next visit returns to cold-path latency.
+
+### 5. Tail Logs
+
+```bash
+kubectl logs -f -l app=historyserver --tail=50
+```
+
+You should see one `current eventFileList for cluster ...` line (and the per-file `Reading event file: ...` entries)
+per first-time cold-path call. Warm-path calls produce no parse log lines.
+
 ## Deployment
 
 History Server can be deployed in Kubernetes using the manifests in the `config/samples/` directory.

historyserver/cmd/historyserver/main.go

@@ -1,57 +1,61 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"flag"
 	"os"
 	"os/signal"
 	"sync"
 	"syscall"
 
+	"github.com/sirupsen/logrus"
+
 	"github.com/ray-project/kuberay/historyserver/pkg/collector"
 	"github.com/ray-project/kuberay/historyserver/pkg/collector/types"
 	"github.com/ray-project/kuberay/historyserver/pkg/eventserver"
 	"github.com/ray-project/kuberay/historyserver/pkg/historyserver"
-	"github.com/sirupsen/logrus"
 )
 
 func main() {
 	runtimeClassName := ""
 	rayRootDir := ""
 	kubeconfigs := ""
-	runtimeClassConfigPath := "/var/collector-config/data"
+	runtimeClassConfigPath := ""
 	dashboardDir := ""
 	useKubernetesProxy := false
-	flag.StringVar(&runtimeClassName, "runtime-class-name", "", "")
-	flag.StringVar(&rayRootDir, "ray-root-dir", "", "")
-	flag.StringVar(&kubeconfigs, "kubeconfigs", "", "")
-	flag.StringVar(&dashboardDir, "dashboard-dir", "/dashboard", "")
-	flag.StringVar(&runtimeClassConfigPath, "runtime-class-config-path", "", "") //"/var/collector-config/data"
-	flag.BoolVar(&useKubernetesProxy, "use-kubernetes-proxy", false, "")
+	flag.StringVar(&runtimeClassName, "runtime-class-name", "", "Storage backend: s3 / gcs / azureblob / aliyunoss / localtest")
+	flag.StringVar(&rayRootDir, "ray-root-dir", "", "Root dir inside the bucket")
+	flag.StringVar(&kubeconfigs, "kubeconfigs", "", "Kubeconfig path; empty = in-cluster")
+	flag.StringVar(&dashboardDir, "dashboard-dir", "/dashboard", "Path to Ray Dashboard static assets")
+	flag.StringVar(&runtimeClassConfigPath, "runtime-class-config-path", "", "Path to backend config JSON")
+	flag.BoolVar(&useKubernetesProxy, "use-kubernetes-proxy", false, "Use local kubeconfig instead of in-cluster config")
 	flag.Parse()
 
+	if runtimeClassName == "" {
+		logrus.Fatal("--runtime-class-name is required")
+	}
+
 	cliMgr, err := historyserver.NewClientManager(kubeconfigs, useKubernetesProxy)
 	if err != nil {
-		logrus.Errorf("Failed to create client manager: %v", err)
-		os.Exit(1)
+		logrus.Fatalf("Failed to create client manager: %v", err)
 	}
 
 	jsonData := make(map[string]interface{})
 	if runtimeClassConfigPath != "" {
 		data, err := os.ReadFile(runtimeClassConfigPath)
 		if err != nil {
-			panic("Failed to read runtime class config " + err.Error())
+			logrus.Fatalf("Failed to read runtime-class-config from %s: %v", runtimeClassConfigPath, err)
 		}
-		err = json.Unmarshal(data, &jsonData)
-		if err != nil {
-			panic("Failed to parse runtime class config: " + err.Error())
+		if err := json.Unmarshal(data, &jsonData); err != nil {
+			logrus.Fatalf("Failed to parse runtime-class-config from %s: %v", runtimeClassConfigPath, err)
 		}
 	}
 
 	registry := collector.GetReaderRegistry()
 	factory, ok := registry[runtimeClassName]
 	if !ok {
-		panic("Not supported runtime class name: " + runtimeClassName + ".")
+		logrus.Fatalf("Unsupported runtime-class-name for reader: %s", runtimeClassName)
 	}
 
 	globalConfig := types.RayHistoryServerConfig{
@@ -60,34 +64,36 @@ func main() {
 
 	reader, err := factory(&globalConfig, jsonData)
 	if err != nil {
-		panic("Failed to create reader for runtime class name: " + runtimeClassName + ".")
+		logrus.Fatalf("Failed to create reader for runtime class name %s: %v", runtimeClassName, err)
 	}
 
-	// Create EventHandler with storage reader
 	eventHandler := eventserver.NewEventHandler(reader)
 
-	// WaitGroup to track goroutine completion
-	var wg sync.WaitGroup
+	serverCtx, serverCancel := signal.NotifyContext(
+		context.Background(),
+		syscall.SIGINT, syscall.SIGTERM,
+	)
+	defer serverCancel()
 
-	sigChan := make(chan os.Signal, 1)
-	stop := make(chan struct{}, 1)
-	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+	resolver, err := historyserver.NewHTTPLiveSessionResolver(cliMgr, useKubernetesProxy)
+	if err != nil {
+		logrus.Fatalf("Failed to create live session resolver: %v", err)
+	}
+	processor := historyserver.NewSessionProcessor(eventHandler, cliMgr.Client(), resolver)
+	sessionLoader := historyserver.NewSessionLoader(processor, serverCtx)
 
-	// Start EventHandler in background goroutine
-	wg.Add(1)
+	// ServerHandler.Run consumes a stop chan; bridge serverCtx into it.
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
 	go func() {
-		defer wg.Done()
-		logrus.Info("Starting EventHandler in background...")
-		if err := eventHandler.Run(stop, 2); err != nil {
-			logrus.Errorf("EventHandler stopped with error: %v", err)
-		}
-		logrus.Info("EventHandler shutdown complete")
+		<-serverCtx.Done()
+		logrus.Info("Received shutdown signal, initiating graceful shutdown...")
+		close(stop)
 	}()
 
-	handler, err := historyserver.NewServerHandler(&globalConfig, dashboardDir, reader, cliMgr, eventHandler, useKubernetesProxy)
+	handler, err := historyserver.NewServerHandler(&globalConfig, dashboardDir, reader, cliMgr, eventHandler, sessionLoader, useKubernetesProxy)
 	if err != nil {
-		logrus.Errorf("Failed to create server handler: %v", err)
-		os.Exit(1)
+		logrus.Fatalf("Failed to create server handler: %v", err)
 	}
 
 	wg.Add(1)
@@ -97,13 +103,6 @@ func main() {
 		logrus.Info("HTTP server shutdown complete")
 	}()
 
-	<-sigChan
-	logrus.Info("Received shutdown signal, initiating graceful shutdown...")
-
-	// Stop both the server and the event handler
-	close(stop)
-
-	// Wait for both goroutines to complete
 	wg.Wait()
 	logrus.Info("Graceful shutdown complete")
 }

historyserver/pkg/eventserver/eventserver.go

@@ -76,7 +76,12 @@ func (h *EventHandler) ProcessEvents(ctx context.Context, ch <-chan map[string]a
 				logrus.Warnf("Channel was closed")
 				return nil
 			}
-			if err := h.storeEvent(currEventData); err != nil {
+			clusterSessionKey, ok := currEventData["clusterName"].(string)
+			if !ok {
+				logrus.Errorf("event missing or non-string 'clusterName' field")
+				continue
+			}
+			if err := h.storeEvent(clusterSessionKey, currEventData); err != nil {
 				logrus.Errorf("Failed to store event: %v", err)
 				continue
 			}
@@ -208,8 +213,8 @@ func (h *EventHandler) Run(stop chan struct{}, numOfEventProcessors int) error {
 	return nil
 }
 
-// storeEvent unmarshals the event map into the correct actor/task struct and then stores it into the corresonding list
-func (h *EventHandler) storeEvent(eventMap map[string]any) error {
+// storeEvent unmarshals the event map into the correct actor/task struct and then stores it into the corresonding list.
+func (h *EventHandler) storeEvent(clusterSessionKey string, eventMap map[string]any) error {
 	eventTypeVal, ok := eventMap["eventType"]
 	if !ok {
 		return fmt.Errorf("event missing 'eventType' field")
@@ -220,17 +225,6 @@ func (h *EventHandler) storeEvent(eventMap map[string]any) error {
 	}
 	eventType := types.EventType(eventTypeStr)
 
-	// clusterSessionKey is injected during event reading (Run function) and contains
-	// the full key: "{clusterName}_{namespace}_{sessionName}"
-	clusterSessionKeyVal, ok := eventMap["clusterName"]
-	if !ok {
-		return fmt.Errorf("event missing 'clusterName' field (clusterSessionKey)")
-	}
-	clusterSessionKey, ok := clusterSessionKeyVal.(string)
-	if !ok {
-		return fmt.Errorf("clusterName is not a string, got %T", clusterSessionKeyVal)
-	}
-
 	logrus.Infof("current eventType: %v", eventType)
 	switch eventType {
 	case types.TASK_DEFINITION_EVENT:
@@ -667,7 +661,7 @@ func (h *EventHandler) GetTasks(clusterSessionKey string) []types.Task {
 
 	taskMap, ok := h.ClusterTaskMap.ClusterTaskMap[clusterSessionKey]
 	if !ok {
-		// TODO(jwj): Add error handling.
+		// TODO(jiangjiawei1103): Add error handling.
 		logrus.Errorf("Task map not found for cluster session: %s", clusterSessionKey)
 		return []types.Task{}
 	}
@@ -938,7 +932,7 @@ func (h *EventHandler) handleTaskLifecycleEvent(eventMap map[string]any, cluster
 	}
 	normalizeTaskIDsToHex(&currTask)
 
-	// TODO(jwj): Clarify if there must be at least one state transition. Can one task have more than one state transition?
+	// TODO(jiangjiawei1103): Clarify if there must be at least one state transition. Can one task have more than one state transition?
 	if len(currTask.StateTransitions) == 0 {
 		return fmt.Errorf("TASK_LIFECYCLE_EVENT must have at least one state transition")
 	}
@@ -974,7 +968,7 @@ func (h *EventHandler) handleTaskLifecycleEvent(eventMap map[string]any, cluster
 			return
 		}
 
-		// TODO(jwj): Before beta, the lifecycle-related fields are overwritten.
+		// TODO(jiangjiawei1103): Before beta, the lifecycle-related fields are overwritten.
 		// In beta, the complete historical replay will be supported.
 		task.RayErrorInfo = currTask.RayErrorInfo
 		if currTask.JobID != "" {
@@ -1531,3 +1525,71 @@ func extractActorIDFromTaskID(taskIDHex string) string {
 
 	return actorPortion + jobPortion
 }
+
+// ProcessSingleSession reads all event files for a single session synchronously
+// and populates the handler's in-memory maps.
+//
+// TODO(jiangjiawei1103): Empty event file list vs ListFiles outage is ambiguous without
+// StorageReader interface surfacing errors.
+func (h *EventHandler) ProcessSingleSession(ctx context.Context, clusterInfo utils.ClusterInfo) error {
+	clusterNameNamespace := clusterInfo.Name + "_" + clusterInfo.Namespace
+	clusterSessionKey := utils.BuildClusterSessionKey(clusterInfo.Name, clusterInfo.Namespace, clusterInfo.SessionName)
+
+	// ClusterLogEventMap backs only the /events endpoint, so log event read failures must not
+	// block marking the session as loaded and force subsequent Ray event re-processing.
+	logEventReader := NewLogEventReader(h.reader)
+	if err := logEventReader.ReadLogEvents(clusterInfo, clusterSessionKey, h.ClusterLogEventMap); err != nil {
+		logrus.Errorf("Incomplete Log Events read for %s: %v. /events endpoint may serve partial data.",
+			clusterSessionKey, err)
+	}
+
+	eventFileList := append(h.getAllJobEventFiles(clusterInfo), h.getAllNodeEventFiles(clusterInfo)...)
+	logrus.Debugf("current eventFileList for cluster %s is: %v", clusterInfo.Name, eventFileList)
+
+	rayEventsTotal := len(eventFileList)
+	rayEventsRead := 0
+	for _, eventFile := range eventFileList {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		logrus.Debugf("Reading event file: %s", eventFile)
+
+		// GetContent and io.ReadAll failures are treated as transient storage errors:
+		// skip this file and continue.
+		eventioReader := h.reader.GetContent(clusterNameNamespace, eventFile)
+		if eventioReader == nil {
+			logrus.Errorf("Failed to get content for event file: %s, skipping", eventFile)
+			continue
+		}
+		eventbytes, err := io.ReadAll(eventioReader)
+		if err != nil {
+			logrus.Errorf("Failed to read events for file %s: %v", eventFile, err)
+			continue
+		}
+		rayEventsRead++
+
+		// json.Unmarshal and storeEvent failures are treated as corrupt-data errors:
+		// retrying won't fix bad bytes, accepting partial loss.
+		var eventList []map[string]any
+		if err := json.Unmarshal(eventbytes, &eventList); err != nil {
+			logrus.Errorf("Failed to unmarshal events for file %s: %v", eventFile, err)
+			continue
+		}
+
+		for _, event := range eventList {
+			if event == nil {
+				continue
+			}
+			if err := h.storeEvent(clusterSessionKey, event); err != nil {
+				logrus.Errorf("Failed to store events for file %s: %v", eventFile, err)
+				continue
+			}
+		}
+	}
+
+	if rayEventsTotal > 0 && rayEventsRead == 0 {
+		return fmt.Errorf("read 0 of %d event files for %s: likely transient storage outage",
+			rayEventsTotal, clusterSessionKey)
+	}
+	return nil
+}