[History Server] Refactor get dir helper function

historyserver/pkg/collector/eventcollector/eventcollector.go

@@ -837,10 +837,17 @@ func (ec *EventCollector) buildEventStorageKey(task rotationTask) string {
 	fileName := fmt.Sprintf("%s-%s-%d%s", nodeID, hour, task.createdAt.UnixNano(), ext)
 
 	var dir string
+	var clusterInfo  = &utils.ClusterInfo{
+		OwnerKind: ec.ownerKind,
+		OwnerName: ec.ownerName,
+		Namespace: ec.clusterNamespace,
+		Name:      ec.clusterName,
+		SessionName: sessionName,
+	}
 	if jobID, ok := strings.CutPrefix(task.category, categoryJobPrefix+"/"); ok {
-		dir = clusterlogs.JobEventsDir(ec.root, ec.ownerKind, ec.ownerName, ec.clusterNamespace, ec.clusterName, sessionName, nodeID, jobID)
+		dir = clusterlogs.JobEventsDir(ec.root, clusterInfo, nodeID, jobID)
 	} else {
-		dir = clusterlogs.NodeEventsDir(ec.root, ec.ownerKind, ec.ownerName, ec.clusterNamespace, ec.clusterName, sessionName, nodeID)
+		dir = clusterlogs.NodeEventsDir(ec.root, clusterInfo, nodeID)
 	}
 	return path.Join(dir, fileName)
 }

historyserver/pkg/collector/logcollector/runtime/logcollector/collector.go

