[WIP] Add historyserver log collector #3973

KunWuLuan · 2025-08-19T12:26:01Z

Why are these changes needed?

Related issue number

To #3966

Checks

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

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

Future-Outlier · 2025-08-22T01:59:47Z

historyserver/utils/utils.go

+		// 列举所有包含指定前缀的文件并删除。
+		marker := oss.Marker("")
+		// 如果您仅需要删除src目录及目录下的所有文件，则prefix设置为src/。
+		prefix := oss.Prefix(objectDir)


Can we use english comments?

Future-Outlier · 2025-08-25T02:12:20Z

historyserver/backend/collector/storage/interface.go

+type StorageWritter interface {
+	CreateDirectory(path string) error


Can we rename StorageWritter to StorageWriter?

Suggested change

type StorageWritter interface {

CreateDirectory(path string) error

type StorageWriter interface {

CreateDirectory(path string) error

Future-Outlier · 2025-08-25T02:12:47Z

historyserver/backend/types/types.go

+}
+
+// ValidateRayHanderConfig is
+func ValidateRayHanderConfig(c *RayCollectorConfig, fldpath *field.Path) field.ErrorList {


Can we rename ValidateRayHanderConfig to ValidateRayHandlerConfig?

Future-Outlier · 2025-08-25T02:16:57Z

historyserver/backend/collector/storage/aliyunoss/ray/config.go

+	c.RayCollectorConfig = *rcc
+	c.OSSBucket = jd["ossBucket"].(string)
+	c.OSSBucket = jd["ossEndpoint"].(string)
+	c.OSSHistoryServerDir = jd["ossHistoryServerDir"].(string)


Should we avoid panic? or is this be designed in purpose?

if bucket, ok := jd["ossBucket"].(string); ok { c.OSSBucket = bucket } else { return fmt.Errorf("invalid ossBucket type") }

Future-Outlier · 2025-08-25T02:17:20Z

historyserver/backend/collector/storage/aliyunoss/ray/ray.go

+Copyright 2024 by the zhangjie [email protected] Authors.
+


Suggested change

Copyright 2024 by the zhangjie bingyu.zj@alibaba-inc.com Authors.

Copyright 2025 by the zhangjie bingyu.zj@alibaba-inc.com Authors.

Future-Outlier · 2025-08-25T02:21:15Z

historyserver/backend/types/types.go

+	if c.Role == "Head" {
+		if len(c.RayClusterName) == 0 {
+			allErrs = append(allErrs, field.Invalid(fldpath, c.RayClusterName, "ray_cluster_name must be set"))
+		}
+		if len(c.RayClusterID) == 0 {
+			allErrs = append(allErrs, field.Invalid(fldpath, c.RayClusterID, "ray_cluster_id must be set"))
+		}
+	}


This can be deleted in my opinion, since it’s checked above.

Future-Outlier · 2025-08-25T02:25:28Z

historyserver/backend/types/types.go

+			allErrs = append(allErrs, field.Invalid(fldpath, c.RayClusterID, "ray_cluster_id must be set"))
+		}
+	}
+	return allErrs


probably add validation logic for LogBatching and PushInterval?

if c.LogBatching <= 0 { allErrs = append(allErrs, field.Invalid(...)) } if c.PushInterval <= 0 { allErrs = append(allErrs, field.Invalid(...)) }

rueian · 2025-08-29T06:05:05Z

go.mod

 go 1.24.0

 require (
+	github.com/alibabacloud-go/tea v1.3.10


I think we'd better have a new go.mod in the historyserver directory.

rueian · 2025-08-29T06:08:01Z

historyserver/backend/collector/runtime/logcollector/collector.go

+	subdir, filename := filepath.Split(relativePath)
+
+	if len(subdir) != 0 {
+		dirName := path.Join(h.RootDir, subdir)


Suggested change

dirName := path.Join(h.RootDir, subdir)

dirName := filepath.Join(h.RootDir, subdir)

rueian · 2025-08-29T06:08:40Z

historyserver/backend/collector/runtime/logcollector/collector.go

+		}
+	}
+
+	objectName := path.Join(h.RootDir, subdir, filename)


Suggested change

objectName := path.Join(h.RootDir, subdir, filename)

objectName := filepath.Join(h.RootDir, subdir, filename)

400Ping · 2025-08-29T06:28:20Z

historyserver/utils/utils.go

+				objects[i] = object.Key
+			}
+
+			// 删除对象


Can you use english comments? Same with L108 & L172

fscnick · 2025-08-29T13:47:14Z

historyserver/backend/collector/runtime/logcollector/collector.go

