add the implementation of historyserver collector #4241

KunWuLuan · 2025-12-01T07:04:38Z

add the implementation of historyserver collector

Why are these changes needed?

Related issue number

#3966
#4274

Checks

I've made sure the tests are passing.
Testing Strategy
- Unit tests
- Manual tests
- This PR is not tested :(

update go.work go.mod Signed-off-by: KunWuLuan <[email protected]>

Signed-off-by: KunWuLuan <[email protected]>

Future-Outlier · 2025-12-03T16:18:49Z

historyserver/pkg/collector/eventserver/eventserver.go

+
+	// 分类事件
+	for _, event := range eventsToFlush {
+		hourKey := event.Timestamp.Truncate(time.Hour).Format("2006-01-02-15")


should we be "2006-01-02-15" or "2006010215"?

Since this is different from the design doc
ray-project/enhancements#62

Fixed at KunWuLuan@0ed5bbb.

machichima · 2025-12-04T12:51:00Z

historyserver/cmd/collector/main.go

+	"github.com/sirupsen/logrus"
+)
+
+const runtimeClassConfigPath = "/var/collector-config/data"


Seems like it's not being used?

Fixed at KunWuLuan@c5ba659.

machichima · 2025-12-04T13:16:33Z

historyserver/pkg/utils/utils.go

+	for i := 0; i < 12; i++ {
+		rp, err := os.Readlink(session_latest_path)
+		if err != nil {
+			logrus.Errorf("read session_latest file error %v", err)
+			time.Sleep(time.Second * 5)
+			continue
+		}


Looks like we are doing the retry logic here. Could we extract the maxAttempts into constant or something we can configure, and maybe use backoff here?

And curious why we set the max retry to 12 times here? I think it may be too many times for retry, can we use 3?

Fixed at KunWuLuan@fafae16.

machichima · 2025-12-04T13:21:01Z

historyserver/pkg/utils/utils.go

+const (
+	RAY_SESSIONDIR_LOGDIR_NAME  = "logs"
+	RAY_SESSIONDIR_METADIR_NAME = "meta"
+)
+
+const (
+	OssMetaFile_BasicInfo = "ack__basicinfo"
+
+	OssMetaFile_NodeSummaryKey                        = "restful__nodes_view_summary"
+	OssMetaFile_Node_Prefix                           = "restful__nodes_"
+	OssMetaFile_JOBTASK_DETAIL_Prefix                 = "restful__api__v0__tasks_detail_job_id_"
+	OssMetaFile_JOBTASK_SUMMARIZE_BY_FUNC_NAME_Prefix = "restful__api__v0__tasks_summarize_by_func_name_job_id_"
+	OssMetaFile_JOBTASK_SUMMARIZE_BY_LINEAGE_Prefix   = "restful__api__v0__tasks_summarize_by_lineage_job_id_"
+	OssMetaFile_JOBDATASETS_Prefix                    = "restful__api__data__datasets_job_id_"
+	OssMetaFile_NodeLogs_Prefix                       = "restful__api__v0__logs_node_id_"
+	OssMetaFile_ClusterStatus                         = "restful__api__cluster_status"
+	OssMetaFile_LOGICAL_ACTORS                        = "restful__logical__actors"
+	OssMetaFile_ALLTASKS_DETAIL                       = "restful__api__v0__tasks_detail"
+	OssMetaFile_Events                                = "restful__events"
+	OssMetaFile_PlacementGroups                       = "restful__api__v0__placement_groups_detail"
+
+	OssMetaFile_ClusterSessionName = "static__api__cluster_session_name"
+
+	OssMetaFile_Jobs         = "restful__api__jobs"
+	OssMetaFile_Applications = "restful__api__serve__applications"
+)


In kuberay we usually put constant in a separate file. For example ray-operator put the constant in ray-operator/controllers/ray/utils/constant.go
Could we move those into a constant.go file?

Fixed at KunWuLuan@cef9755.

I'd like to discuss naming convention with you if you're available. Thanks.

machichima · 2025-12-04T13:21:44Z

historyserver/pkg/utils/utils.go

+}
+
+func GetSessionDir() (string, error) {
+	session_latest_path := "/tmp/ray/session_latest"


Could we also move this to constant?

Fixed at KunWuLuan@cef9755.

Future-Outlier · 2025-12-04T15:36:49Z

historyserver/pkg/collector/logcollector/storage/interface.go

+// StorageWriter is the interface for storage writer.
+type StorageWritter interface {


Can StorageWritter be StorageWriter?

Fixed at 99fd96f.

Future-Outlier · 2025-12-04T15:41:59Z

historyserver/pkg/collector/types/types.go

+	PushInterval   time.Duration
+}


Where is PushInterval be used?

Saw this line in NewCollector, but indeed never used:

kuberay/historyserver/pkg/collector/logcollector/runtime/runtime.go

Line 29 in c83f5e2

PushInterval: config.PushInterval,

Future-Outlier · 2025-12-04T16:02:29Z

historyserver/pkg/collector/types/types.go

+	Role           string
+	RayClusterName string
+	RayClusterID   string
+	LogBatching    int


Where is this be used?

Saw this line in NewCollector, but indeed never used:

kuberay/historyserver/pkg/collector/logcollector/runtime/runtime.go

Line 28 in c83f5e2

LogBatching: config.LogBatching,

fscnick · 2025-12-06T03:43:11Z

historyserver/Dockerfile

+ARG TARGETARCH
+
+FROM --platform=$BUILDPLATFORM golang:1.25.1 as builder
+ENV GOPROXY=https://goproxy.cn,direct


Should it be the official proxy? https://proxy.golang.org/

ref: https://proxy.golang.org/

Nice catch, make sense!

Fixed at KunWuLuan@09697d5.

fscnick · 2025-12-06T06:47:44Z

historyserver/Dockerfile

+
+RUN apt-get update && apt-get upgrade -y && rm -rf /var/cache/apt/ && apt-get install -y ca-certificates
+
+COPY --from=builder /historyserver/output/bin/historyserver /usr/local/bin/historyserver


The historyserver is not ready in this pr. Should it be removed?

yes it should

Fixed at KunWuLuan@7d267d9.

fscnick · 2025-12-06T06:49:11Z

historyserver/Dockerfile

+COPY --from=builder /historyserver/dashboard/v2.51.0/client/build /dashboard/v2.51.0/client/build
+COPY --from=builder /historyserver/dashboard/homepage /dashboard/homepage


Should these COPY consider BUILD_RAYSERVER_DASHBOARD? Or, it might be failed on building.

fscnick · 2025-12-06T06:52:25Z

historyserver/Makefile

+	rm -rf $(OUT_DIR)
+
+.PHONY: build
+build: buildcollector buildhistoryserver


should the buildhistoryserver be removed because it is not ready in this pr? And so does the historyserver related content in this file.

Yes, we need to remove all snippets related to building history server binary to keep this PR focused.

Fixed at KunWuLuan@06175f4.

fscnick · 2025-12-06T07:03:56Z

historyserver/cmd/collector/main.go

+	flag.StringVar(&role, "role", "Worker", "")
+	flag.StringVar(&runtimeClassName, "runtime-class-name", "", "")
+	flag.StringVar(&rayClusterName, "ray-cluster-name", "", "")
+	flag.StringVar(&rayClusterId, "ray-cluster-id", "default", "")
+	flag.StringVar(&rayRootDir, "ray-root-dir", "", "")
+	flag.IntVar(&logBatching, "log-batching", 1000, "")
+	flag.IntVar(&eventsPort, "events-port", 8080, "")
+	flag.StringVar(&runtimeClassConfigPath, "runtime-class-config-path", "", "") //"/var/collector-config/data"
+	flag.DurationVar(&pushInterval, "push-interval", time.Minute, "")


It might be better to have some usage description.

fscnick · 2025-12-08T15:13:06Z

historyserver/cmd/collector/main.go

+	eventServer := eventserver.NewEventServer(writter, rayRootDir, sessionDir, rayNodeId, rayClusterName, rayClusterId, sessionName)
+	eventServer.InitServer(eventsPort)
+
+	collector := runtime.NewCollector(&globalConfig, writter)


The collector seems collide with the package name.

Should be renamed to logCollector to avoid confusion.

Fixed at KunWuLuan@2aa3eb1.

fscnick · 2025-12-08T15:25:44Z

historyserver/cmd/collector/main.go

+	eventServer.InitServer(eventsPort)
+
+	collector := runtime.NewCollector(&globalConfig, writter)
+	_ = collector.Start(context.TODO().Done())


If I didn't get it wrong, it seems no place to cancel the context. Do you mean graceful shutdown here?

fscnick · 2025-12-10T14:37:12Z

historyserver/pkg/collector/eventserver/eventserver.go

+
+// watchNodeIDFile 监听 /tmp/ray/raylet_node_id 文件内容变化
+func (es *EventServer) watchNodeIDFile() {
+	nodeIDFilePath := "/tmp/ray/raylet_node_id"


Could the /tmp/ray/raylet_node_id become a constant that it could be reused?

Fixed at KunWuLuan@9b2ca52. We reuse the const definition.

fscnick · 2025-12-10T15:06:23Z

historyserver/pkg/collector/logcollector/storage/s3/s3.go

+	if err != nil {
+		// Directory doesn't exist, create it
+		logrus.Infof("Begin to create s3 dir %s ...", objectDir)
+		_, err = r.S3Client.PutObject(&s3.PutObjectInput{
+			Bucket: aws.String(r.S3Bucket),
+			Key:    aws.String(objectDir),
+			Body:   bytes.NewReader([]byte("")),
+		})


Should it check the error if it is directory not existed before creating it?

Nice suggestion! Fixed at KunWuLuan@7a28442, aiming to handle different err types explicitly, including "NoSuchKey" error you just mentioned. Thanks!

win5923 · 2025-12-10T17:47:37Z

historyserver/pkg/collector/eventserver/eventserver.go

+		// 检查sessionName是否发生变化
+		if es.currentSessionName != sessionNameStr {
+			logrus.Infof("Session name changed from %s to %s, flushing events", es.currentSessionName, sessionNameStr)
+			// 保存当前事件后再刷新
+			eventsToFlush := make([]Event, len(es.events))
+			copy(eventsToFlush, es.events)
+
+			// 清空事件列表
+			es.events = es.events[:0]
+
+			// 更新当前sessionName
+			es.currentSessionName = sessionNameStr
+
+			// 解锁后执行刷新操作
+			es.mutex.Unlock()
+
+			// 刷新之前的事件
+			es.flushEventsInternal(eventsToFlush)
+			return


Should we send a response before returning? Right now this branch exits without writing any HTTP response, which will cause clients to hang.

Fixed at KunWuLuan@0337030. Thanks!

Future-Outlier · 2025-12-11T06:19:49Z

historyserver/Dockerfile

+
+RUN apt-get update && apt-get upgrade -y && rm -rf /var/cache/apt/ && apt-get install -y ca-certificates
+
+COPY --from=builder /historyserver/output/bin/historyserver /usr/local/bin/historyserver


yes it should

Future-Outlier · 2025-12-11T06:25:01Z

historyserver/pkg/collector/logcollector/storage/s3/config.go

+}
+
+func (c *config) complete(rcc *types.RayCollectorConfig, jd map[string]interface{}) {
+	c.RayCollectorConfig = *rcc
+	c.S3ID = os.Getenv("AWS_S3ID")
+	c.S3Secret = os.Getenv("AWS_S3SECRET")
+	c.S3Token = os.Getenv("AWS_S3TOKEN")
+	if len(jd) == 0 {
+		c.S3Bucket = os.Getenv("S3_BUCKET")
+		c.S3Endpoint = os.Getenv("S3_ENDPOINT")
+		c.S3Region = os.Getenv("S3_REGION")
+		if os.Getenv("S3FORCE_PATH_STYPE") != "" {
+			c.S3ForcePathStyle = aws.Bool(os.Getenv("S3FORCE_PATH_STYPE") == "true")
+		}
+		if os.Getenv("S3DISABLE_SSL") != "" {
+			c.DisableSSL = aws.Bool(os.Getenv("S3DISABLE_SSL") == "true")
+		}
+	} else {
+		if bucket, ok := jd["s3Bucket"]; ok {
+			c.S3Bucket = bucket.(string)
+		}
+		if endpoint, ok := jd["s3Endpoint"]; ok {
+			c.S3Endpoint = endpoint.(string)
+		}
+		if region, ok := jd["s3Region"]; ok {
+			c.S3Region = region.(string)
+		}
+		if forcePathStyle, ok := jd["s3ForcePathStyle"]; ok {
+			c.S3ForcePathStyle = aws.Bool(forcePathStyle.(string) == "true")
+		}
+		if s3disableSSL, ok := jd["S3DISABLE_SSL"]; ok {
+			c.DisableSSL = aws.Bool(s3disableSSL.(string) == "true")
+		}
+	}
+}
+
+func (c *config) completeHSConfig(rcc *types.RayHistoryServerConfig, jd map[string]interface{}) {
+	c.RayCollectorConfig = types.RayCollectorConfig{
+		RootDir: rcc.RootDir,
+	}
+	c.S3ID = os.Getenv("AWS_S3ID")
+	c.S3Secret = os.Getenv("AWS_S3SECRET")
+	c.S3Token = os.Getenv("AWS_S3TOKEN")
+	if len(jd) == 0 {


Can we add a TODO and talk about we should refactor complete and completeHSConfig to the same function?

Future-Outlier · 2025-12-12T07:55:23Z

historyserver/pkg/collector/eventserver/eventserver.go

+	totalEvents := len(eventsToFlush)
+	logrus.Infof("Successfully flushed %d events to storage (%d node events, %d job events)",
+		totalEvents,
+		countEventsInMap(nodeEventsByHour),
+		countEventsInMap(jobEventsByHour))
+}


for more correct information

Suggested change

totalEvents := len(eventsToFlush)

logrus.Infof("Successfully flushed %d events to storage (%d node events, %d job events)",

totalEvents,

countEventsInMap(nodeEventsByHour),

countEventsInMap(jobEventsByHour))

}

totalEvents := len(eventsToFlush)

errorCount := len(errChan)

if errorCount > 0 {

logrus.Warnf("Attempted to flush %d events to storage, but %d upload tasks failed (%d node events, %d job events)",

totalEvents, errorCount,

countEventsInMap(nodeEventsByHour),

countEventsInMap(jobEventsByHour))

} else {

logrus.Infof("Successfully flushed %d events to storage (%d node events, %d job events)",

totalEvents,

countEventsInMap(nodeEventsByHour),

countEventsInMap(jobEventsByHour))

}

Fixed at KunWuLuan@1fbfae0 to warn upon errors.

Copilot

Pull request overview

This pull request adds the implementation of a history server collector for Ray clusters. It introduces a comprehensive system for collecting, storing, and managing logs and metadata from Ray clusters using multiple storage backends (S3, Aliyun OSS, and local testing).

Key changes:

Collector service to gather logs and events from Ray nodes
Support for multiple storage backends (S3, Aliyun OSS, local test)
Event server for collecting and persisting Ray events
History server infrastructure for viewing collected data

Reviewed changes

Copilot reviewed 30 out of 34 changed files in this pull request and generated 21 comments.

Show a summary per file

File	Description
historyserver/go.mod	Declares Go module with dependencies for the history server implementation
go.work	Workspace configuration for multi-module Go project
historyserver/cmd/collector/main.go	Entry point for the collector service that runs as a sidecar
historyserver/pkg/collector/*	Core collector logic and storage interfaces
historyserver/pkg/utils/*	Utility functions for OSS operations and Ray cluster management
historyserver/Makefile	Build configuration for compiling binaries and Docker images
historyserver/Dockerfile	Multi-stage build for containerized deployment
historyserver/README.md	Documentation for building, configuring, and deploying the history server

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-12T14:33:37Z

historyserver/pkg/collector/logcollector/storage/aliyunoss/config/types.go

+	OSSHistoryServerDir string
+}
+
+type RayMetaHanderConfig struct {


The type name 'RayMetaHanderConfig' contains a typo. 'Hander' should be 'Handler'.

Suggested change

type RayMetaHanderConfig struct {

type RayMetaHandlerConfig struct {

Copilot · 2025-12-12T14:33:38Z

historyserver/pkg/collector/logcollector/storage/s3/config.go

+		if os.Getenv("S3FORCE_PATH_STYPE") != "" {
+			c.S3ForcePathStyle = aws.Bool(os.Getenv("S3FORCE_PATH_STYPE") == "true")


The environment variable name 'S3FORCE_PATH_STYPE' contains a typo. 'STYPE' should be 'STYLE'. This appears in multiple places (lines 34, 35, 70, 71) and needs to be corrected for proper environment variable reading.

Fixed at 99fd96f.

Copilot · 2025-12-12T14:33:38Z

historyserver/pkg/collector/logcollector/runtime/logcollector/collector.go

+	logFilePaths map[string]bool
+	filePathMu   sync.Mutex
+
+	// Channel for signaling shutdown
+	ShutdownChan chan struct{}


The comment uses Chinese characters. For consistency with the rest of the codebase and international collaboration, comments should be in English. Consider translating '用于跟踪当前sessionName' to 'Used to track the current sessionName' and '用于跟踪当前nodeId' to 'Used to track the current nodeId'.

Copilot · 2025-12-12T14:33:39Z

historyserver/pkg/utils/utils_test.go

+
+	clusterNameId = "a-s-sdf-sdfsdfsdfisfdf1A_B2safd-0sdf-sdf-000"
+	strs = strings.Split(clusterNameId, "_")
+	t.Logf("laster %s", strs[len(strs)-1])


The variable name 'laster' is misspelled. It should be 'last' instead of 'laster'.

Copilot · 2025-12-12T14:33:39Z

historyserver/pkg/utils/utils_test.go

+	clusterNameId = "a-s-sdf-sdfsdfsdfisfdf1A_B2safd-0sdf-sdf-000"
+	strs = strings.Split(clusterNameId, "_")
+	t.Logf("laster %s", strs[len(strs)-1])
+	t.Logf("not laster all %s", strings.Join(strs[:len(strs)-1], "_"))


The variable name 'laster' is misspelled. It should be 'last' instead of 'laster'.

Copilot · 2025-12-12T14:33:43Z

historyserver/pkg/utils/utils_test.go

+
+func TestSlice(t *testing.T) {
+	strs := []string{"1", "2", "3", "4", "5"}
+	t.Logf("laster %s", strs[len(strs)-1])


The variable name 'laster' is misspelled. It should be 'last' instead of 'laster'.

Copilot · 2025-12-12T14:33:43Z

historyserver/pkg/collector/types/types.go

+	PushInterval   time.Duration
+}
+
+// ValidateRayHanderConfig is


The function name 'ValidateRayHanderConfig' contains a typo. 'Hander' should be 'Handler'.

Copilot · 2025-12-12T14:33:43Z

historyserver/pkg/collector/registry.go

+type WriterRegistry map[string]func(globalData *types.RayCollectorConfig, data map[string]interface{}) (storage.StorageWritter, error)
+
+func GetWriterRegistry() WriterRegistry {
+	return writerRegistry
+}
+
+var writerRegistry = WriterRegistry{
+	"aliyunoss": ray.NewWritter,
+	"s3":        s3.NewWritter,


The type name 'WriterRegistry' maps to functions returning 'StorageWritter', but 'Writter' is misspelled. It should be 'Writer' throughout the codebase for consistency and correctness.

Suggested change

type WriterRegistry map[string]func(globalData *types.RayCollectorConfig, data map[string]interface{}) (storage.StorageWritter, error)

func GetWriterRegistry() WriterRegistry {

return writerRegistry

}

var writerRegistry = WriterRegistry{

"aliyunoss": ray.NewWritter,

"s3": s3.NewWritter,

type WriterRegistry map[string]func(globalData *types.RayCollectorConfig, data map[string]interface{}) (storage.StorageWriter, error)

func GetWriterRegistry() WriterRegistry {

return writerRegistry

}

var writerRegistry = WriterRegistry{

"aliyunoss": ray.NewWriter,

"s3": s3.NewWriter,

Copilot · 2025-12-12T14:33:44Z

historyserver/pkg/collector/logcollector/storage/interface.go

+)
+
+// StorageWriter is the interface for storage writer.
+type StorageWritter interface {


The interface name 'StorageWritter' is misspelled. It should be 'StorageWriter' with one 't'. This typo affects the entire codebase and should be corrected for consistency and professionalism.

Suggested change

type StorageWritter interface {

type StorageWriter interface {

Copilot · 2025-12-12T14:33:44Z

historyserver/Dockerfile

+FROM --platform=$BUILDPLATFORM golang:1.25.1 as builder
+ENV GOPROXY=https://goproxy.cn,direct
+ARG BUILD_RAYSERVER_DASHBOARD 
+
+RUN if [ "$BUILD_RAYSERVER_DASHBOARD" = "yes" ] ; then \
+ curl -o install.sh https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh && chmod +x install.sh && ./install.sh && /bin/bash -c "source $HOME/.nvm/nvm.sh && nvm install 14 && nvm use 14" ;\
+else \
+ echo "$BUILD_RAYSERVER_DASHBOARD not yes, no need install nvm"; \
+fi


The Dockerfile downloads and executes a remote installation script via curl from https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh without any integrity verification, which introduces a supply-chain risk: if the remote script or delivery channel is compromised, arbitrary code can run in your build environment and be baked into produced images. An attacker who can tamper with that script (e.g., via GitHub compromise, DNS/CA issues, or MITM) could inject backdoors into the historyserver/collector binaries or exfiltrate build secrets. To mitigate this, avoid executing remote scripts directly; instead vendor the script or use a package-manager-installed nvm, and if you must download it, pin to an immutable reference and verify a known checksum or signature before execution.

400Ping · 2025-12-12T15:44:26Z

historyserver/pkg/collector/logcollector/runtime/logcollector/collector.go

+	// Walk through the logs directory and process all files
+	err := filepath.WalkDir(logsDir, func(path string, info fs.DirEntry, err error) error {
+		if err != nil {
+			logrus.Errorf("Error walking logs path %s: %v", path, err)
+			return nil
+		}
+
+		// Skip directories
+		if info.IsDir() {
+			return nil
+		}
+
+		// Process log file
+		if err := r.processPrevLogFile(path, logsDir, sessionID, nodeID); err != nil {
+			logrus.Errorf("Failed to process prev-log file %s: %v", path, err)
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		logrus.Errorf("Error walking logs directory %s: %v", logsDir, err)
+		return
+	}
+
+	// After successfully processing all files, remove the node directory
+	logrus.Infof("Finished processing all logs for session: %s, node: %s. Removing node directory.", sessionID, nodeID)
+	if err := os.RemoveAll(sessionNodeDir); err != nil {
+		logrus.Errorf("Failed to remove node directory %s: %v", sessionNodeDir, err)
+	} else {
+		logrus.Infof("Successfully removed node directory: %s", sessionNodeDir)
+	}


We delete sessionNodeDir unconditionally after the walk, even if any file upload failed. To avoid losing logs on a transient storage error, consider: bubble up failures from the walk (return err from the callback), track a per-dir success flag, and only RemoveAll when all files were written/moved. Alternatively, add a retry path (e.g., leave the dir in place and mark it for retry) instead of deleting on first pass.

Fixed at KunWuLuan@014a6ff.

For now, we use processingErrors to record file processing errors in the callback function. Then, those files can be processed again in the next retry. WDYT?

400Ping · 2025-12-12T15:51:14Z

historyserver/pkg/collector/logcollector/runtime/logcollector/collector.go

+	select {
+	case <-sigChan:
+		logrus.Info("Received SIGTERM, processing all logs...")
+		// r.processAllLogs()
+		r.processSessionLatestLogs()
+		// r.processPrevLogsOnShutdown()
+		close(r.ShutdownChan)
+	case <-stop:
+		logrus.Info("Received stop signal, processing all logs...")
+		// r.processAllLogs()
+		r.processSessionLatestLogs()
+		// r.processPrevLogsOnShutdown()
+		close(r.ShutdownChan)
+	}


Can we coordinate shutdown so background uploads finish before exit? Suggest: add a context.WithCancel + WaitGroup, pass the ctx into WatchPrevLogsLoops/processPrevLogsDir, wg.Add/Done in each goroutine, then on stop call cancel() and wg.Wait() (optionally with a timeout) before closing ShutdownChan. That way uploads/moves complete (or time out) instead of being cut off mid-flight.

400Ping · 2025-12-12T16:01:08Z

historyserver/pkg/collector/logcollector/storage/s3/config.go

+		if os.Getenv("S3FORCE_PATH_STYPE") != "" {
+			c.S3ForcePathStyle = aws.Bool(os.Getenv("S3FORCE_PATH_STYPE") == "true")


andrewsykim · 2025-12-13T20:07:54Z

historyserver/cmd/collector/main.go

+const runtimeClassConfigPath = "/var/collector-config/data"
+
+func main() {
+	role := ""


nit: Use var role string format, same for other lines

Fixed at KunWuLuan@55b9a92. Thanks!

andrewsykim · 2025-12-13T20:11:18Z

historyserver/cmd/collector/main.go

+
+	sessionDir, err := utils.GetSessionDir()
+	if err != nil {
+		panic("Failed to get session dir: " + err.Error())


Instead of panic, log the error and call os.Exit(1)

See example https://github.com/ray-project/kuberay/blob/master/ray-operator/main.go#L315-L320

Fixed at KunWuLuan@4e9b8bb.

andrewsykim · 2025-12-13T20:11:45Z

historyserver/cmd/collector/main.go

+
+	rayNodeId, err := utils.GetRayNodeID()
+	if err != nil {
+		panic("Failed to get ray node id: " + err.Error())


same here, os.Exit(1), see example https://github.com/ray-project/kuberay/blob/master/ray-operator/main.go#L315-L320

Fixed at KunWuLuan@4e9b8bb.

andrewsykim · 2025-12-13T20:15:12Z

historyserver/pkg/collector/registry.go

+
+var writerRegistry = WriterRegistry{
+	"aliyunoss": ray.NewWritter,
+	"s3":        s3.NewWritter,


Is this any s3 compatible storage solution? If not please also add Google Cloud Storage (gcs)

andrewsykim · 2025-12-13T20:18:20Z

historyserver/pkg/collector/logcollector/storage/s3/config.go

+	}
+}
+
+func (c *config) completeHSConfig(rcc *types.RayHistoryServerConfig, jd map[string]interface{}) {


It's been a long time since I worked with AWS API / SDKs, but isn't there an AWS Go library that handles all the authentication checks. That seems like a better approach than trying to load envirnoment variables ourselves

andrewsykim · 2025-12-13T20:21:58Z

historyserver/pkg/collector/logcollector/storage/localtest/reader.go

+		},
+	}
+
+	data := map[string]map[string]string{


Instead of hardcoding data here, have NewMockReader receive the data as a parameter. Same thing for clusters

Fixed at KunWuLuan@8b2d34a.

andrewsykim · 2025-12-13T20:22:22Z

historyserver/pkg/utils/types.go

+	Name            string `json:"name"`
+	Namespace       string `json:"namespace"`
+	SessionName     string `json:"sessionName"`
+	CreateTime      string `json:"createTime"`


nit: CreationTime

Fixed at KunWuLuan@a81027f.

andrewsykim · 2025-12-13T20:22:31Z

historyserver/pkg/utils/types.go

+	Namespace       string `json:"namespace"`
+	SessionName     string `json:"sessionName"`
+	CreateTime      string `json:"createTime"`
+	CreateTimeStamp int64  `json:"createTimeStamp"`


nit: CreationTimestamp

Also, why do we need both CreateTime and CreateTimestamp?

I think CreationTime is more human-friendly and can be displayed directly in the UI. In contrast, CreationTimestamp is more machine-oriented and better suited for sorting and comparisons, for example:

kuberay/historyserver/pkg/utils/types.go

Line 31 in 7b2acae

func (a ClusterInfoList) Less(i, j int) bool { return a[i].CreateTimeStamp > a[j].CreateTimeStamp } // 降序排序

If this feels redundant, we can discuss which one to keep to improve maintainability. Thanks!

andrewsykim · 2025-12-13T20:23:12Z

historyserver/pkg/utils/utils.go

+	OssMetaFile_Node_Prefix                           = "restful__nodes_"
+	OssMetaFile_JOBTASK_DETAIL_Prefix                 = "restful__api__v0__tasks_detail_job_id_"
+	OssMetaFile_JOBTASK_SUMMARIZE_BY_FUNC_NAME_Prefix = "restful__api__v0__tasks_summarize_by_func_name_job_id_"
+	OssMetaFile_JOBTASK_SUMMARIZE_BY_LINEAGE_Prefix   = "restful__api__v0__tasks_summarize_by_lineage_job_id_"


Replace Oss with Ray for all these environment variables

andrewsykim · 2025-12-13T20:24:46Z

historyserver/Dockerfile

+fi
+
+WORKDIR /historyserver
+COPY . .


I would try to avoid COPY . ., please only add directories that contain source code needed for the build. See example here: https://github.com/ray-project/kuberay/blob/master/ray-operator/Dockerfile#L12-L19

andrewsykim · 2025-12-13T20:27:29Z

historyserver/pkg/collector/eventserver/eventserver.go

@@ -0,0 +1,519 @@
+package eventserver


Can we add eventserver in a separate PR? I think @chiayi started this here: #4253

Is there a dependency to eventserver required for log collector?

Fixed at KunWuLuan@1aea830. We've removed eventserver-related files to keep this PR focused on the log collector!

Signed-off-by: Future-Outlier <[email protected]>

Future-Outlier · 2025-12-15T13:20:13Z

Hi, @andrewsykim

After discussion with @kevin85421, @rueian , and @KunWuLuan, we will address your comment in a follow-up PR and merge this ASAP. This will allow other contributors to iterate on the collector together, and we will not block other PRs such as the event processor by Aaron.

Signed-off-by: Future-Outlier <[email protected]>

Future-Outlier

let's merge this first so that

other people can work on the improvement in pararrel, see here #4274
other people can develop other component (like event server, history server's web server, and ray dashboard) more easily

Future-Outlier · 2025-12-17T07:48:28Z

cc @JiangJiaWei1103 , can you open a PR to address this PR's comment?

Future-Outlier · 2025-12-17T07:50:49Z

let's merge this first so that

other people can work on the improvement in pararrel, see here [Epic][Feature] history server collctor #4274

other people can develop other component (like event server, history server's web server, and ray dashboard) more easily

update: let's wait for @andrewsykim to take a look

JiangJiaWei1103 · 2025-12-17T08:33:00Z

cc @JiangJiaWei1103 , can you open a PR to address this PR's comment?

No problem, let's go!

JiangJiaWei1103 · 2025-12-18T23:33:55Z

Hi @andrewsykim and @Future-Outlier,

As AWS SDK for Go v1 has come to the end-of-support phase since July 31, 2025 [1]. Would it make sense to switch to actively maintained v2 [2] in this PR? Thanks.

[1] Announcing end-of-support for AWS SDK for Go (v1) effective July 31, 2025
[2] AWS SDK for Go v2

andrewsykim · 2025-12-22T02:09:23Z

historyserver/pkg/collector/logcollector/storage/aliyunoss/ray/ray.go

@@ -0,0 +1,295 @@
+// Package ray is


incomplete comment?

suggest just removing it, we don't usually need package level comments

andrewsykim · 2025-12-22T02:10:16Z

historyserver/cmd/collector/main.go

+	runtimeClassConfigPath := "/var/collector-config/data"
+
+	flag.StringVar(&role, "role", "Worker", "")
+	flag.StringVar(&runtimeClassName, "runtime-class-name", "", "")


I think there was a previous comment that was resolved to name this to storage-provider instead of runtime-class, is that still the case?

andrewsykim · 2025-12-22T02:11:56Z

As AWS SDK for Go v1 has come to the end-of-support phase since July 31, 2025 [1]. Would it make sense to switch to actively maintained v2 [2] in this PR? Thanks.

Using latest SDK makes sense to me unless there are technical issues with v2. Is the plan to update this PR to use the SDK or is that follow-up work?

andrewsykim · 2025-12-22T02:12:51Z

historyserver/Dockerfile

+fi
+
+WORKDIR /historyserver
+COPY . .


Avoid COPY on entire directories. Please only add packages containing source code required to compile binaries

andrewsykim · 2025-12-22T02:14:01Z

historyserver/pkg/collector/logcollector/storage/s3/config.go

+		RootDir: rcc.RootDir,
+	}
+	c.S3ID = os.Getenv("AWS_S3ID")
+	c.S3Secret = os.Getenv("AWS_S3SECRET")


Are S3 secret and token environment variables here still required if we use official SDK? I imagine the SDK can handle that?

andrewsykim · 2025-12-22T02:17:20Z

Spoke with @Future-Outlier and @rueian offline. Let's merge this PR as-is and address remaining comments in a follow-up PR.

Can we fix the merge conflict to unblock merge?

Signed-off-by: Future-Outlier <[email protected]>

JiangJiaWei1103 · 2025-12-22T02:27:14Z

Spoke with @Future-Outlier and @rueian offline. Let's merge this PR as-is and address remaining comments in a follow-up PR.

Can we fix the merge conflict to unblock merge?

Thanks Andrew. We're actively addressing review comments at the moment [1] [2].

[1] KunWuLuan#1
[2] KunWuLuan#2

* add the implementation of historyserver collector update go.work go.mod Signed-off-by: KunWuLuan <[email protected]> * update the func judging if the event is releated to the Nodes. Signed-off-by: KunWuLuan <[email protected]> * S3FORCE_PATH_STYPE -> S3FORCE_PATH_STYLE Signed-off-by: Future-Outlier <[email protected]> * S3DISABLE_SSL -> s3DisableSSL (camel case) Signed-off-by: Future-Outlier <[email protected]> * Add comments to explain WatchSessionLatestLoops Signed-off-by: Future-Outlier <[email protected]> * update Signed-off-by: Future-Outlier <[email protected]> * update Signed-off-by: Future-Outlier <[email protected]> * update Signed-off-by: Future-Outlier <[email protected]> --------- Signed-off-by: KunWuLuan <[email protected]> Signed-off-by: Future-Outlier <[email protected]> Co-authored-by: Future-Outlier <[email protected]>

Future-Outlier · 2026-01-07T05:23:30Z

cursor review

cursor

✅ Bugbot reviewed your changes and found no bugs!

Future-Outlier · 2026-01-14T07:16:04Z

cursor review

cursor · 2026-01-14T07:17:33Z

historyserver/pkg/collector/logcollector/storage/s3/config.go

+			c.S3ForcePathStyle = aws.Bool(forcePathStyle.(string) == "true")
+		}
+		if s3disableSSL, ok := jd["s3DisableSSLs3DisableSSL"]; ok {
+			c.DisableSSL = aws.Bool(s3disableSSL.(string) == "true")


Duplicated key name in JSON config parsing

High Severity

The JSON key "s3DisableSSLs3DisableSSL" in the complete function is malformed - it appears to be an accidental duplication of "s3DisableSSL". This typo prevents the DisableSSL configuration from being correctly read from JSON data, causing the SSL disable setting to be silently ignored when configured via JSON. The correct key would be "s3DisableSSL" to match the pattern used elsewhere in the file (line 87).

add the implementation of historyserver collector

091caec

update go.work go.mod Signed-off-by: KunWuLuan <[email protected]>

Future-Outlier self-assigned this Dec 1, 2025

KunWuLuan mentioned this pull request Dec 1, 2025

[Epic][Feature] Support History Server #3966

Open

2 tasks

update the func judging if the event is releated to the Nodes.

7b2acae

Signed-off-by: KunWuLuan <[email protected]>

Future-Outlier reviewed Dec 3, 2025

View reviewed changes

machichima reviewed Dec 4, 2025

View reviewed changes

Future-Outlier reviewed Dec 4, 2025

View reviewed changes

fscnick reviewed Dec 6, 2025

View reviewed changes

fscnick reviewed Dec 8, 2025

View reviewed changes

fscnick reviewed Dec 10, 2025

View reviewed changes

win5923 reviewed Dec 10, 2025

View reviewed changes

Future-Outlier reviewed Dec 12, 2025

View reviewed changes

Future-Outlier requested a review from Copilot December 12, 2025 14:25

Copilot started reviewing on behalf of Future-Outlier December 12, 2025 14:26 View session

Copilot AI reviewed Dec 12, 2025

View reviewed changes

400Ping reviewed Dec 12, 2025

View reviewed changes

andrewsykim reviewed Dec 13, 2025

View reviewed changes

JiangJiaWei1103 mentioned this pull request Dec 14, 2025

[WIP] Add historyserver log collector #3973

Closed

4 tasks

Future-Outlier added 4 commits December 15, 2025 16:39

S3FORCE_PATH_STYPE -> S3FORCE_PATH_STYLE

99fd96f

Signed-off-by: Future-Outlier <[email protected]>

S3DISABLE_SSL -> s3DisableSSL (camel case)

65aa6ce

Signed-off-by: Future-Outlier <[email protected]>

Add comments to explain WatchSessionLatestLoops

0761969

Signed-off-by: Future-Outlier <[email protected]>

update

ab9695b

Signed-off-by: Future-Outlier <[email protected]>

Future-Outlier added 3 commits December 15, 2025 21:32

update

47db5ac

Signed-off-by: Future-Outlier <[email protected]>

Merge branch 'master' into historyserver-beta-collector

95dda91

update

c83f5e2

Signed-off-by: Future-Outlier <[email protected]>

Future-Outlier mentioned this pull request Dec 16, 2025

[Epic][Feature][history server] collector #4274

Open

1 task

Future-Outlier approved these changes Dec 17, 2025

View reviewed changes

Future-Outlier mentioned this pull request Dec 17, 2025

[history server][collector] Move storage's interface to correct order #4278

Closed

JiangJiaWei1103 mentioned this pull request Dec 18, 2025

[WIP] Address review comments KunWuLuan/kuberay#1

Open

4 tasks

andrewsykim reviewed Dec 22, 2025

View reviewed changes

Merge branch 'master' into historyserver-beta-collector

f2dfa4d

Signed-off-by: Future-Outlier <[email protected]>

andrewsykim merged commit 21a7348 into ray-project:master Dec 22, 2025
24 of 27 checks passed

JiangJiaWei1103 mentioned this pull request Dec 22, 2025

[history server] [collector] Address log collector's review comments #4304

Open

JiangJiaWei1103 mentioned this pull request Jan 3, 2026

[history server] [collector] Fix collector 4241 review comments #4325

Draft

4 tasks

cursor bot reviewed Jan 7, 2026

View reviewed changes

cursor bot reviewed Jan 14, 2026

View reviewed changes

		// StorageWriter is the interface for storage writer.
		type StorageWritter interface {


		RUN apt-get update && apt-get upgrade -y && rm -rf /var/cache/apt/ && apt-get install -y ca-certificates

		COPY --from=builder /historyserver/output/bin/historyserver /usr/local/bin/historyserver

		COPY --from=builder /historyserver/dashboard/v2.51.0/client/build /dashboard/v2.51.0/client/build
		COPY --from=builder /historyserver/dashboard/homepage /dashboard/homepage No newline at end of file

	type RayMetaHanderConfig struct {
	type RayMetaHandlerConfig struct {

		if os.Getenv("S3FORCE_PATH_STYPE") != "" {
		c.S3ForcePathStyle = aws.Bool(os.Getenv("S3FORCE_PATH_STYPE") == "true")

	type StorageWritter interface {
	type StorageWriter interface {

add the implementation of historyserver collector #4241

add the implementation of historyserver collector #4241

Uh oh!

Conversation

KunWuLuan commented Dec 1, 2025 • edited by Future-Outlier Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why are these changes needed?

Related issue number

Checks

Uh oh!

Future-Outlier Dec 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JiangJiaWei1103 Dec 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JiangJiaWei1103 Dec 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JiangJiaWei1103 Dec 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KunWuLuan commented Dec 1, 2025 •

edited by Future-Outlier

Loading

Future-Outlier Dec 3, 2025 •

edited

Loading

JiangJiaWei1103 Dec 17, 2025 •

edited

Loading

JiangJiaWei1103 Dec 17, 2025 •

edited

Loading

JiangJiaWei1103 Dec 17, 2025 •

edited

Loading

400Ping Dec 12, 2025 •

edited

Loading