@@ -212,7 +212,14 @@ func (r *RayLogHandler) processSessionLatestLogFile(absoluteLogPathName, session
 	subdir, _ := filepath.Split(relativePath)
 
 	// Build the object name using the standard path structure
-	logDir := clusterlogs.LogsDir(r.RootDir, r.OwnerKind, r.OwnerName, r.RayClusterNamespace, r.RayClusterName, sessionID, nodeID)
+	var clusterInfo = &utils.ClusterInfo{
+		Name:         r.RayClusterName,
+		OwnerKind:    r.OwnerKind,
+		OwnerName:    r.OwnerName,
+		Namespace:    r.RayClusterNamespace,
+		SessionName: sessionID,
+	}
+	logDir := clusterlogs.LogsDir(r.RootDir, clusterInfo, nodeID)
 
 	if subdir != "" && subdir != "." {
 		// Remove trailing separator if present
@@ -649,7 +656,14 @@ func (r *RayLogHandler) processPrevLogFile(absoluteLogPathName, localLogDir, ses
 	subdir, _ := filepath.Split(relativePath)
 
 	// Build the object name using the standard path structure
-	logDir := clusterlogs.LogsDir(r.RootDir, r.OwnerKind, r.OwnerName, r.RayClusterNamespace, r.RayClusterName, sessionID, nodeID)
+	var clusterInfo = &utils.ClusterInfo{
+		Name:         r.RayClusterName,
+		OwnerKind:    r.OwnerKind,
+		OwnerName:    r.OwnerName,
+		Namespace:    r.RayClusterNamespace,
+		SessionName: sessionID,
+	}
+	logDir := clusterlogs.LogsDir(r.RootDir, clusterInfo, nodeID)
 
 	if subdir != "" && subdir != "." {
 		// Remove trailing separator if present

historyserver/pkg/collector/logcollector/runtime/runtime.go

@@ -52,7 +52,14 @@ func NewCollector(config *types.RayCollectorConfig, writer storage.StorageWriter
 
 	logDir := strings.TrimSpace(filepath.Join(config.SessionDir, utils.RAY_SESSIONDIR_LOGDIR_NAME))
 	handler.LogDir = logDir
-	clusterRootDir := clusterlogs.Prefix(handler.RootDir, handler.OwnerKind, handler.OwnerName, handler.RayClusterNamespace, handler.RayClusterName)
+	clusterInfo := &utils.ClusterInfo{
+		Name:         config.RayClusterName,
+		OwnerKind:    config.OwnerKind,
+		OwnerName:    config.OwnerName,
+		Namespace:    config.RayClusterNamespace,
+		SessionName:  config.SessionDir,
+	}
+	clusterRootDir := clusterlogs.Prefix(handler.RootDir, clusterInfo)
 	handler.ClusterDir = clusterRootDir
 
 	return &handler

historyserver/pkg/eventserver/eventserver.go

@@ -522,7 +522,7 @@ func (h *EventHandler) storeEvent(clusterSessionKey string, eventMap map[string]
 // getClusterLogPathPrefix returns the cluster storage prefix:
 // e.g., cluster-history/{ownerKind}/{namespace}/{ownerName}/{clusterName}
 func (h *EventHandler) getClusterLogPathPrefix(clusterInfo utils.ClusterInfo) string {
-	return clusterlogs.Prefix("", clusterInfo.OwnerKind, clusterInfo.OwnerName, clusterInfo.Namespace, clusterInfo.Name)
+	return clusterlogs.Prefix("", &clusterInfo)
 }
 
 // getAllJobEventFiles get all the job event files for the given cluster.

historyserver/pkg/eventserver/log_event_reader.go

@@ -54,7 +54,7 @@ func NewLogEventReader(reader storage.StorageReader) *LogEventReader {
 // Return an error if any listed file fails to read (total or partial).
 func (r *LogEventReader) ReadLogEvents(clusterInfo utils.ClusterInfo, clusterSessionKey string, eventStore *types.ClusterLogEventMap) error {
 	// Build cluster ID (clusterLogPathPrefix) used by StorageReader
-	clusterLogPathPrefix := clusterlogs.Prefix("", clusterInfo.OwnerKind, clusterInfo.OwnerName, clusterInfo.Namespace, clusterInfo.Name)
+	clusterLogPathPrefix := clusterlogs.Prefix("", &clusterInfo)
 
 	// Get or create the JobEventMap for this cluster session
 	jobEventMap := eventStore.GetOrCreateJobEventMap(clusterSessionKey)

historyserver/pkg/historyserver/router.go

@@ -1266,7 +1266,14 @@ func (s *ServerHandler) getClusterLogPathPrefix(req *restful.Request) string {
 	clusterNamespace, _ := req.Attribute(COOKIE_CLUSTER_NAMESPACE_KEY).(string)
 	ownerKind, _ := req.Attribute(COOKIE_OWNER_KIND_KEY).(string)
 	ownerName, _ := req.Attribute(COOKIE_OWNER_NAME_KEY).(string)
-	return clusterlogs.Prefix("", ownerKind, ownerName, clusterNamespace, clusterName)
+	clusterInfo := &utils.ClusterInfo{
+		OwnerKind:      ownerKind,
+		OwnerName:      ownerName,
+		Name:    clusterName,
+		Namespace:      clusterNamespace,
+		SessionName: req.Attribute(COOKIE_SESSION_NAME_KEY).(string),
+	}
+	return clusterlogs.Prefix("", clusterInfo)
 }
 
 func (s *ServerHandler) getNodeLogs(req *restful.Request, resp *restful.Response) {

historyserver/pkg/storage/clusterlogs/clusterlogs.go

@@ -18,29 +18,29 @@ const (
 // - raycluster: rootDir/cluster-history/raycluster/<namespace>/<cluster-name>
 // - rayjob:     rootDir/cluster-history/rayjob/<namespace>/<rayjob-name>/<cluster-name>
 // - rayservice: rootDir/cluster-history/rayservice/<namespace>/<rayservice-name>/<cluster-name>
-func Prefix(rootDir, ownerKind, ownerName, namespace, clusterName string) string {
-	k := strings.ToLower(strings.TrimSpace(ownerKind))
-	hasOwner := (k == utils.RayJobKind || k == utils.RayServiceKind) && strings.TrimSpace(ownerName) != ""
+func Prefix(rootDir string, c *utils.ClusterInfo) string {
+	k := strings.ToLower(strings.TrimSpace(c.OwnerKind))
+	hasOwner := (k == utils.RayJobKind || k == utils.RayServiceKind) && strings.TrimSpace(c.OwnerName) != ""
 
 	subDir := utils.RayClusterKind
 	if hasOwner {
 		subDir = k
 	}
 
-	parts := []string{rootDir, ClusterHistoryDir, subDir, namespace}
+	parts := []string{rootDir, ClusterHistoryDir, subDir, c.Namespace}
 	if hasOwner {
-		parts = append(parts, strings.TrimSpace(ownerName))
+		parts = append(parts, strings.TrimSpace(c.OwnerName))
 	}
-	parts = append(parts, clusterName)
+	parts = append(parts, c.Name)
 
 	return path.Join(parts...)
 }
 
 // SessionDir returns the path to a session's directory under a cluster:
 // <prefix>/<session-name>
-func SessionDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName string) string {
-	cp := Prefix(rootDir, ownerKind, ownerName, namespace, clusterName)
-	return path.Join(cp, sessionName)
+func SessionDir(rootDir string, c *utils.ClusterInfo) string {
+	cp := Prefix(rootDir, c)
+	return path.Join(cp, c.SessionName)
 }
 
 // FetchedEndpointsDir returns the directory containing dashboard endpoint snapshots:
@@ -51,29 +51,29 @@ func FetchedEndpointsDir(prefix, sessionName string) string {
 
 // NodeDir returns the path to a node's directory under a session:
 // <prefix>/<session-name>/<node-name>
-func NodeDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName string) string {
-	sDir := SessionDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName)
+func NodeDir(rootDir string, c *utils.ClusterInfo, nodeName string) string {
+	sDir := SessionDir(rootDir, c)
 	return path.Join(sDir, nodeName)
 }
 
 // LogsDir returns the log directory for a specific node and session:
 // <prefix>/<session-name>/<node-name>/logs
-func LogsDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName string) string {
-	nDir := NodeDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName)
+func LogsDir(rootDir string, c *utils.ClusterInfo, nodeName string) string {
+	nDir := NodeDir(rootDir, c, nodeName)
 	return path.Join(nDir, LogsSubDir)
 }
 
 // NodeEventsDir returns the node_events directory for a specific node and session:
 // <prefix>/<session-name>/<node-name>/node_events
-func NodeEventsDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName string) string {
-	nDir := NodeDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName)
+func NodeEventsDir(rootDir string, c *utils.ClusterInfo, nodeName string) string {
+	nDir := NodeDir(rootDir, c, nodeName)
 	return path.Join(nDir, NodeEventsSubDir)
 }
 
 // JobEventsDir returns the job_events directory for a specific node and session (and optional jobID):
 // <prefix>/<session-name>/<node-name>/job_events/[jobID]
-func JobEventsDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName, jobID string) string {
-	nDir := NodeDir(rootDir, ownerKind, ownerName, namespace, clusterName, sessionName, nodeName)
+func JobEventsDir(rootDir string, c *utils.ClusterInfo, nodeName, jobID string) string {
+	nDir := NodeDir(rootDir, c, nodeName)
 	if jobID == "" {
 		return path.Join(nDir, JobEventsSubDir)
 	}

historyserver/pkg/storage/clusterlogs/clusterlogs_test.go

@@ -2,6 +2,8 @@ package clusterlogs
 
 import (
 	"testing"
+
+	"github.com/ray-project/kuberay/historyserver/pkg/utils"
 )
 
 func TestClusterLogsPaths(t *testing.T) {
@@ -15,12 +17,23 @@ func TestClusterLogsPaths(t *testing.T) {
 	jobID := "01000000"
 
 	wantPrefix := "cluster-history/rayjob/default/job-1/cluster-1"
-	if got := Prefix(rootDir, ownerKind, ownerName, ns, cluster); got != wantPrefix {
+	if got := Prefix(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+	}); got != wantPrefix {
 		t.Errorf("Prefix() = %q, want %q", got, wantPrefix)
 	}
 
 	wantSession := wantPrefix + "/session-1"
-	if got := SessionDir(rootDir, ownerKind, ownerName, ns, cluster, session); got != wantSession {
+	if got := SessionDir(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+		SessionName: session,
+	}); got != wantSession {
 		t.Errorf("SessionDir() = %q, want %q", got, wantSession)
 	}
 
@@ -30,27 +43,57 @@ func TestClusterLogsPaths(t *testing.T) {
 	}
 
 	wantNode := wantSession + "/node-1"
-	if got := NodeDir(rootDir, ownerKind, ownerName, ns, cluster, session, node); got != wantNode {
+	if got := NodeDir(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+		SessionName: session,
+	}, node); got != wantNode {
 		t.Errorf("NodeDir() = %q, want %q", got, wantNode)
 	}
 
 	wantLogs := wantNode + "/logs"
-	if got := LogsDir(rootDir, ownerKind, ownerName, ns, cluster, session, node); got != wantLogs {
+	if got := LogsDir(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+		SessionName: session,
+	}, node); got != wantLogs {
 		t.Errorf("LogsDir() = %q, want %q", got, wantLogs)
 	}
 
 	wantNodeEvents := wantNode + "/node_events"
-	if got := NodeEventsDir(rootDir, ownerKind, ownerName, ns, cluster, session, node); got != wantNodeEvents {
+	if got := NodeEventsDir(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+		SessionName: session,
+	}, node); got != wantNodeEvents {
 		t.Errorf("NodeEventsDir() = %q, want %q", got, wantNodeEvents)
 	}
 
 	wantJobEvents := wantNode + "/job_events/01000000"
-	if got := JobEventsDir(rootDir, ownerKind, ownerName, ns, cluster, session, node, jobID); got != wantJobEvents {
+	if got := JobEventsDir(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+		SessionName: session,
+	}, node, jobID); got != wantJobEvents {
 		t.Errorf("JobEventsDir() = %q, want %q", got, wantJobEvents)
 	}
 
 	wantJobEventsNoID := wantNode + "/job_events"
-	if got := JobEventsDir(rootDir, ownerKind, ownerName, ns, cluster, session, node, ""); got != wantJobEventsNoID {
+	if got := JobEventsDir(rootDir, &utils.ClusterInfo{
+		OwnerKind: ownerKind,
+		OwnerName: ownerName,
+		Namespace: ns,
+		Name:      cluster,
+		SessionName: session,
+	}, node, ""); got != wantJobEventsNoID {
 		t.Errorf("JobEventsDir(no jobID) = %q, want %q", got, wantJobEventsNoID)
 	}
 

historyserver/test/e2e/azureblob_collector_test.go

@@ -64,7 +64,14 @@ func testAzureBlobUploadOnGracefulShutdown(test Test, g *WithT, namespace *corev
 	sessionID := GetSessionIDFromHeadPod(test, g, rayCluster)
 	headNodeID := GetNodeIDFromPod(test, g, HeadPod(test, rayCluster), "ray-head")
 	workerNodeID := GetNodeIDFromPod(test, g, FirstWorkerPod(test, rayCluster), "ray-worker")
-	sessionPrefix := fmt.Sprintf("%s/", clusterlogs.SessionDir("", "", "", rayCluster.Namespace, rayCluster.Name, sessionID))
+	clusterInfo := &utils.ClusterInfo{
+		Namespace: rayCluster.Namespace,
+		Name:      rayCluster.Name,
+		SessionName: sessionID,
+		OwnerName: "",
+		OwnerKind: "",
+	}
+	sessionPrefix := fmt.Sprintf("%s/", clusterlogs.SessionDir("", clusterInfo))
 
 	err := test.Client().Ray().RayV1().
 		RayClusters(rayCluster.Namespace).
@@ -89,7 +96,14 @@ func testAzureBlobSeparatesFilesBySession(test Test, g *WithT, namespace *corev1
 	sessionID := GetSessionIDFromHeadPod(test, g, rayCluster)
 	headNodeID := GetNodeIDFromPod(test, g, HeadPod(test, rayCluster), "ray-head")
 	workerNodeID := GetNodeIDFromPod(test, g, FirstWorkerPod(test, rayCluster), "ray-worker")
-	sessionPrefix := fmt.Sprintf("%s/", clusterlogs.SessionDir("", "", "", rayCluster.Namespace, rayCluster.Name, sessionID))
+	clusterInfo := &utils.ClusterInfo{
+		Namespace: rayCluster.Namespace,
+		Name:      rayCluster.Name,
+		SessionName: sessionID,
+		OwnerName: "",
+		OwnerKind: "",
+	}
+	sessionPrefix := fmt.Sprintf("%s/", clusterlogs.SessionDir("", clusterInfo))
 
 	killContainerAndWaitForRestart(test, g, HeadPod(test, rayCluster), "ray-head")
 	killContainerAndWaitForRestart(test, g, FirstWorkerPod(test, rayCluster), "ray-worker")
@@ -109,7 +123,14 @@ func testAzureBlobResumesUploadsOnRestart(test Test, g *WithT, namespace *corev1
 
 	dummySessionID := fmt.Sprintf("test-recovery-session-%s", namespace.Name)
 	dummyNodeID := fmt.Sprintf("head-node-%s", namespace.Name)
-	sessionPrefix := fmt.Sprintf("%s/", clusterlogs.SessionDir("", "", "", rayCluster.Namespace, rayCluster.Name, dummySessionID))
+	clusterInfo := &utils.ClusterInfo{
+		Namespace: rayCluster.Namespace,
+		Name:      rayCluster.Name,
+		SessionName: dummySessionID,
+		OwnerName: "",
+		OwnerKind: "",
+	}
+	sessionPrefix := fmt.Sprintf("%s/", clusterlogs.SessionDir("", clusterInfo))
 
 	headPod, err := GetHeadPod(test, rayCluster)
 	g.Expect(err).NotTo(HaveOccurred())