[History Server][Feat] Add metadata collect logic

historyserver/cmd/collector/main.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"flag"
+	"fmt"
 	"os"
 	"path"
+	"strconv"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -28,6 +30,7 @@ func main() {
 	logBatching := 1000
 	eventsPort := 8080
 	pushInterval := time.Minute
+	supportUnsupportedData := true
 	runtimeClassConfigPath := "/var/collector-config/data"
 
 	flag.StringVar(&role, "role", "Worker", "")
@@ -42,6 +45,20 @@ func main() {
 
 	flag.Parse()
 
+	// Read SUPPORT_RAY_EVENT_UNSUPPORTED_DATA environment variable
+	if envValue := os.Getenv("SUPPORT_RAY_EVENT_UNSUPPORTED_DATA"); 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())
@@ -73,14 +90,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,6 +75,8 @@ 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;"
@@ -161,6 +163,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
@@ -182,6 +184,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/pkg/collector/logcollector/runtime/logcollector/collector.go

@@ -17,29 +17,34 @@ import (
 	"github.com/fsnotify/fsnotify"
 	"github.com/sirupsen/logrus"
 
+	"github.com/ray-project/kuberay/historyserver/pkg/collector/types"
 	"github.com/ray-project/kuberay/historyserver/pkg/storage"
 	"github.com/ray-project/kuberay/historyserver/pkg/utils"
 )
 
 type RayLogHandler struct {
-	Writer                 storage.StorageWriter
-	LogFiles               chan string
-	HttpClient             *http.Client
-	ShutdownChan           chan struct{}
-	logFilePaths           map[string]bool
-	MetaDir                string
-	RayClusterName         string
-	LogDir                 string
-	RayNodeName            string
-	RayClusterID           string
-	RootDir                string
-	SessionDir             string
-	prevLogsDir            string
-	persistCompleteLogsDir string
-	PushInterval           time.Duration
-	LogBatching            int
-	filePathMu             sync.Mutex
-	EnableMeta             bool
+	Writer                       storage.StorageWriter
+	LogFiles                     chan string
+	HttpClient                   *http.Client
+	ShutdownChan                 chan struct{}
+	logFilePaths                 map[string]bool
+	MetaDir                      string
+	RayClusterName               string
+	LogDir                       string
+	RayNodeName                  string
+	RayClusterID                 string
+	RootDir                      string
+	SessionDir                   string
+	prevLogsDir                  string
+	persistCompleteLogsDir       string
+	PushInterval                 time.Duration
+	LogBatching                  int
+	filePathMu                   sync.Mutex
+	EnableMeta                   bool
+	DashboardAddress             string
+	SupportRayEventUnSupportData bool
+	MetaCommonUrlInfo            []*types.UrlInfo
+	JobsUrlInfo                  *types.UrlInfo
 }
 
 func (r *RayLogHandler) Start(stop <-chan struct{}) error {
@@ -73,6 +78,10 @@ func (r *RayLogHandler) Run(stop <-chan struct{}) error {
 	if r.EnableMeta {
 		go r.WatchSessionLatestLoops() // Watch session_latest symlink changes
 	}
+	//Todo(alex): This should be removed when Ray core implemented events for placement groups, applications, and datasets
+	if r.SupportRayEventUnSupportData {
+		go r.PersistMetaLoop(stop)
+	}
 
 	select {
 	case <-sigChan: