[History Server] Treat suspended RayCluster as dead in isDead

historyserver/pkg/historyserver/session_processor.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	k8stypes "k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -23,8 +24,8 @@ type SessionStatus int
 const (
 	// SessionStatusUnknown is the zero value, reserved as a defensive guard.
 	SessionStatusUnknown SessionStatus = iota
-	// SessionStatusLive means the RayCluster CR is still present and the
-	// session is intentionally skipped.
+	// SessionStatusLive means the RayCluster CR is still present and not
+	// suspended, so the session is intentionally skipped.
 	SessionStatusLive
 	// SessionStatusProcessed means events were ingested into EventHandler's
 	// in-memory state.
@@ -78,7 +79,7 @@ func (p *SessionProcessor) ProcessSession(ctx context.Context, session utils.Clu
 	return SessionStatusProcessed, h.BuildSnapshot(session), nil
 }
 
-// isDead determines if the RayCluster CR is absent.
+// isDead determines if the RayCluster CR is absent or suspended.
 //
 // Known limit: An old session of a still-running RayCluster will be misclassified as live.
 func (p *SessionProcessor) isDead(ctx context.Context, session utils.ClusterInfo) (bool, error) {
@@ -93,6 +94,9 @@ func (p *SessionProcessor) isDead(ctx context.Context, session utils.ClusterInfo
 	if err != nil {
 		return false, err
 	}
+	if meta.IsStatusConditionTrue(rc.Status.Conditions, string(rayv1.RayClusterSuspended)) {
+		return true, nil
+	}
 	return false, nil
 }
 

historyserver/pkg/historyserver/session_processor_test.go

@@ -32,6 +32,19 @@ func rayCluster(namespace, name string) *rayv1.RayCluster {
 	}
 }
 
+func rayClusterWithCondition(namespace, name string, condType rayv1.RayClusterConditionType) *rayv1.RayCluster {
+	rc := rayCluster(namespace, name)
+	rc.Status.Conditions = []metav1.Condition{
+		{
+			Type:               string(condType),
+			Status:             metav1.ConditionTrue,
+			Reason:             string(condType),
+			LastTransitionTime: metav1.Now(),
+		},
+	}
+	return rc
+}
+
 func TestIsDead(t *testing.T) {
 	const (
 		ns             = "default"
@@ -54,6 +67,16 @@ func TestIsDead(t *testing.T) {
 			cr:       rayCluster(ns, name),
 			wantDead: false,
 		},
+		{
+			name:     "RayCluster CR present and fully suspended -> dead",
+			cr:       rayClusterWithCondition(ns, name, rayv1.RayClusterSuspended),
+			wantDead: true,
+		},
+		{
+			name:     "RayCluster CR present but still suspending -> alive",
+			cr:       rayClusterWithCondition(ns, name, rayv1.RayClusterSuspending),
+			wantDead: false,
+		},
 	}
 
 	for _, tc := range tests {