[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 compatible 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,63 @@
 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, "")
+	sessionProcessTimeout := historyserver.DefaultSessionProcessTimeout
+	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.DurationVar(&sessionProcessTimeout, "session-process-timeout", historyserver.DefaultSessionProcessTimeout, "Timeout duration for processing and loading a single Ray cluster session.")
 	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-path 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-path 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 +66,32 @@ 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)
+	processor := historyserver.NewSessionProcessor(eventHandler, cliMgr.Client())
+	sessionLoader := historyserver.NewSessionLoader(processor, serverCtx, sessionProcessTimeout)
 
-	// 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 +101,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")
 }