fix: Align all nullable fields

Signed-off-by: JiangJiaWei1103 <waynechuang97@gmail.com>

Author: JiangJiaWei1103

historyserver/pkg/eventserver/eventserver.go

@@ -972,17 +972,6 @@ func normalizeTaskIDsToHex(task *types.Task) {
 			return ""
 		}
 
-		// Check if the Ray ID is nil.
-		isNil, err := utils.IsBase64Nil(base64ID)
-		if err != nil {
-			logrus.Errorf("Failed to check if Ray ID is nil: %v", err)
-			return base64ID
-		}
-		if isNil {
-			logrus.Infof("Ray ID is nil, keeping original base64 ID: %s", base64ID)
-			return base64ID
-		}
-
 		hexID, err := utils.ConvertBase64ToHex(base64ID)
 		if err != nil {
 			logrus.Errorf("Failed to convert ID from base64 to hex, keeping original: %v", err)

historyserver/pkg/eventserver/types/task.go

@@ -101,6 +101,10 @@ type TaskLogInfo struct {
 
 // Task's fields are populated from the TASK_DEFINITION_EVENT, ACTOR_TASK_DEFINITION_EVENT, and TASK_LIFECYCLE_EVENT.
 // A TASK_DEFINITION_EVENT or an ACTOR_TASK_DEFINITION_EVENT is expected to be emitted once per task attempt,
+// For proto definitions, please refer to:
+// https://github.com/ray-project/ray/tree/master/src/ray/protobuf/public.
+// For field population, please refer to:
+// https://github.com/ray-project/ray/blob/36be009ae360788550e541d81806493f52963730/src/ray/core_worker/task_event_buffer.cc#L189-L295.
 type Task struct {
 	// TaskID and TaskAttempt form the unique identifier for a task.
 	TaskID      string `json:"taskId"`
@@ -130,21 +134,23 @@ type Task struct {
 	RefIDs               map[string]string `json:"refIds,omitempty"`
 	SerializedRuntimeEnv string            `json:"serializedRuntimeEnv,omitempty"`
 	// CallSite is the human readable stacktrace of the actor task invocation.
-	CallSite string `json:"callSite,omitempty"`
+	CallSite *string `json:"callSite,omitempty"`
 	// LabelSelector is the key-value label constraints of the node to schedule this actor task on.
 	LabelSelector map[string]string `json:"labelSelector,omitempty"`
 
 	// The task execution information, populated from TASK_LIFECYCLE_EVENT.
 	StateTransitions []TaskStateTransition `json:"stateTransitions,omitempty"`
-	RayErrorInfo     RayErrorInfo          `json:"rayErrorInfo,omitempty"`
+	// RayErrorInfo is only populated when the state error info has any values.
+	// Ref: https://github.com/ray-project/ray/blob/36be009ae360788550e541d81806493f52963730/src/ray/core_worker/task_event_buffer.cc#L263-L265.
+	RayErrorInfo *RayErrorInfo `json:"rayErrorInfo,omitempty"`
 
 	NodeID    string `json:"nodeId,omitempty"`
 	WorkerID  string `json:"workerId,omitempty"`
 	WorkerPID int    `json:"workerPid,omitempty"`
 	// Whether the task is paused by the debugger.
-	IsDebuggerPaused bool `json:"isDebuggerPaused,omitempty"`
+	IsDebuggerPaused *bool `json:"isDebuggerPaused,omitempty"`
 	// Actor task repr name, if applicable.
-	ActorReprName string `json:"actorReprName,omitempty"`
+	ActorReprName *string `json:"actorReprName,omitempty"`
 
 	// TaskLogInfo is just added at https://github.com/ray-project/ray/pull/60287.
 	// TODO(jwj): Add support for TaskLogInfo.
@@ -340,6 +346,21 @@ func (t Task) DeepCopy() Task {
 		copy(cp.StateTransitions, t.StateTransitions)
 	}
 
+	if t.RayErrorInfo != nil {
+		rayErrorInfo := *t.RayErrorInfo
+		cp.RayErrorInfo = &rayErrorInfo
+	}
+
+	if t.IsDebuggerPaused != nil {
+		cp.IsDebuggerPaused = new(bool)
+		*cp.IsDebuggerPaused = *t.IsDebuggerPaused
+	}
+
+	if t.ActorReprName != nil {
+		actorReprName := *t.ActorReprName
+		cp.ActorReprName = &actorReprName
+	}
+
 	if t.TaskLogInfo != nil {
 		taskLogInfo := *t.TaskLogInfo
 		cp.TaskLogInfo = &taskLogInfo

historyserver/pkg/historyserver/router.go

@@ -939,6 +939,14 @@ func (s *ServerHandler) getTasks(req *restful.Request, resp *restful.Response) {
 // The schema aligns with the Ray Dashboard API.
 // Ref: https://github.com/ray-project/ray/blob/d0b1d151d8ea964a711e451d0ae736f8bf95b629/python/ray/util/state/common.py#L730-L819.
 func formatTaskForResponse(task eventtypes.Task, detail bool) map[string]interface{} {
+	setNullableField := func(result map[string]interface{}, key string, value interface{}) {
+		if value != nil {
+			result[key] = value
+		} else {
+			result[key] = nil
+		}
+	}
+
 	// TODO(jwj): Maybe define result schema in types.go.
 	result := map[string]interface{}{
 		"task_id":            task.TaskID,
@@ -953,7 +961,11 @@ func formatTaskForResponse(task eventtypes.Task, detail bool) map[string]interfa
 		"node_id":            task.NodeID,
 		"worker_id":          task.WorkerID,
 		"worker_pid":         task.WorkerPID,
-		"error_type":         string(task.RayErrorInfo.ErrorType),
+	}
+	if task.RayErrorInfo != nil {
+		result["error_type"] = string(task.RayErrorInfo.ErrorType)
+	} else {
+		result["error_type"] = nil
 	}
 
 	if detail {
@@ -969,7 +981,12 @@ func formatTaskForResponse(task eventtypes.Task, detail bool) map[string]interfa
 				"log_files":             []string{},
 			},
 		}
-		result["placement_group_id"] = task.PlacementGroupID
+		isNil, err := utils.IsHexNil(task.PlacementGroupID)
+		if isNil || err != nil {
+			result["placement_group_id"] = nil
+		} else {
+			result["placement_group_id"] = task.PlacementGroupID
+		}
 
 		events := make([]map[string]interface{}, 0, len(task.StateTransitions))
 		for _, event := range task.StateTransitions {
@@ -983,9 +1000,13 @@ func formatTaskForResponse(task eventtypes.Task, detail bool) map[string]interfa
 		// Ref: https://github.com/ray-project/ray/blob/d0b1d151d8ea964a711e451d0ae736f8bf95b629/python/ray/util/state/common.py#L1616-L1622.
 		// result["profiling_data"] = task.ProfilingData
 		result["task_log_info"] = task.TaskLogInfo
-		result["error_message"] = task.RayErrorInfo.ErrorMessage
-		result["is_debugger_paused"] = task.IsDebuggerPaused
-		result["call_site"] = task.CallSite
+		if task.RayErrorInfo != nil {
+			result["error_message"] = task.RayErrorInfo.ErrorMessage
+		} else {
+			result["error_message"] = nil
+		}
+		setNullableField(result, "is_debugger_paused", task.IsDebuggerPaused)
+		setNullableField(result, "call_site", task.CallSite)
 		if task.LabelSelector != nil {
 			result["label_selector"] = task.LabelSelector
 		} else {

historyserver/pkg/utils/utils.go

@@ -254,22 +254,29 @@ func ConvertBase64ToHex(input string) (string, error) {
 	return hexStr, nil
 }
 
-func IsBase64Nil(base64Str string) (bool, error) {
-	bytes, err := base64.StdEncoding.DecodeString(base64Str)
-	if err != nil {
-		return false, err
-	}
+// IsHexNil returns true if hexStr decodes to a non-empty byte slice where every byte is 0xff.
+func IsHexNil(hexStr string) (bool, error) {
+	s := strings.TrimSpace(hexStr)
 
-	if len(bytes) == 0 {
+	if len(s) == 0 {
 		return false, nil
 	}
 
-	for i := range bytes {
-		if bytes[i] != 0xff {
+	// Hex string must have even length.
+	if len(s)%2 != 0 {
+		return false, hex.ErrLength
+	}
+
+	bytes, err := hex.DecodeString(s)
+	if err != nil {
+		return false, err
+	}
+
+	for _, v := range bytes {
+		if v != 0xff {
 			return false, nil
 		}
 	}
-
 	return true, nil
 }