[History Server][Feat] Add metadata collect logic

historyserver/cmd/collector/main.go

@@ -3,9 +3,11 @@ package main
 import (
 	"encoding/json"
 	"flag"
+	"fmt"
 	"os"
 	"os/signal"
 	"path"
+	"strconv"
 	"sync"
 	"syscall"
 	"time"
@@ -30,6 +32,7 @@ func main() {
 	logBatching := 1000
 	eventsPort := 8080
 	pushInterval := time.Minute
+	supportUnsupportedData := false
 	runtimeClassConfigPath := "/var/collector-config/data"
 
 	flag.StringVar(&role, "role", "Worker", "")
@@ -44,6 +47,20 @@ func main() {
 
 	flag.Parse()
 
+	// Read SUPPORT_RAY_EVENT_UNSUPPORTED_DATA environment variable if collector runs in head node
+	if envValue := os.Getenv("SUPPORT_RAY_EVENT_UNSUPPORTED_DATA"); role == "Head" && envValue != "" {
+		if parsed, err := strconv.ParseBool(envValue); err == nil {
+			supportUnsupportedData = parsed
+		} else {
+			logrus.Warnf("Invalid value for SUPPORT_RAY_EVENT_UNSUPPORTED_DATA: %s, using default: %v", envValue, supportUnsupportedData)
+		}
+	}
+
+	dashboardAddress := os.Getenv("RAY_DASHBOARD_ADDRESS")
+	if dashboardAddress == "" {
+		panic(fmt.Errorf("missing RAY_DASHBOARD_ADDRESS in environment variables"))
+	}
+
 	sessionDir, err := utils.GetSessionDir()
 	if err != nil {
 		panic("Failed to get session dir: " + err.Error())
@@ -75,14 +92,16 @@ func main() {
 	}
 
 	globalConfig := types.RayCollectorConfig{
-		RootDir:        rayRootDir,
-		SessionDir:     sessionDir,
-		RayNodeName:    rayNodeId,
-		Role:           role,
-		RayClusterName: rayClusterName,
-		RayClusterID:   rayClusterId,
-		PushInterval:   pushInterval,
-		LogBatching:    logBatching,
+		RootDir:                      rayRootDir,
+		SessionDir:                   sessionDir,
+		RayNodeName:                  rayNodeId,
+		Role:                         role,
+		RayClusterName:               rayClusterName,
+		RayClusterID:                 rayClusterId,
+		PushInterval:                 pushInterval,
+		LogBatching:                  logBatching,
+		DashboardAddress:             dashboardAddress,
+		SupportRayEventUnSupportData: supportUnsupportedData,
 	}
 	logrus.Info("Using collector config: ", globalConfig)
 

historyserver/config/raycluster-azureblob.yaml

@@ -75,11 +75,15 @@ spec:
           image: collector:v0.1.0
           imagePullPolicy: IfNotPresent
           env:
+          - name: RAY_DASHBOARD_ADDRESS
+            value: "http://localhost:8265"
           # reference: https://learn.microsoft.com/en-us/azure/storage/common/storage-use-azurite#connect-to-the-emulator-by-using-the-azure-storage-explorer
           - name: AZURE_STORAGE_CONNECTION_STRING
             value: "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite-service.azurite-dev.svc.cluster.local:10000/devstoreaccount1;"
           - name: AZURE_STORAGE_CONTAINER
             value: ray-historyserver
+          - name: SUPPORT_RAY_EVENT_UNSUPPORTED_DATA
+            value: "true"
           command: [collector, --role=Head, --runtime-class-name=azureblob, --ray-cluster-name=raycluster-historyserver, --ray-root-dir=log, --events-port=8084]
           volumeMounts:
           - name: historyserver
@@ -161,6 +165,8 @@ spec:
           image: collector:v0.1.0
           imagePullPolicy: IfNotPresent
           env:
+          - name: RAY_DASHBOARD_ADDRESS
+            value: "http://raycluster-historyserver-head-svc:8265"
           # reference: https://learn.microsoft.com/en-us/azure/storage/common/storage-use-azurite#connect-to-the-emulator-by-using-the-azure-storage-explorer
           - name: AZURE_STORAGE_CONNECTION_STRING
             value: "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite-service.azurite-dev.svc.cluster.local:10000/devstoreaccount1;"

historyserver/config/raycluster.yaml

@@ -77,6 +77,8 @@ spec:
           image: collector:v0.1.0
           imagePullPolicy: IfNotPresent
           env:
+          - name: RAY_DASHBOARD_ADDRESS
+            value: "http://localhost:8265"
           - name: S3DISABLE_SSL
             value: "true"
           - name: AWS_S3ID
@@ -93,6 +95,8 @@ spec:
             value: "test"
           - name: S3FORCE_PATH_STYLE
             value: "true"
+          - name: SUPPORT_RAY_EVENT_UNSUPPORTED_DATA
+            value: "true"
           command:
           - collector
           - --role=Head
@@ -182,6 +186,8 @@ spec:
           image: collector:v0.1.0
           imagePullPolicy: IfNotPresent
           env:
+          - name: RAY_DASHBOARD_ADDRESS
+            value: "http://raycluster-historyserver-head-svc:8265"
           - name: AWS_S3ID
             value: minioadmin
           - name: AWS_S3SECRET

historyserver/config/rayjob_workaround.yaml

@@ -0,0 +1,48 @@
+# RayJob that demonstrates endpoints used by the "persist metadata" workaround:
+# - Placement groups (api/v0/placement_groups)
+# - Datasets (api/jobs/ + api/data/datasets/{job_id})
+#
+# Use this with the History Server setup guide (Step 6). Ensure the collector
+# has SUPPORT_RAY_EVENT_UNSUPPORTED_DATA enabled so it persists these to blob storage.
+# After the job runs, you can view the data in Ray Dashboard, and after cluster
+# teardown, via History Server (from persisted meta).
+#
+# Note: RayJob connects to the RayCluster created by RayService above via clusterSelector
+#
+apiVersion: ray.io/v1
+kind: RayJob
+metadata:
+  name: rayjob-workaround
+spec:
+  entrypoint: |
+    python -c "
+    import ray
+    import time
+
+    ray.init()
+
+    # --- 1. Placement group (visible under api/v0/placement_groups) ---
+    pg = ray.util.placement_group(
+        name='workaround_pg',
+        strategy='PACK',
+        bundles=[{'CPU': 0.2}]
+    )
+    ray.get(pg.ready())
+    print('Placement group created:', pg.id.hex())
+
+    # --- 2. Dataset (creates job + dataset; visible under api/jobs/ and api/data/datasets/) ---
+    ds = ray.data.range(10)
+    n = ds.count()
+    print('Dataset created, count:', n)
+
+    # Keep running so the collector can poll Dashboard API and persist meta (e.g. every 5s).
+    # Wait long enough for at least one full persist cycle.
+    print('Waiting 15s for collector to persist metadata...')
+    time.sleep(15)
+
+    # Align with existing guide: allow events to be sent before shutdown.
+    time.sleep(5)
+    print('Done.')
+    "
+  clusterSelector:
+    ray.io/cluster: raycluster-historyserver