Skip to content

[History Server] Cover GetContent object paths for the s3 and azureblob backends #5075

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
stantheman0128 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[History Server] Cover GetContent object paths for the s3 and azureblob backends #5075

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 169 additions & 0 deletions historyserver/pkg/storage/azureblob/azureblob_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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("<Blob><Name>%s</Name><Properties></Properties></Blob>", name))
}
w.Header().Set("Content-Type", "application/xml")
_, _ = io.WriteString(w, fmt.Sprintf(`<?xml version="1.0" encoding="utf-8"?>
<EnumerationResults ContainerName="%s"><Prefix>%s</Prefix><Delimiter>/</Delimiter><Blobs>%s</Blobs><NextMarker /></EnumerationResults>`, 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 {
Expand Down
185 changes: 185 additions & 0 deletions historyserver/pkg/storage/s3/s3_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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, `<?xml version="1.0" encoding="UTF-8"?>
<Error><Code>NoSuchKey</Code><Message>The specified key does not exist.</Message></Error>`)
}

func writeEmptyListResult(w http.ResponseWriter, prefix string) {
w.Header().Set("Content-Type", "application/xml")
_, _ = io.WriteString(w, fmt.Sprintf(`<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult><Name>%s</Name><Prefix>%s</Prefix><KeyCount>0</KeyCount><MaxKeys>100</MaxKeys><IsTruncated>false</IsTruncated></ListBucketResult>`, testBucket, prefix))
}

func writeListResult(w http.ResponseWriter, prefix string, keys ...string) {
var contents strings.Builder
for _, key := range keys {
contents.WriteString(fmt.Sprintf("<Contents><Key>%s</Key><Size>1</Size></Contents>", key))
}
w.Header().Set("Content-Type", "application/xml")
_, _ = io.WriteString(w, fmt.Sprintf(`<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult><Name>%s</Name><Prefix>%s</Prefix><KeyCount>%d</KeyCount><MaxKeys>100</MaxKeys><IsTruncated>false</IsTruncated>%s</ListBucketResult>`, 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 {
Expand Down