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/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.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) + } +} 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 {