From da0e2f9c4bd3d62cbf0eca7960930fa79862a110 Mon Sep 17 00:00:00 2001 From: stantheman0128 Date: Tue, 4 Aug 2026 17:38:21 +0800 Subject: [PATCH 1/2] [History Server] Cover GetContent object paths for the s3 and azureblob backends Neither backend had any test touching GetContent. s3_test.go and azureblob_test.go held only TestTrim and TestWalk, which assert nothing and never construct a handler, so the object path that GetContent builds from rootDir, cluster prefix and file name was unverified on both. That path is easy to get wrong: the aliyunoss backend shipped without the root dir in it (#4820), and the same mistake in either of these would break every log fetch on a deployment that configures a root dir. Add two tests per backend, driving the real SDK against an httptest server so the assertion is on the path that reaches the wire rather than on a helper's return value. s3 uses path style addressing, which the MinIO support already relies on. The second test in each pair covers the list-and-retry fallback, whose listing prefix has to be rooted as well. No production code changes. --- .../pkg/storage/azureblob/azureblob_test.go | 169 ++++++++++++++++ historyserver/pkg/storage/s3/s3_test.go | 185 ++++++++++++++++++ 2 files changed, 354 insertions(+) diff --git a/historyserver/pkg/storage/azureblob/azureblob_test.go b/historyserver/pkg/storage/azureblob/azureblob_test.go index 8e7d4ec2939..2242c62bb34 100644 --- a/historyserver/pkg/storage/azureblob/azureblob_test.go +++ b/historyserver/pkg/storage/azureblob/azureblob_test.go @@ -2,12 +2,17 @@ package azureblob import ( "fmt" + "io" + "net/http" + "net/http/httptest" "os" "path" "path/filepath" "strings" + "sync" "testing" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" "github.com/sirupsen/logrus" "github.com/ray-project/kuberay/historyserver/pkg/utils" @@ -33,6 +38,170 @@ func TestTrim(t *testing.T) { t.Logf("test_path_join [%s]", test_path_join) } +// GetContent builds its blob path from three pieces, and a deployment that sets a +// root dir only works if all three end up in the path. These tests assert on the +// path that actually reaches the server rather than on a helper's return value. +const ( + testContainer = "test-container" + testRootDir = "ray-logs" + // Callers pass a root-dir-relative path prefix here, not a bare cluster id. + // See clusterlogs.Prefix("", ...) in pkg/historyserver/router.go. + testClusterPrefix = "ray_cluster_history/raycluster/default/my-cluster" + testFileName = "session_2026-05-08_18-35-06_774618_1/logs/node123/events/event_CORE_WORKER_256.log" +) + +// recorder collects what a test server was asked for. Requests are served on the +// server's own goroutines, so access is mutex-guarded to stay clean under -race. +type recorder struct { + mu sync.Mutex + values []string +} + +func (rec *recorder) add(value string) { + rec.mu.Lock() + defer rec.mu.Unlock() + rec.values = append(rec.values, value) +} + +func (rec *recorder) snapshot() []string { + rec.mu.Lock() + defer rec.mu.Unlock() + return append([]string(nil), rec.values...) +} + +func newTestHandler(t *testing.T, srv *httptest.Server) *RayLogsHandler { + t.Helper() + + client, err := container.NewClientWithNoCredential(srv.URL+"/"+testContainer, nil) + if err != nil { + t.Fatalf("creating test container client: %v", err) + } + + return &RayLogsHandler{ + ContainerClient: client, + ContainerName: testContainer, + RootDir: testRootDir, + } +} + +// blobPath returns the blob the request addressed, and whether the request was a +// container listing rather than a blob download. +func blobPath(r *http.Request) (name string, isList bool) { + name = strings.TrimPrefix(r.URL.Path, "/"+testContainer) + name = strings.TrimPrefix(name, "/") + return name, r.URL.Query().Get("comp") == "list" +} + +func TestGetContentUsesRootDir(t *testing.T) { + wantPath := path.Join(testRootDir, testClusterPrefix, testFileName) + const wantContent = "core worker log line" + + var requested recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + name, isList := blobPath(r) + if isList { + writeListResult(w, r.URL.Query().Get("prefix")) + return + } + requested.add(name) + if name != wantPath { + writeBlobNotFound(w) + return + } + _, _ = io.WriteString(w, wantContent) + })) + defer srv.Close() + + reader := newTestHandler(t, srv).GetContent(testClusterPrefix, testFileName) + gotPaths := requested.snapshot() + if reader == nil { + t.Fatalf("GetContent returned nil; blobs requested: %v, want %q", gotPaths, wantPath) + } + + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("reading returned content: %v", err) + } + if string(got) != wantContent { + t.Errorf("content = %q, want %q", got, wantContent) + } + if len(gotPaths) == 0 || gotPaths[0] != wantPath { + t.Errorf("first requested blob = %v, want %q", gotPaths, wantPath) + } +} + +// When the direct download fails, GetContent lists the containing directory and +// retries any blob whose full path matches. That listing prefix has to be rooted +// too, or the retry has nothing to find. The first download here fails with a +// server error so the fallback is the only way to reach the content. +func TestGetContentFallbackListsUnderRootDir(t *testing.T) { + wantPath := path.Join(testRootDir, testClusterPrefix, testFileName) + const wantContent = "recovered log line" + + var listed recorder + var downloads recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + name, isList := blobPath(r) + if isList { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if strings.HasPrefix(wantPath, prefix) { + writeListResult(w, prefix, wantPath) + return + } + writeListResult(w, prefix) + return + } + downloads.add(name) + // Miss the first attempt so the fallback has to do the work. BlobNotFound + // is used rather than a server error because the SDK retries the latter, + // which would satisfy the download before the fallback ever runs. + if name != wantPath || len(downloads.snapshot()) == 1 { + writeBlobNotFound(w) + return + } + _, _ = io.WriteString(w, wantContent) + })) + defer srv.Close() + + reader := newTestHandler(t, srv).GetContent(testClusterPrefix, testFileName) + listPrefixes := listed.snapshot() + if reader == nil { + t.Fatalf("GetContent returned nil; list prefixes tried: %v", listPrefixes) + } + + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("reading returned content: %v", err) + } + if string(got) != wantContent { + t.Errorf("content = %q, want %q", got, wantContent) + } + + wantPrefix := path.Dir(wantPath) + "/" + for _, prefix := range listPrefixes { + if prefix == wantPrefix { + return + } + } + t.Errorf("list prefixes = %v, want one equal to %q", listPrefixes, wantPrefix) +} + +func writeBlobNotFound(w http.ResponseWriter) { + w.Header().Set("x-ms-error-code", "BlobNotFound") + w.WriteHeader(http.StatusNotFound) +} + +func writeListResult(w http.ResponseWriter, prefix string, names ...string) { + var blobs strings.Builder + for _, name := range names { + blobs.WriteString(fmt.Sprintf("%s", name)) + } + w.Header().Set("Content-Type", "application/xml") + _, _ = io.WriteString(w, fmt.Sprintf(` +%s/%s`, testContainer, prefix, blobs.String())) +} + func TestWalk(t *testing.T) { watchPath := fmt.Sprintf("%s/test/LLogs/", utils.GetTmpRayRoot()) filepath.Walk(watchPath, func(path string, info os.FileInfo, err error) error { diff --git a/historyserver/pkg/storage/s3/s3_test.go b/historyserver/pkg/storage/s3/s3_test.go index 6da970e5463..f545321aaaa 100644 --- a/historyserver/pkg/storage/s3/s3_test.go +++ b/historyserver/pkg/storage/s3/s3_test.go @@ -18,12 +18,20 @@ package s3 import ( "fmt" + "io" + "net/http" + "net/http/httptest" "os" "path" "path/filepath" "strings" + "sync" "testing" + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/session" + awss3 "github.com/aws/aws-sdk-go/service/s3" "github.com/sirupsen/logrus" "github.com/ray-project/kuberay/historyserver/pkg/utils" @@ -50,6 +58,183 @@ func TestTrim(t *testing.T) { t.Logf("test_path_join [%s]", test_path_join) } +// GetContent builds its object key from three pieces, and a deployment that sets +// a root dir only works if all three end up in the key. These tests pin that down +// by asserting on the key that actually reaches the server, using path style +// addressing (the same mode the MinIO support already relies on) so the full key +// stays in the request path. +const ( + testBucket = "test-bucket" + testRootDir = "ray-logs" + // Callers pass a root-dir-relative path prefix here, not a bare cluster id. + // See clusterlogs.Prefix("", ...) in pkg/historyserver/router.go. + testClusterPrefix = "ray_cluster_history/raycluster/default/my-cluster" + testFileName = "session_2026-05-08_18-35-06_774618_1/logs/node123/events/event_CORE_WORKER_256.log" +) + +// recorder collects what a test server was asked for. Requests are served on the +// server's own goroutines, so access is mutex-guarded to stay clean under -race. +type recorder struct { + mu sync.Mutex + values []string +} + +func (rec *recorder) add(value string) { + rec.mu.Lock() + defer rec.mu.Unlock() + rec.values = append(rec.values, value) +} + +func (rec *recorder) snapshot() []string { + rec.mu.Lock() + defer rec.mu.Unlock() + return append([]string(nil), rec.values...) +} + +func newTestHandler(t *testing.T, srv *httptest.Server) *RayLogsHandler { + t.Helper() + + sess, err := session.NewSession(&aws.Config{ + Credentials: credentials.NewStaticCredentials("test-ak", "test-sk", ""), + Endpoint: aws.String(srv.URL), + Region: aws.String("us-east-1"), + DisableSSL: aws.Bool(true), + S3ForcePathStyle: aws.Bool(true), + MaxRetries: aws.Int(0), + }) + if err != nil { + t.Fatalf("creating test session: %v", err) + } + + return &RayLogsHandler{ + S3Client: awss3.New(sess), + S3Bucket: testBucket, + S3RootDir: testRootDir, + } +} + +// requestKey returns the object key a path style request addressed, and whether the +// request was a ListObjectsV2 call rather than a GetObject call. +func requestKey(r *http.Request) (key string, isList bool) { + key = strings.TrimPrefix(r.URL.Path, "/"+testBucket) + key = strings.TrimPrefix(key, "/") + return key, r.URL.Query().Get("list-type") == "2" +} + +func TestGetContentUsesRootDir(t *testing.T) { + wantKey := path.Join(testRootDir, testClusterPrefix, testFileName) + const wantContent = "core worker log line" + + var requested recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key, isList := requestKey(r) + if isList { + writeEmptyListResult(w, r.URL.Query().Get("prefix")) + return + } + requested.add(key) + if key != wantKey { + writeNoSuchKey(w) + return + } + _, _ = io.WriteString(w, wantContent) + })) + defer srv.Close() + + reader := newTestHandler(t, srv).GetContent(testClusterPrefix, testFileName) + gotKeys := requested.snapshot() + if reader == nil { + t.Fatalf("GetContent returned nil; keys requested: %v, want %q", gotKeys, wantKey) + } + + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("reading returned content: %v", err) + } + if string(got) != wantContent { + t.Errorf("content = %q, want %q", got, wantContent) + } + if len(gotKeys) == 0 || gotKeys[0] != wantKey { + t.Errorf("first requested key = %v, want %q", gotKeys, wantKey) + } +} + +// The recovery path lists the containing directory when the direct fetch misses, +// and that listing prefix has to be rooted too or it silently finds nothing. +func TestGetContentFallbackListsUnderRootDir(t *testing.T) { + wantKey := path.Join(testRootDir, testClusterPrefix, testFileName) + // The object sits one level deeper than asked for, so only the fallback can + // reach it. + nestedKey := path.Join(path.Dir(wantKey), "rotated", path.Base(wantKey)) + const wantContent = "recovered log line" + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key, isList := requestKey(r) + if isList { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if strings.HasPrefix(nestedKey, prefix) { + writeListResult(w, prefix, nestedKey) + return + } + writeEmptyListResult(w, prefix) + return + } + if key != nestedKey { + writeNoSuchKey(w) + return + } + _, _ = io.WriteString(w, wantContent) + })) + defer srv.Close() + + reader := newTestHandler(t, srv).GetContent(testClusterPrefix, testFileName) + listPrefixes := listed.snapshot() + if reader == nil { + t.Fatalf("GetContent returned nil; list prefixes tried: %v", listPrefixes) + } + + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("reading returned content: %v", err) + } + if string(got) != wantContent { + t.Errorf("content = %q, want %q", got, wantContent) + } + + wantPrefix := path.Dir(wantKey) + "/" + for _, prefix := range listPrefixes { + if prefix == wantPrefix { + return + } + } + t.Errorf("list prefixes = %v, want one equal to %q", listPrefixes, wantPrefix) +} + +func writeNoSuchKey(w http.ResponseWriter) { + w.Header().Set("Content-Type", "application/xml") + w.WriteHeader(http.StatusNotFound) + _, _ = io.WriteString(w, ` +NoSuchKeyThe specified key does not exist.`) +} + +func writeEmptyListResult(w http.ResponseWriter, prefix string) { + w.Header().Set("Content-Type", "application/xml") + _, _ = io.WriteString(w, fmt.Sprintf(` +%s%s0100false`, testBucket, prefix)) +} + +func writeListResult(w http.ResponseWriter, prefix string, keys ...string) { + var contents strings.Builder + for _, key := range keys { + contents.WriteString(fmt.Sprintf("%s1", key)) + } + w.Header().Set("Content-Type", "application/xml") + _, _ = io.WriteString(w, fmt.Sprintf(` +%s%s%d100false%s`, testBucket, prefix, len(keys), contents.String())) +} + func TestWalk(t *testing.T) { watchPath := fmt.Sprintf("%s/test/LLogs/", utils.GetTmpRayRoot()) filepath.Walk(watchPath, func(path string, info os.FileInfo, err error) error { From 34bc8d02ea1bbd2a3739a30a198e3bad6858a6fc Mon Sep 17 00:00:00 2001 From: stantheman0128 Date: Wed, 5 Aug 2026 23:47:52 +0800 Subject: [PATCH 2/2] [History Server] Keep directory placeholders out of ListFiles results ListFiles reports two kinds of entries: files, plain, and subdirectories, with a trailing slash. Callers depend on that distinction. ServerHandler .listFilesRecursive recurses into an entry only when it ends in a slash, and _getNodeLogs hands everything else to the log categorizer. The s3 and aliyunoss backends break the distinction. CreateDirectory writes a trailing-slash placeholder object for the directory itself, and a listing whose prefix is that directory returns that object as a key. path.Base then drops the slash, so the directory reports itself as a file one level down. Listing a node's events directory yields an entry named "events" next to the real event files, and it reaches the user through the log listing and the "internal" log category. The event readers happen to be immune, since isValidEventFile and the event_ prefix check reject the name; ipToNodeId only wastes a lookup on it. The same placeholder can also be picked up by the GetContent fallback, which matches on path.Base: a marker for a directory named x base-matches a request for a file named x, and the fallback then serves the empty marker instead of continuing the scan. The gcs backend already filters these out, with a comment saying so. Skip them in s3 and aliyunoss too, and in azureblob, which writes no placeholder itself but is just as exposed to one written by another tool. Cover the four methods that had no test on these three backends: List, ListFiles, CreateDirectory and WriteFile, which gcs_handler_test.go has covered for gcs all along. As with the GetContent tests, the SDK runs against an httptest server so the assertions land on the requests that reach the wire. aliyunoss had no test that constructed a handler at all. --- .../pkg/storage/aliyunoss/ray/ray.go | 9 + .../storage/aliyunoss/ray/ray_storage_test.go | 278 ++++++++++++++++++ .../pkg/storage/azureblob/azureblob.go | 15 + .../azureblob/azureblob_storage_test.go | 187 ++++++++++++ historyserver/pkg/storage/s3/s3.go | 9 + .../pkg/storage/s3/s3_storage_test.go | 240 +++++++++++++++ 6 files changed, 738 insertions(+) create mode 100644 historyserver/pkg/storage/aliyunoss/ray/ray_storage_test.go create mode 100644 historyserver/pkg/storage/azureblob/azureblob_storage_test.go create mode 100644 historyserver/pkg/storage/s3/s3_storage_test.go diff --git a/historyserver/pkg/storage/aliyunoss/ray/ray.go b/historyserver/pkg/storage/aliyunoss/ray/ray.go index 1e50f6a9f09..1cb570b5a18 100644 --- a/historyserver/pkg/storage/aliyunoss/ray/ray.go +++ b/historyserver/pkg/storage/aliyunoss/ray/ray.go @@ -105,6 +105,15 @@ func (r *RayLogsHandler) _listFiles(prefix string, delimiter string, onlyBase bo logrus.Infof("[ListFiles]Returned objects in %v. length of Contents: %v, length of CommonPrefixes: %v", prefix+"/", len(page.Contents), len(page.CommonPrefixes)) for _, objects := range page.Contents { + // CreateDirectory writes a trailing-slash placeholder object for the + // directory itself, and a listing whose prefix is that directory + // returns it as a key. It is not a file: path.Base would drop the + // slash and report it as one to the callers, which tell files from + // subdirectories by the trailing slash. Skip it, as the gcs backend + // already does. + if strings.HasSuffix(*objects.Key, "/") { + continue + } objName := *objects.Key if onlyBase { objName = path.Base(*objects.Key) diff --git a/historyserver/pkg/storage/aliyunoss/ray/ray_storage_test.go b/historyserver/pkg/storage/aliyunoss/ray/ray_storage_test.go new file mode 100644 index 00000000000..57d4016f053 --- /dev/null +++ b/historyserver/pkg/storage/aliyunoss/ray/ray_storage_test.go @@ -0,0 +1,278 @@ +package ray + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "path" + "strings" + "sync" + "testing" + "time" + + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss" + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials" + "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/retry" + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + + "github.com/ray-project/kuberay/historyserver/pkg/utils" +) + +// This backend had no test that constructed a handler at all. These cover the +// same four methods gcs_handler_test.go covers for the gcs backend: List, +// ListFiles, CreateDirectory and WriteFile. The assertions are on the requests +// that reach the server, so a change that stops rooting a key or drops a bucket +// operation shows up. + +const ( + testBucket = "test-bucket" + testRootDir = "ray-logs" + // Callers pass a root-dir-relative path prefix here, not a bare cluster id. + // See clusterlogs.Prefix("", ...) in pkg/historyserver/router.go. + testClusterPrefix = "ray_cluster_history/raycluster/default/my-cluster" + metadataPrefix = testRootDir + "/cluster-metadata/" +) + +// recorder collects what a test server was asked for. Requests are served on +// the server's own goroutines, so access is mutex-guarded to stay clean under +// -race. +type recorder struct { + mu sync.Mutex + values []string +} + +func (rec *recorder) add(value string) { + rec.mu.Lock() + defer rec.mu.Unlock() + rec.values = append(rec.values, value) +} + +func (rec *recorder) snapshot() []string { + rec.mu.Lock() + defer rec.mu.Unlock() + return append([]string(nil), rec.values...) +} + +func newTestHandler(_ *testing.T, srv *httptest.Server) *RayLogsHandler { + cfg := oss.LoadDefaultConfig(). + WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test-ak", "test-sk")). + WithRegion("cn-hangzhou"). + WithEndpoint(srv.URL). + // Path style keeps the bucket in the request path, so the test server + // sees the full key instead of a bucket-qualified host name. + WithUsePathStyle(true). + WithRetryer(retry.NopRetryer{}) + + return &RayLogsHandler{ + OssClient: oss.NewClient(cfg), + OssBucket: testBucket, + OssRootDir: testRootDir, + } +} + +// objectKey returns the key a path style request addressed. +func objectKey(r *http.Request) string { + key := strings.TrimPrefix(r.URL.Path, "/"+testBucket) + return strings.TrimPrefix(key, "/") +} + +func writeListResult(w http.ResponseWriter, prefix string, keys []string, commonPrefixes []string) { + var body strings.Builder + for _, key := range keys { + body.WriteString(fmt.Sprintf("%s1", key)) + } + for _, cp := range commonPrefixes { + body.WriteString(fmt.Sprintf("%s", cp)) + } + w.Header().Set("Content-Type", "application/xml") + _, _ = io.WriteString(w, fmt.Sprintf(` +%s%s%d100false%s`, + testBucket, prefix, len(keys)+len(commonPrefixes), body.String())) +} + +// ListFiles has to hand back files without a trailing slash and subdirectories +// with one: callers such as ServerHandler.listFilesRecursive tell the two apart +// that way. The directory's own placeholder object, which CreateDirectory +// writes and a listing rooted at that directory returns as a key, is neither. +func TestListFilesSeparatesFilesFromDirectories(t *testing.T) { + const dir = "session_2026-05-08_18-35-06_774618_1/logs/node123/events" + wantPrefix := path.Join(testRootDir, testClusterPrefix, dir) + "/" + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if prefix != wantPrefix { + writeListResult(w, prefix, nil, nil) + return + } + if r.URL.Query().Get("delimiter") != "/" { + // Without the delimiter the listing reports no CommonPrefixes and + // returns everything below the prefix instead, so a request that + // forgets it gets the nested file rather than the subdirectory it + // stands for. + writeListResult(w, prefix, + []string{prefix, prefix + "event_RAYLET.log", prefix + "event_GCS.log", prefix + "old/rotated.log"}, + nil) + return + } + writeListResult(w, prefix, + []string{prefix, prefix + "event_RAYLET.log", prefix + "event_GCS.log"}, + []string{prefix + "old/"}) + })) + defer srv.Close() + + got := newTestHandler(t, srv).ListFiles(testClusterPrefix, dir) + + // Order is up to the backend; the contract is which entries come back and + // whether each carries the trailing slash. + want := []string{"event_GCS.log", "event_RAYLET.log", "old/"} + if diff := cmp.Diff(want, got, cmpopts.SortSlices(func(a, b string) bool { return a < b })); diff != "" { + t.Errorf("ListFiles() diff (-want +got):\n%s\nprefixes listed: %v", diff, listed.snapshot()) + } +} + +// The keys are written out in full here rather than built with +// clustermetadata.EncodePath, so that a change in the layout has to be made in +// two places before this test agrees with it. +func TestListReadsClusterMetadataUnderRootDir(t *testing.T) { + newerSession := "session_2026-05-08_18-35-06_774618" + olderSession := "session_2026-05-07_09-00-00_000001" + newer := time.Date(2026, 5, 8, 18, 35, 6, 774618000, time.UTC) + older := time.Date(2026, 5, 7, 9, 0, 0, 1000, time.UTC) + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if prefix != metadataPrefix { + writeListResult(w, prefix, nil, nil) + return + } + writeListResult(w, prefix, []string{ + metadataPrefix + "raycluster/defaultns_mycluster1/" + olderSession, + metadataPrefix + "rayjob/defaultns_myrayjob_mycluster2/" + newerSession, + // A directory placeholder and a malformed entry: neither can be + // decoded, and neither may take the whole listing down with it. + metadataPrefix + "raycluster/", + metadataPrefix + "raycluster/not-a-cluster-dir", + }, nil) + })) + defer srv.Close() + + got := newTestHandler(t, srv).List() + + // ClusterInfoList sorts newest first, so the rayjob entry leads. + want := []utils.ClusterInfo{ + { + Name: "mycluster2", Namespace: "defaultns", + OwnerKind: "rayjob", OwnerName: "myrayjob", + SessionName: newerSession, + CreateTimeStamp: newer.Unix(), + CreateTime: "2026-05-08T18:35:06Z", + }, + { + Name: "mycluster1", Namespace: "defaultns", + SessionName: olderSession, + CreateTimeStamp: older.Unix(), + CreateTime: "2026-05-07T09:00:00Z", + }, + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("List() diff (-want +got):\n%s\nprefixes listed: %v", diff, listed.snapshot()) + } +} + +func TestCreateDirectoryWritesPlaceholderWhenMissing(t *testing.T) { + const dir = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs/node123/events" + wantKey := dir + "/" + + var puts recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key := objectKey(r) + switch r.Method { + case http.MethodHead: + w.WriteHeader(http.StatusNotFound) + case http.MethodPut: + body, _ := io.ReadAll(r.Body) + puts.add(fmt.Sprintf("%s|%d", key, len(body))) + default: + t.Errorf("unexpected %s request for %q", r.Method, key) + } + })) + defer srv.Close() + + if err := newTestHandler(t, srv).CreateDirectory(dir); err != nil { + t.Fatalf("CreateDirectory: %v", err) + } + + // The placeholder is what makes the directory visible to tools that list by + // prefix; it carries no content. + want := []string{wantKey + "|0"} + if diff := cmp.Diff(want, puts.snapshot()); diff != "" { + t.Errorf("objects written diff (-want +got):\n%s", diff) + } +} + +func TestCreateDirectoryLeavesExistingDirectoryAlone(t *testing.T) { + const dir = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs" + + var puts recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key := objectKey(r) + switch r.Method { + case http.MethodHead: + if key != dir+"/" { + t.Errorf("existence check key = %q, want %q", key, dir+"/") + } + w.Header().Set("Content-Length", "0") + w.WriteHeader(http.StatusOK) + case http.MethodPut: + puts.add(key) + default: + t.Errorf("unexpected %s request for %q", r.Method, key) + } + })) + defer srv.Close() + + if err := newTestHandler(t, srv).CreateDirectory(dir); err != nil { + t.Fatalf("CreateDirectory: %v", err) + } + + if written := puts.snapshot(); len(written) != 0 { + t.Errorf("CreateDirectory rewrote an existing directory: %v", written) + } +} + +func TestWriteFileUploadsBodyToGivenKey(t *testing.T) { + const key = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs/node123/raylet.out" + const content = "raylet line one\nraylet line two\n" + + var uploads recorder + srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + gotKey := objectKey(r) + if r.Method != http.MethodPut { + t.Errorf("unexpected %s request for %q", r.Method, gotKey) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("reading uploaded body: %v", err) + return + } + uploads.add(gotKey + "|" + string(body)) + })) + defer srv.Close() + + if err := newTestHandler(t, srv).WriteFile(key, bytes.NewReader([]byte(content))); err != nil { + t.Fatalf("WriteFile: %v", err) + } + + want := []string{key + "|" + content} + if diff := cmp.Diff(want, uploads.snapshot()); diff != "" { + t.Errorf("uploads diff (-want +got):\n%s", diff) + } +} diff --git a/historyserver/pkg/storage/azureblob/azureblob.go b/historyserver/pkg/storage/azureblob/azureblob.go index 1a6264fde3d..2c9e620d7a6 100644 --- a/historyserver/pkg/storage/azureblob/azureblob.go +++ b/historyserver/pkg/storage/azureblob/azureblob.go @@ -70,6 +70,15 @@ func (r *RayLogsHandler) WriteFile(file string, reader io.ReadSeeker) error { return nil } +// isDirectoryPlaceholder reports whether a blob name denotes a directory rather +// than a file. Other tools (and the s3 and aliyunoss collectors) write these +// trailing-slash marker blobs; a listing whose prefix is that directory returns +// the marker as an item. It must not reach the callers as a file, since they +// tell files from subdirectories by the trailing slash and path.Base drops it. +func isDirectoryPlaceholder(name string) bool { + return strings.HasSuffix(name, "/") +} + func (r *RayLogsHandler) listBlobs(prefix string, delimiter string, onlyBase bool) []string { ctx, cancel := context.WithTimeout(context.Background(), listTimeout) defer cancel() @@ -95,6 +104,9 @@ func (r *RayLogsHandler) listBlobs(prefix string, delimiter string, onlyBase boo prefixWithSlash, len(resp.Segment.BlobItems), len(resp.Segment.BlobPrefixes)) for _, blob := range resp.Segment.BlobItems { + if isDirectoryPlaceholder(*blob.Name) { + continue + } objName := *blob.Name if onlyBase { objName = path.Base(*blob.Name) @@ -128,6 +140,9 @@ func (r *RayLogsHandler) listBlobs(prefix string, delimiter string, onlyBase boo prefixWithSlash, len(resp.Segment.BlobItems)) for _, blob := range resp.Segment.BlobItems { + if isDirectoryPlaceholder(*blob.Name) { + continue + } objName := *blob.Name if onlyBase { objName = path.Base(*blob.Name) diff --git a/historyserver/pkg/storage/azureblob/azureblob_storage_test.go b/historyserver/pkg/storage/azureblob/azureblob_storage_test.go new file mode 100644 index 00000000000..ee38baace3c --- /dev/null +++ b/historyserver/pkg/storage/azureblob/azureblob_storage_test.go @@ -0,0 +1,187 @@ +package azureblob + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "path" + "strings" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + + "github.com/ray-project/kuberay/historyserver/pkg/utils" +) + +// The four methods below round out what gcs_handler_test.go already covers for +// the gcs backend: List, ListFiles, CreateDirectory and WriteFile. As in the +// GetContent tests, the assertions are on the requests that reach the server. + +const metadataPrefix = testRootDir + "/cluster-metadata/" + +// writeHierarchicalListResult answers a delimiter listing: blobs go in Blobs, +// "subdirectories" in BlobPrefix entries. +func writeHierarchicalListResult(w http.ResponseWriter, prefix string, names []string, blobPrefixes []string) { + var body strings.Builder + for _, name := range names { + body.WriteString(fmt.Sprintf("%s", name)) + } + for _, bp := range blobPrefixes { + body.WriteString(fmt.Sprintf("%s", bp)) + } + w.Header().Set("Content-Type", "application/xml") + _, _ = io.WriteString(w, fmt.Sprintf(` +%s/%s`, + testContainer, prefix, body.String())) +} + +// ListFiles has to hand back files without a trailing slash and subdirectories +// with one: callers such as ServerHandler.listFilesRecursive tell the two apart +// that way. Directory marker blobs are neither. This backend does not write +// them itself, but they are a common Azure convention and other writers leave +// them behind, so a listing may still return one. +func TestListFilesSeparatesFilesFromDirectories(t *testing.T) { + const dir = "session_2026-05-08_18-35-06_774618_1/logs/node123/events" + wantPrefix := path.Join(testRootDir, testClusterPrefix, dir) + "/" + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if prefix != wantPrefix { + writeListResult(w, prefix) + return + } + if r.URL.Query().Get("delimiter") != "/" { + // A flat listing reports no BlobPrefix entries and returns + // everything below the prefix instead, so a request that forgets + // the delimiter gets the nested blob rather than the subdirectory + // it stands for. + writeHierarchicalListResult(w, prefix, + []string{prefix, prefix + "event_RAYLET.log", prefix + "event_GCS.log", prefix + "old/rotated.log"}, + nil) + return + } + writeHierarchicalListResult(w, prefix, + []string{prefix, prefix + "event_RAYLET.log", prefix + "event_GCS.log"}, + []string{prefix + "old/"}) + })) + defer srv.Close() + + got := newTestHandler(t, srv).ListFiles(testClusterPrefix, dir) + + // Order is up to the backend; the contract is which entries come back and + // whether each carries the trailing slash. + want := []string{"event_GCS.log", "event_RAYLET.log", "old/"} + if diff := cmp.Diff(want, got, cmpopts.SortSlices(func(a, b string) bool { return a < b })); diff != "" { + t.Errorf("ListFiles() diff (-want +got):\n%s\nprefixes listed: %v", diff, listed.snapshot()) + } +} + +// The blob names are written out in full here rather than built with +// clustermetadata.EncodePath, so that a change in the layout has to be made in +// two places before this test agrees with it. +func TestListReadsClusterMetadataUnderRootDir(t *testing.T) { + newerSession := "session_2026-05-08_18-35-06_774618" + olderSession := "session_2026-05-07_09-00-00_000001" + newer := time.Date(2026, 5, 8, 18, 35, 6, 774618000, time.UTC) + older := time.Date(2026, 5, 7, 9, 0, 0, 1000, time.UTC) + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if prefix != metadataPrefix { + writeListResult(w, prefix) + return + } + writeListResult(w, prefix, + metadataPrefix+"raycluster/defaultns_mycluster1/"+olderSession, + metadataPrefix+"rayjob/defaultns_myrayjob_mycluster2/"+newerSession, + // A directory marker and a malformed entry: neither can be decoded, + // and neither may take the whole listing down with it. + metadataPrefix+"raycluster/", + metadataPrefix+"raycluster/not-a-cluster-dir", + ) + })) + defer srv.Close() + + got := newTestHandler(t, srv).List() + + // ClusterInfoList sorts newest first, so the rayjob entry leads. + want := []utils.ClusterInfo{ + { + Name: "mycluster2", Namespace: "defaultns", + OwnerKind: "rayjob", OwnerName: "myrayjob", + SessionName: newerSession, + CreateTimeStamp: newer.Unix(), + CreateTime: "2026-05-08T18:35:06Z", + }, + { + Name: "mycluster1", Namespace: "defaultns", + SessionName: olderSession, + CreateTimeStamp: older.Unix(), + CreateTime: "2026-05-07T09:00:00Z", + }, + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("List() diff (-want +got):\n%s\nprefixes listed: %v", diff, listed.snapshot()) + } +} + +// Unlike s3 and aliyunoss, this backend deliberately writes no marker blob: +// virtual directories are inferred from blob paths, and a marker shows up as +// "" in Azure Storage Explorer. Pin that down so the empty body is +// read as a decision rather than as something left unfinished. +func TestCreateDirectoryWritesNothing(t *testing.T) { + var requests recorder + srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + name, _ := blobPath(r) + requests.add(r.Method + " " + name) + })) + defer srv.Close() + + handler := newTestHandler(t, srv) + if err := handler.CreateDirectory(path.Join(testRootDir, testClusterPrefix, "session_1/logs/node123/events")); err != nil { + t.Fatalf("CreateDirectory: %v", err) + } + + if sent := requests.snapshot(); len(sent) != 0 { + t.Errorf("CreateDirectory talked to the container: %v", sent) + } +} + +func TestWriteFileUploadsBodyToGivenBlob(t *testing.T) { + const name = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs/node123/raylet.out" + const content = "raylet line one\nraylet line two\n" + + var uploads recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotName, _ := blobPath(r) + if r.Method != http.MethodPut { + t.Errorf("unexpected %s request for %q", r.Method, gotName) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("reading uploaded body: %v", err) + return + } + uploads.add(gotName + "|" + string(body)) + w.WriteHeader(http.StatusCreated) + })) + defer srv.Close() + + if err := newTestHandler(t, srv).WriteFile(name, bytes.NewReader([]byte(content))); err != nil { + t.Fatalf("WriteFile: %v", err) + } + + want := []string{name + "|" + content} + if diff := cmp.Diff(want, uploads.snapshot()); diff != "" { + t.Errorf("uploads diff (-want +got):\n%s", diff) + } +} diff --git a/historyserver/pkg/storage/s3/s3.go b/historyserver/pkg/storage/s3/s3.go index b7e99126066..7f175a094c7 100644 --- a/historyserver/pkg/storage/s3/s3.go +++ b/historyserver/pkg/storage/s3/s3.go @@ -104,6 +104,15 @@ func (r *RayLogsHandler) _listFiles(prefix string, delimiter string, onlyBase bo prefix+"/", len(page.Contents), len(page.CommonPrefixes)) for _, object := range page.Contents { + // CreateDirectory writes a trailing-slash placeholder object for the + // directory itself, and a listing whose prefix is that directory + // returns it as a key. It is not a file: path.Base would drop the + // slash and report it as one to the callers, which tell files from + // subdirectories by the trailing slash. Skip it, as the gcs backend + // already does. + if strings.HasSuffix(*object.Key, "/") { + continue + } objName := *object.Key if onlyBase { objName = path.Base(*object.Key) diff --git a/historyserver/pkg/storage/s3/s3_storage_test.go b/historyserver/pkg/storage/s3/s3_storage_test.go new file mode 100644 index 00000000000..e71e0cdc007 --- /dev/null +++ b/historyserver/pkg/storage/s3/s3_storage_test.go @@ -0,0 +1,240 @@ +// Package s3 is +/* +Copyright 2024 by the kuberay 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 s3 + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "path" + "strings" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + + "github.com/ray-project/kuberay/historyserver/pkg/utils" +) + +// The four methods below round out what gcs_handler_test.go already covers for +// the gcs backend: List, ListFiles, CreateDirectory and WriteFile. As in the +// GetContent tests, the assertions are on the requests that reach the server, +// so a change that stops rooting a key or drops a bucket operation shows up. + +const metadataPrefix = testRootDir + "/cluster-metadata/" + +// writeHierarchicalListResult answers a delimiter listing: object keys go in +// Contents, "subdirectories" in CommonPrefixes. +func writeHierarchicalListResult(w http.ResponseWriter, prefix string, keys []string, commonPrefixes []string) { + var body strings.Builder + for _, key := range keys { + body.WriteString(fmt.Sprintf("%s1", key)) + } + for _, cp := range commonPrefixes { + body.WriteString(fmt.Sprintf("%s", cp)) + } + w.Header().Set("Content-Type", "application/xml") + _, _ = io.WriteString(w, fmt.Sprintf(` +%s%s/%d100false%s`, + testBucket, prefix, len(keys)+len(commonPrefixes), body.String())) +} + +// ListFiles has to hand back files without a trailing slash and subdirectories +// with one: callers such as ServerHandler.listFilesRecursive tell the two apart +// that way. The directory's own placeholder object, which CreateDirectory +// writes and a listing rooted at that directory returns as a key, is neither. +func TestListFilesSeparatesFilesFromDirectories(t *testing.T) { + const dir = "session_2026-05-08_18-35-06_774618_1/logs/node123/events" + wantPrefix := path.Join(testRootDir, testClusterPrefix, dir) + "/" + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if prefix != wantPrefix { + writeEmptyListResult(w, prefix) + return + } + if r.URL.Query().Get("delimiter") != "/" { + // Without the delimiter S3 reports no CommonPrefixes and returns + // everything below the prefix instead, so a request that forgets it + // gets the nested file rather than the subdirectory it stands for. + writeHierarchicalListResult(w, prefix, + []string{prefix, prefix + "event_RAYLET.log", prefix + "event_GCS.log", prefix + "old/rotated.log"}, + nil) + return + } + writeHierarchicalListResult(w, prefix, + []string{prefix, prefix + "event_RAYLET.log", prefix + "event_GCS.log"}, + []string{prefix + "old/"}) + })) + defer srv.Close() + + got := newTestHandler(t, srv).ListFiles(testClusterPrefix, dir) + + // Order is up to the backend; the contract is which entries come back and + // whether each carries the trailing slash. + want := []string{"event_GCS.log", "event_RAYLET.log", "old/"} + if diff := cmp.Diff(want, got, cmpopts.SortSlices(func(a, b string) bool { return a < b })); diff != "" { + t.Errorf("ListFiles() diff (-want +got):\n%s\nprefixes listed: %v", diff, listed.snapshot()) + } +} + +// The keys are written out in full here rather than built with +// clustermetadata.EncodePath, so that a change in the layout has to be made in +// two places before these tests agree with it. +func TestListReadsClusterMetadataUnderRootDir(t *testing.T) { + newerSession := "session_2026-05-08_18-35-06_774618" + olderSession := "session_2026-05-07_09-00-00_000001" + newer := time.Date(2026, 5, 8, 18, 35, 6, 774618000, time.UTC) + older := time.Date(2026, 5, 7, 9, 0, 0, 1000, time.UTC) + + var listed recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + prefix := r.URL.Query().Get("prefix") + listed.add(prefix) + if prefix != metadataPrefix { + writeEmptyListResult(w, prefix) + return + } + writeListResult(w, prefix, + metadataPrefix+"raycluster/defaultns_mycluster1/"+olderSession, + metadataPrefix+"rayjob/defaultns_myrayjob_mycluster2/"+newerSession, + // A directory placeholder and a malformed entry: neither can be + // decoded, and neither may take the whole listing down with it. + metadataPrefix+"raycluster/", + metadataPrefix+"raycluster/not-a-cluster-dir", + ) + })) + defer srv.Close() + + got := newTestHandler(t, srv).List() + + // ClusterInfoList sorts newest first, so the rayjob entry leads. + want := []utils.ClusterInfo{ + { + Name: "mycluster2", Namespace: "defaultns", + OwnerKind: "rayjob", OwnerName: "myrayjob", + SessionName: newerSession, + CreateTimeStamp: newer.Unix(), + CreateTime: "2026-05-08T18:35:06Z", + }, + { + Name: "mycluster1", Namespace: "defaultns", + SessionName: olderSession, + CreateTimeStamp: older.Unix(), + CreateTime: "2026-05-07T09:00:00Z", + }, + } + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("List() diff (-want +got):\n%s\nprefixes listed: %v", diff, listed.snapshot()) + } +} + +func TestCreateDirectoryWritesPlaceholderWhenMissing(t *testing.T) { + const dir = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs/node123/events" + wantKey := dir + "/" + + var puts recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key, _ := requestKey(r) + switch r.Method { + case http.MethodHead: + w.WriteHeader(http.StatusNotFound) + case http.MethodPut: + body, _ := io.ReadAll(r.Body) + puts.add(fmt.Sprintf("%s|%d", key, len(body))) + default: + t.Errorf("unexpected %s request for %q", r.Method, key) + } + })) + defer srv.Close() + + if err := newTestHandler(t, srv).CreateDirectory(dir); err != nil { + t.Fatalf("CreateDirectory: %v", err) + } + + // The placeholder is what makes the directory visible to tools that list by + // prefix; it carries no content. + want := []string{wantKey + "|0"} + if diff := cmp.Diff(want, puts.snapshot()); diff != "" { + t.Errorf("objects written diff (-want +got):\n%s", diff) + } +} + +func TestCreateDirectoryLeavesExistingDirectoryAlone(t *testing.T) { + const dir = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs" + + var puts recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key, _ := requestKey(r) + switch r.Method { + case http.MethodHead: + if key != dir+"/" { + t.Errorf("HeadObject key = %q, want %q", key, dir+"/") + } + w.Header().Set("Content-Length", "0") + w.WriteHeader(http.StatusOK) + case http.MethodPut: + puts.add(key) + default: + t.Errorf("unexpected %s request for %q", r.Method, key) + } + })) + defer srv.Close() + + if err := newTestHandler(t, srv).CreateDirectory(dir); err != nil { + t.Fatalf("CreateDirectory: %v", err) + } + + if written := puts.snapshot(); len(written) != 0 { + t.Errorf("CreateDirectory rewrote an existing directory: %v", written) + } +} + +func TestWriteFileUploadsBodyToGivenKey(t *testing.T) { + const key = "ray-logs/ray_cluster_history/raycluster/default/my-cluster/session_1/logs/node123/raylet.out" + const content = "raylet line one\nraylet line two\n" + + var uploads recorder + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotKey, _ := requestKey(r) + if r.Method != http.MethodPut { + t.Errorf("unexpected %s request for %q", r.Method, gotKey) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("reading uploaded body: %v", err) + return + } + uploads.add(gotKey + "|" + string(body)) + })) + defer srv.Close() + + if err := newTestHandler(t, srv).WriteFile(key, bytes.NewReader([]byte(content))); err != nil { + t.Fatalf("WriteFile: %v", err) + } + + want := []string{key + "|" + content} + if diff := cmp.Diff(want, uploads.snapshot()); diff != "" { + t.Errorf("uploads diff (-want +got):\n%s", diff) + } +}