add e2e test for endpoint api/v0/tasks/timeline

Signed-off-by: AndySung320 <andysung0320@gmail.com>

Author: AndySung320

historyserver/test/e2e/collector_test.go

@@ -87,7 +87,7 @@ func testCollectorUploadOnGracefulShutdown(test Test, g *WithT, namespace *corev
 	_ = ApplyRayJobAndWaitForCompletion(test, g, namespace, rayCluster)
 
 	// Define variables for constructing S3 object prefix.
-	clusterNameID := fmt.Sprintf("%s_%s", rayCluster.Name, RayClusterID)
+	clusterNameID := fmt.Sprintf("%s_%s", rayCluster.Name, rayCluster.Namespace)
 	sessionID := GetSessionIDFromHeadPod(test, g, rayCluster)
 	headNodeID := GetNodeIDFromPod(test, g, HeadPod(test, rayCluster), "ray-head")
 	workerNodeID := GetNodeIDFromPod(test, g, FirstWorkerPod(test, rayCluster), "ray-worker")
@@ -134,7 +134,7 @@ func testCollectorSeparatesFilesBySession(test Test, g *WithT, namespace *corev1
 	// Submit a Ray job to the existing cluster.
 	_ = ApplyRayJobAndWaitForCompletion(test, g, namespace, rayCluster)
 
-	clusterNameID := fmt.Sprintf("%s_%s", rayCluster.Name, RayClusterID)
+	clusterNameID := fmt.Sprintf("%s_%s", rayCluster.Name, rayCluster.Namespace)
 	sessionID := GetSessionIDFromHeadPod(test, g, rayCluster)
 	headNodeID := GetNodeIDFromPod(test, g, HeadPod(test, rayCluster), "ray-head")
 	workerNodeID := GetNodeIDFromPod(test, g, FirstWorkerPod(test, rayCluster), "ray-worker")
@@ -196,7 +196,7 @@ func testCollectorResumesUploadsOnRestart(test Test, g *WithT, namespace *corev1
 	// Use namespace name to ensure test isolation (avoid conflicts from previous test runs)
 	dummySessionID := fmt.Sprintf("test-recovery-session-%s", namespace.Name)
 	dummyNodeID := fmt.Sprintf("head-node-%s", namespace.Name)
-	clusterNameID := fmt.Sprintf("%s_%s", rayCluster.Name, RayClusterID)
+	clusterNameID := fmt.Sprintf("%s_%s", rayCluster.Name, rayCluster.Namespace)
 	sessionPrefix := fmt.Sprintf("log/%s/%s/", clusterNameID, dummySessionID)
 
 	// Inject "leftover" logs BEFORE killing collector.

historyserver/test/e2e/historyserver_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/ray-project/kuberay/historyserver/pkg/utils"
 	. "github.com/ray-project/kuberay/historyserver/test/support"
@@ -30,6 +32,14 @@ func TestHistoryServer(t *testing.T) {
 			name:     "Live cluster: historyserver endpoints should be accessible",
 			testFunc: testLiveClusters,
 		},
+		{
+			name:     "/api/v0/tasks/timeline endpoint (live cluster)",
+			testFunc: testTimelineEndpointLiveCluster,
+		},
+		{
+			name:     "/api/v0/tasks/timeline endpoint (dead cluster)",
+			testFunc: testTimelineEndpointDeadCluster,
+		},
 	}
 
 	for _, tt := range tests {
@@ -133,3 +143,129 @@ func getClusterFromList(test Test, g *WithT, historyServerURL, clusterName, name
 
 	return result
 }
+
+// testTimelineEndpointLiveCluster verifies that the history server can return timeline data from a live cluster.
+//
+// The test case follows these steps:
+// 1. Prepare test environment by applying a Ray cluster
+// 2. Submit a Ray job to the existing cluster
+// 3. Apply History Server and get its URL
+// 4. Verify that the timeline endpoint returns valid Chrome Tracing format
+func testTimelineEndpointLiveCluster(test Test, g *WithT, namespace *corev1.Namespace, s3Client *s3.S3) {
+	rayCluster := PrepareTestEnv(test, g, namespace, s3Client)
+	ApplyRayJobAndWaitForCompletion(test, g, namespace, rayCluster)
+	ApplyHistoryServer(test, g, namespace)
+	historyServerURL := GetHistoryServerURL(test, g, namespace)
+
+	clusterInfo := getClusterFromList(test, g, historyServerURL, rayCluster.Name, namespace.Name)
+	g.Expect(clusterInfo.SessionName).To(Equal(LiveSessionName), "Live cluster should have sessionName='live'")
+
+	client := CreateHTTPClientWithCookieJar(g)
+	setClusterContext(test, g, client, historyServerURL, namespace.Name, rayCluster.Name, clusterInfo.SessionName)
+	test.T().Run("should return valid timeline data", func(t *testing.T) {
+		g := NewWithT(t)
+		verifyTimelineResponse(g, client, historyServerURL)
+	})
+
+	DeleteS3Bucket(test, g, s3Client)
+	LogWithTimestamp(test.T(), "Live cluster timeline endpoint test completed")
+}
+
+// testTimelineEndpointDeadCluster verifies that the history server can return timeline data from S3 after a cluster is deleted.
+//
+// The test case follows these steps:
+// 1. Prepare test environment by applying a Ray cluster
+// 2. Submit a Ray job to the existing cluster
+// 3. Delete RayCluster to trigger event upload to S3
+// 4. Apply History Server and get its URL
+// 5. Verify that the timeline endpoint returns valid Chrome Tracing format from S3 data
+func testTimelineEndpointDeadCluster(test Test, g *WithT, namespace *corev1.Namespace, s3Client *s3.S3) {
+	rayCluster := PrepareTestEnv(test, g, namespace, s3Client)
+	ApplyRayJobAndWaitForCompletion(test, g, namespace, rayCluster)
+
+	// Delete RayCluster to trigger event upload
+	err := test.Client().Ray().RayV1().RayClusters(namespace.Name).Delete(test.Ctx(), rayCluster.Name, metav1.DeleteOptions{})
+	g.Expect(err).NotTo(HaveOccurred())
+	LogWithTimestamp(test.T(), "Deleted RayCluster %s/%s", namespace.Name, rayCluster.Name)
+
+	// Wait for cluster to be fully deleted (ensures events are uploaded to S3)
+	g.Eventually(func() error {
+		_, err := GetRayCluster(test, namespace.Name, rayCluster.Name)
+		return err
+	}, TestTimeoutMedium).Should(WithTransform(k8serrors.IsNotFound, BeTrue()))
+
+	ApplyHistoryServer(test, g, namespace)
+	historyServerURL := GetHistoryServerURL(test, g, namespace)
+
+	clusterInfo := getClusterFromList(test, g, historyServerURL, rayCluster.Name, namespace.Name)
+	g.Expect(clusterInfo.SessionName).NotTo(Equal(LiveSessionName))
+
+	client := CreateHTTPClientWithCookieJar(g)
+	setClusterContext(test, g, client, historyServerURL, namespace.Name, rayCluster.Name, clusterInfo.SessionName)
+
+	test.T().Run("should return timeline data from S3", func(t *testing.T) {
+		g := NewWithT(t)
+		verifyTimelineResponse(g, client, historyServerURL)
+	})
+
+	DeleteS3Bucket(test, g, s3Client)
+	LogWithTimestamp(test.T(), "Dead cluster timeline endpoint test completed")
+}
+
+// verifyTimelineResponse verifies the timeline endpoint returns valid Chrome Tracing format
+func verifyTimelineResponse(g *WithT, client *http.Client, historyServerURL string) {
+	g.Eventually(func(gg Gomega) {
+		resp, err := client.Get(historyServerURL + "/api/v0/tasks/timeline")
+		gg.Expect(err).NotTo(HaveOccurred())
+		defer resp.Body.Close()
+		gg.Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+		body, err := io.ReadAll(resp.Body)
+		gg.Expect(err).NotTo(HaveOccurred())
+
+		var events []map[string]any
+		err = json.Unmarshal(body, &events)
+		gg.Expect(err).NotTo(HaveOccurred())
+
+		// Should have at least some events
+		gg.Expect(len(events)).To(BeNumerically(">", 0), "Timeline should have at least one event")
+
+		// Verify metadata and trace events exist
+		hasProcessName := false
+		hasThreadName := false
+		hasTraceEvent := false
+
+		for _, event := range events {
+			name, _ := event["name"].(string)
+			ph, _ := event["ph"].(string)
+
+			if name == "process_name" && ph == "M" {
+				hasProcessName = true
+				args, ok := event["args"].(map[string]any)
+				gg.Expect(ok).To(BeTrue(), "process_name should have args")
+				gg.Expect(args["name"]).NotTo(BeNil(), "process_name args should have 'name'")
+			}
+			if name == "thread_name" && ph == "M" {
+				hasThreadName = true
+			}
+			if ph == "X" {
+				hasTraceEvent = true
+				// Verify trace event has required fields
+				gg.Expect(event["cat"]).NotTo(BeNil(), "Trace event should have 'cat'")
+				gg.Expect(event["ts"]).NotTo(BeNil(), "Trace event should have 'ts'")
+				gg.Expect(event["dur"]).NotTo(BeNil(), "Trace event should have 'dur'")
+				gg.Expect(event["args"]).NotTo(BeNil(), "Trace event should have 'args'")
+
+				// Verify args structure
+				args, ok := event["args"].(map[string]any)
+				gg.Expect(ok).To(BeTrue())
+				gg.Expect(args["task_id"]).NotTo(BeNil(), "args should have 'task_id'")
+				gg.Expect(args["job_id"]).NotTo(BeNil(), "args should have 'job_id'")
+			}
+		}
+
+		gg.Expect(hasProcessName).To(BeTrue(), "Should have process_name metadata event")
+		gg.Expect(hasThreadName).To(BeTrue(), "Should have thread_name metadata event")
+		gg.Expect(hasTraceEvent).To(BeTrue(), "Should have at least one trace event")
+	}, TestTimeoutShort).Should(Succeed())
+}

historyserver/test/support/raycluster.go

@@ -17,11 +17,31 @@ const (
 	RayClusterID           = "default"
 )
 
+// injectCollectorRayClusterID injects the ray-cluster-id argument into all collector containers.
+func injectCollectorRayClusterID(containers []corev1.Container, rayClusterID string) {
+	for i := range containers {
+		if containers[i].Name == "collector" {
+			containers[i].Command = append(
+				containers[i].Command,
+				fmt.Sprintf("--ray-cluster-id=%s", rayClusterID),
+			)
+		}
+	}
+}
+
 // ApplyRayClusterWithCollector deploys a Ray cluster with the collector sidecar into the test namespace.
 func ApplyRayClusterWithCollector(test Test, g *WithT, namespace *corev1.Namespace) *rayv1.RayCluster {
 	rayClusterFromYaml := DeserializeRayClusterYAML(test, RayClusterManifestPath)
 	rayClusterFromYaml.Namespace = namespace.Name
 
+	// Inject namespace name as ray-cluster-id for head group collector
+	injectCollectorRayClusterID(rayClusterFromYaml.Spec.HeadGroupSpec.Template.Spec.Containers, namespace.Name)
+
+	// Inject namespace name as ray-cluster-id for worker group collectors
+	for wg := range rayClusterFromYaml.Spec.WorkerGroupSpecs {
+		injectCollectorRayClusterID(rayClusterFromYaml.Spec.WorkerGroupSpecs[wg].Template.Spec.Containers, namespace.Name)
+	}
+
 	rayCluster, err := test.Client().Ray().RayV1().
 		RayClusters(namespace.Name).
 		Create(test.Ctx(), rayClusterFromYaml, metav1.CreateOptions{})