[WIP] Add historyserver log collector #3973
+44,710
−1
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| package runtime | ||
|
|
||
| type RayLogCollector interface { | ||
| Start(<-chan int) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,251 @@ | ||||||
| package logcollector | ||||||
|
|
||||||
| import ( | ||||||
| "bytes" | ||||||
| "fmt" | ||||||
| "io" | ||||||
| "os" | ||||||
| "path" | ||||||
| "path/filepath" | ||||||
| "strings" | ||||||
| "time" | ||||||
|
|
||||||
| "github.com/fsnotify/fsnotify" | ||||||
| "github.com/hpcloud/tail" | ||||||
| "github.com/ray-project/kuberay/historyserver/backend/collector/storage" | ||||||
| "github.com/sirupsen/logrus" | ||||||
| ) | ||||||
|
|
||||||
| type RayLogHandler struct { | ||||||
| EnableMeta bool | ||||||
|
|
||||||
| RootDir string | ||||||
| LogDir string | ||||||
| LogFiles chan string | ||||||
|
|
||||||
| LogBatching int | ||||||
| PushInterval time.Duration | ||||||
|
|
||||||
| Writter storage.StorageWritter | ||||||
| } | ||||||
|
|
||||||
| func (r *RayLogHandler) Start(stop <-chan int) { | ||||||
|
|
||||||
| } | ||||||
|
|
||||||
| func (h *RayLogHandler) Run(stop chan struct{}) error { | ||||||
| watchPath := h.LogDir | ||||||
|
|
||||||
| watcher, err := fsnotify.NewWatcher() | ||||||
| if err != nil { | ||||||
| logrus.Fatalf("Create fsnotify NewWatcher error %v", err) | ||||||
| } | ||||||
| defer watcher.Close() | ||||||
| go h.WatchLogsLoops(watcher, watchPath, stop) | ||||||
| go h.PushLogsLoops(stop) | ||||||
|
|
||||||
| if h.EnableMeta { | ||||||
| // Persist meta data | ||||||
| // go h.PersistMetaLoop(stop) | ||||||
| } | ||||||
|
|
||||||
| select { | ||||||
| case <-stop: | ||||||
| logrus.Warnf("Receive stop single, so stop ray collector ") | ||||||
| return nil | ||||||
| } | ||||||
| logrus.Errorf("Run exist ...") | ||||||
| return nil | ||||||
| } | ||||||
|
|
||||||
| func (h *RayLogHandler) AddLogFile(absoluteLogPathName string) { | ||||||
| h.LogFiles <- absoluteLogPathName | ||||||
| } | ||||||
|
|
||||||
| func (h *RayLogHandler) PushLog(absoluteLogPathName string) error { | ||||||
| // 判断文件是否存在 | ||||||
| // 计算相对路径 | ||||||
| absoluteLogPathName = strings.TrimSpace(absoluteLogPathName) | ||||||
| absoluteLogPathName = filepath.Clean(absoluteLogPathName) | ||||||
|
|
||||||
| relativePath := strings.TrimPrefix(absoluteLogPathName, fmt.Sprintf("%s/", h.LogDir)) | ||||||
| // 分割相对路径为子目录和文件名 | ||||||
| // subdir events/aa/ | ||||||
| // filename a.txt | ||||||
| subdir, filename := filepath.Split(relativePath) | ||||||
|
|
||||||
| if len(subdir) != 0 { | ||||||
| dirName := path.Join(h.RootDir, subdir) | ||||||
|
||||||
| dirName := path.Join(h.RootDir, subdir) | |
| dirName := filepath.Join(h.RootDir, subdir) |
Outdated
There was a problem hiding this comment.
Suggested change
| objectName := path.Join(h.RootDir, subdir, filename) | |
| objectName := filepath.Join(h.RootDir, subdir, filename) |
Outdated
There was a problem hiding this comment.
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:
...There was a problem hiding this comment.
Could it use buf.Reset() after the buffer has been successful written to writer?
Outdated
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Why does it skip the error? Is it meaningless?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| package runtime | ||
|
|
||
| import ( | ||
| "github.com/ray-project/kuberay/historyserver/backend/collector/runtime/logcollector" | ||
| "github.com/ray-project/kuberay/historyserver/backend/collector/storage" | ||
| "github.com/ray-project/kuberay/historyserver/backend/types" | ||
| ) | ||
|
|
||
| func NewCollector(config *types.RayCollectorConfig, writter storage.StorageWritter) RayLogCollector { | ||
| handler := logcollector.RayLogHandler{ | ||
| LogFiles: make(chan string), | ||
|
|
||
| LogBatching: config.LogBatching, | ||
| PushInterval: config.PushInterval, | ||
|
|
||
| Writter: writter, | ||
| } | ||
|
|
||
| return &handler | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| // Package config is | ||
| /* | ||
| Copyright 2024 by the bingyu [email protected] Authors. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); | ||
| you may not use this file except in compliance with the License. | ||
| You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package config | ||
|
|
||
| import ( | ||
| "k8s.io/apimachinery/pkg/util/sets" | ||
| ) | ||
|
|
||
| type EnterConfig struct { | ||
| Module string | ||
| ModuleExecBinary string | ||
| EnterOssExpireHour int | ||
| WatchLogDir string | ||
| EnableMeta bool | ||
| } | ||
|
|
||
| type GlobalConfig struct { | ||
| OSSEndpoint string | ||
| OSSRegion string | ||
| OSSBucket string | ||
| OSSHistoryServerDir string | ||
| } | ||
|
|
||
| type RayMetaHanderConfig struct { | ||
| GlobalConfig | ||
| RayClusterName string | ||
| RayClusterID string | ||
| OSSExpireHour int | ||
| } | ||
|
|
||
| type RayHistoryServerConfig struct { | ||
| GlobalConfig | ||
| DashBoardDir string | ||
|
|
||
| ArmsRegionId string | ||
| ACKClusterId string | ||
| WebAppId string | ||
| WebAppSecret string | ||
| LocalServiceName string | ||
| AllowedUID sets.Set[string] | ||
| AllowedAID sets.Set[string] | ||
|
|
||
| MaxOssListLimit int | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we'd better have a new go.mod in the historyserver directory.