[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 — Supervisor'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,79 @@
+// Package main is the entrypoint for the History Server HTTP daemon.
+// It exposes Ray Dashboard-shaped API endpoints over HTTP and drives
+// per-session event processing on demand via a Supervisor when
+// /enter_cluster hits a dead session.
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"flag"
 	"os"
 	"os/signal"
 	"sync"
 	"syscall"
 
+	"github.com/sirupsen/logrus"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+
 	"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"
-	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, "")
+	// ===== Flags =====
+	var (
+		runtimeClassName       string
+		rayRootDir             string
+		kubeconfigs            string
+		dashboardDir           string
+		runtimeClassConfigPath string
+		useKubernetesProxy     bool
+	)
+	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")
+	}
+
+	// ===== ClientManager =====
 	cliMgr, err := historyserver.NewClientManager(kubeconfigs, useKubernetesProxy)
 	if err != nil {
-		logrus.Errorf("Failed to create client manager: %v", err)
-		os.Exit(1)
+		logrus.Fatalf("client manager: %v", err)
 	}
 
+	// ===== Backend config =====
 	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("read runtime-class-config: %v", 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("parse runtime-class-config: %v", err)
 		}
 	}
 
+	// ===== Reader factory =====
 	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,50 +82,81 @@ func main() {
 
 	reader, err := factory(&globalConfig, jsonData)
 	if err != nil {
-		panic("Failed to create reader for runtime class name: " + runtimeClassName + ".")
+		logrus.Fatalf("create reader: %v", err)
 	}
 
-	// Create EventHandler with storage reader
+	// ===== EventHandler =====
 	eventHandler := eventserver.NewEventHandler(reader)
 
-	// WaitGroup to track goroutine completion
-	var wg sync.WaitGroup
+	// ===== Server context =====
+	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)
+	// ===== K8s client (for Pipeline.isDead) =====
+	k8sClient, err := buildK8sClient(kubeconfigs, useKubernetesProxy)
+	if err != nil {
+		logrus.Fatalf("build k8s client: %v", err)
+	}
 
-	// Start EventHandler in background goroutine
-	wg.Add(1)
+	// ===== Pipeline & Supervisor =====
+	pipeline := historyserver.NewPipeline(eventHandler, k8sClient)
+	supervisor := historyserver.NewSupervisor(pipeline, serverCtx)
+
+	// ===== Shutdown signaling =====
+	// Bridge serverCtx into the legacy stop channel that ServerHandler.Run
+	// consumes; the existing chan-based API is preserved.
+	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)
+	// ===== ServerHandler =====
+	handler, err := historyserver.NewServerHandler(&globalConfig, dashboardDir, reader, cliMgr, eventHandler, supervisor, useKubernetesProxy)
 	if err != nil {
-		logrus.Errorf("Failed to create server handler: %v", err)
-		os.Exit(1)
+		logrus.Fatalf("create server handler: %v", err)
 	}
 
+	// ===== Run HTTP server =====
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		handler.Run(stop)
 		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
+	// ===== Wait for graceful shutdown =====
 	wg.Wait()
 	logrus.Info("Graceful shutdown complete")
 }
+
+// buildK8sClient constructs a controller-runtime client.Client with the
+// rayv1 scheme registered. Used by Pipeline.isDead.
+func buildK8sClient(kubeconfigs string, useKubeProxy bool) (client.Client, error) {
+	var cfg *rest.Config
+	var err error
+
+	switch {
+	case kubeconfigs != "":
+		cfg, err = clientcmd.BuildConfigFromFlags("", kubeconfigs)
+	case useKubeProxy:
+		loading := clientcmd.NewDefaultClientConfigLoadingRules()
+		cfg, err = clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loading, &clientcmd.ConfigOverrides{}).ClientConfig()
+	default:
+		cfg, err = rest.InClusterConfig()
+	}
+	if err != nil {
+		return nil, err
+	}
+	cfg.QPS = 50
+	cfg.Burst = 100
+
+	scheme := runtime.NewScheme()
+	utilruntime.Must(rayv1.AddToScheme(scheme))
+	return client.New(cfg, client.Options{Scheme: scheme})
+}

historyserver/pkg/eventserver/eventserver.go

@@ -1531,3 +1531,80 @@ 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.
+//
+// Per-file failure handling:
+//   - GetContent/ReadAll failures are likely transient storage errors:
+//     skip this file.
+//   - json.Unmarshal/storeEvent failures are treated as corrupt-file:
+//     don't count as a hard failure since retrying won't help.
+//
+// TODO(jwj): Empty event file list vs ListFiles outage is ambiguous without
+// StorageReader interface surfacing errors.
+func (h *EventHandler) ProcessSingleSession(clusterInfo utils.ClusterInfo) error {
+	clusterNameNamespace := clusterInfo.Name + "_" + clusterInfo.Namespace
+	clusterSessionKey := utils.BuildClusterSessionKey(clusterInfo.Name, clusterInfo.Namespace, clusterInfo.SessionName)
+
+	// Read Log Events from logs/{nodeId}/events/event_*.log
+	// This is the format used by Ray Dashboard's /events API.
+	logEventReader := NewLogEventReader(h.reader)
+	logEventErr := logEventReader.ReadLogEvents(clusterInfo, clusterSessionKey, h.ClusterLogEventMap)
+	if logEventErr != nil {
+		logrus.Errorf("Failed to read Log Events for %s: %v", clusterSessionKey, logEventErr)
+	}
+
+	// Read RayEvents (Export Events) from node_events/ and job_events/.
+	// These are used for task/actor/job/node data APIs.
+	eventFileList := append(h.getAllJobEventFiles(clusterInfo), h.getAllNodeEventFiles(clusterInfo)...)
+	logrus.Infof("current eventFileList for cluster %s is: %v", clusterInfo.Name, eventFileList)
+
+	rayEventsAttempted := len(eventFileList)
+	rayEventsSucceeded := 0
+	for _, eventFile := range eventFileList {
+		logrus.Infof("Reading event file: %s", eventFile)
+
+		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 event file: %v", err)
+			continue
+		}
+		rayEventsSucceeded++
+
+		var eventList []map[string]any
+		if err := json.Unmarshal(eventbytes, &eventList); err != nil {
+			logrus.Errorf("Failed to unmarshal event: %v", err)
+			continue
+		}
+
+		for _, event := range eventList {
+			if event == nil {
+				continue
+			}
+			event["clusterName"] = clusterSessionKey
+			if err := h.storeEvent(event); err != nil {
+				logrus.Errorf("Failed to store event: %v", err)
+				continue
+			}
+		}
+	}
+
+	// If every attempted file failed, treat as transient outage and surface
+	// so the session is not marked loaded.
+	if rayEventsAttempted > 0 && rayEventsSucceeded == 0 {
+		return fmt.Errorf("ingested 0 of %d RayEvent files for %s: likely transient storage outage",
+			rayEventsAttempted, clusterSessionKey)
+	}
+
+	if logEventErr != nil {
+		return fmt.Errorf("read log events for %s: %w", clusterSessionKey, logEventErr)
+	}
+
+	return nil
+}