+	buf := bytes.NewBufferString("")
+	for {
+		select {
+		case <-time.Tick(h.PushInterval):


The time.Tick will create a ticker underneath. Even though it might not cause a leakage after 1.23. Did you mean ?

for { select { case <-time.After(h.PushInterval): ...

or

ticker := time.Tick(h.PushInterval) for { select { case <-ticker: ...

fscnick · 2025-08-29T13:49:15Z

historyserver/backend/collector/runtime/logcollector/collector.go

+
+				lastPush = now
+				lines = 0
+				buf = bytes.NewBufferString("")


Could it use buf.Reset() after the buffer has been successful written to writer?

fscnick · 2025-08-29T13:54:46Z

historyserver/backend/collector/runtime/logcollector/collector.go

+				nextPos, err = h.Writter.Append(objectName, buf, nextPos)
+				if err != nil {
+					logrus.Errorf("Tail file %s to object %s error, append value: %v, nextPos %d error [%v]",
+						absoluteLogPathName,
+						objectName,
+						buf.String(),
+						nextPos,
+						err)
+					return err
+				}
+
+				now := time.Now()
+				logrus.Infof("Tail file %s to object %s success, by line count, append value %s, lines %v, interval %v",
+					absoluteLogPathName, objectName, buf.String(), lines, now.Sub(lastPush).Seconds())
+
+				lastPush = now
+				lines = 0
+				buf = bytes.NewBufferString("")


Would it possible to put this part of code into a function? It looks similar to the code in the branch above if the lines is greater than zero.

fscnick · 2025-08-29T14:05:56Z

historyserver/backend/collector/runtime/logcollector/collector.go

+					}
+				}
+			}
+		case _, ok := <-watcher.Errors:


Why does it skip the error? Is it meaningless?

kenchung285 · 2025-08-29T15:45:50Z

historyserver/backend/collector/storage/aliyunoss/ray/ray_test.go

@@ -0,0 +1,62 @@
+// Package logs is
+/*
+Copyright 2024 by the zhangjie [email protected] Authors.


Suggested change

Copyright 2024 by the zhangjie bingyu.zj@alibaba-inc.com Authors.

Copyright 2025 by the zhangjie bingyu.zj@alibaba-inc.com Authors.

kenchung285 · 2025-08-29T15:49:29Z

historyserver/cmd/collector/main.go

+	registry := backend.GetRegistry()
+	factory, ok := registry[runtimeClassName]
+	if !ok {
+		panic("Not supported runtime class name: " + runtimeClassName + " for role: " + role + ".")


nit: Use format string instead

mlmhl · 2025-09-08T03:17:08Z

historyserver/backend/collector/storage/aliyunoss/ray/config.go

+func (c *config) complete(rcc *types.RayCollectorConfig, jd map[string]interface{}) {
+	c.RayCollectorConfig = *rcc
+	c.OSSBucket = jd["ossBucket"].(string)
+	c.OSSBucket = jd["ossEndpoint"].(string)


Should this be c. OSSEndpoint = jd["ossEndpoint"].(string)？

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

add interface

d2fb9c8

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

Future-Outlier reviewed Aug 22, 2025

View reviewed changes

Future-Outlier reviewed Aug 25, 2025

View reviewed changes

rueian reviewed Aug 29, 2025

View reviewed changes

400Ping reviewed Aug 29, 2025

View reviewed changes

historyserver/utils/utils.go

objects[i] = object.Key

}

// 删除对象

Copy link

Contributor

400Ping Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you use english comments? Same with L108 & L172

fscnick reviewed Aug 29, 2025

View reviewed changes

kenchung285 reviewed Aug 29, 2025

View reviewed changes

mlmhl reviewed Sep 8, 2025

View reviewed changes

KunWuLuan added 2 commits September 8, 2025 16:02

add implementation of collector

6932572

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

add implementation of historyserver

0fa13e6

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

KunWuLuan force-pushed the historyserver branch 3 times, most recently from 3e6f7e7 to dbd4bee Compare September 10, 2025 07:44

add dockerfile and makefile

57edeac

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

KunWuLuan force-pushed the historyserver branch from dbd4bee to 57edeac Compare September 10, 2025 09:16

		type StorageWritter interface {
		CreateDirectory(path string) error

	Copyright 2024 by the zhangjie bingyu.zj@alibaba-inc.com Authors.
	Copyright 2025 by the zhangjie bingyu.zj@alibaba-inc.com Authors.

	dirName := path.Join(h.RootDir, subdir)
	dirName := filepath.Join(h.RootDir, subdir)

	objectName := path.Join(h.RootDir, subdir, filename)
	objectName := filepath.Join(h.RootDir, subdir, filename)

[WIP] Add historyserver log collector #3973

Are you sure you want to change the base?

[WIP] Add historyserver log collector #3973

Conversation

KunWuLuan commented Aug 19, 2025

Why are these changes needed?

Related issue number

Checks

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!

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!

Uh oh!