Merge branch 'master' of github.com:ray-project/kuberay into improve-logs-file-test

Signed-off-by: machichima <nary12321@gmail.com>

Author: machichima

.github/ISSUE_TEMPLATE/bug-report.yml

@@ -34,6 +34,7 @@ body:
         - "kubectl-plugin"
         - "dashboard"
         - "ci"
+        - "historyserver"
         - "Others"
     validations:
       required: true

historyserver/docs/set_up_historyserver.md

@@ -170,3 +170,19 @@ curl -b ~/cookies.txt "http://localhost:8080/api/jobs/"
 # Cluster Status
 curl -b ~/cookies.txt "http://localhost:8080/api/cluster_status"
 ```
+
+### Live Cluster with grafana
+
+```bash
+# Install grafana. ref: https://docs.ray.io/en/latest/cluster/kubernetes/k8s-ecosystem/prometheus-grafana.html#step-2-install-kubernetes-prometheus-stack-via-helm-chart
+./install/prometheus/install.sh --auto-load-dashboard true
+
+# Apply RayCluster with Grafana setting
+kubectl apply -f ray-operator/config/samples/ray-cluster.embed-grafana.yaml
+
+# Get live session cookie. (Port-forward is required)
+curl -c ~/cookies.txt "http://localhost:8080/enter_cluster/default/raycluster-embed-grafana/live"
+
+# Request to grafana health endpoint
+curl -b ~/cookies.txt http://localhost:8080/api/grafana_health
+```

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

@@ -49,8 +49,8 @@ func (r *RayLogHandler) Start(stop <-chan struct{}) error {
 
 func (r *RayLogHandler) Run(stop <-chan struct{}) error {
 	// watchPath := r.LogDir
-	r.prevLogsDir = "/tmp/ray/prev-logs"
-	r.persistCompleteLogsDir = "/tmp/ray/persist-complete-logs"
+	r.prevLogsDir = filepath.Join("/tmp", "ray", "prev-logs")
+	r.persistCompleteLogsDir = filepath.Join("/tmp", "ray", "persist-complete-logs")
 
 	// Initialize log file paths storage
 	r.logFilePaths = make(map[string]bool)
@@ -98,7 +98,7 @@ func (r *RayLogHandler) processSessionLatestLogs() {
 	logrus.Info("Processing session_latest logs on shutdown...")
 
 	// Resolve the session_latest symlink to get the real session directory
-	sessionLatestDir := "/tmp/ray/session_latest"
+	sessionLatestDir := filepath.Join("/tmp", "ray", "session_latest")
 	sessionRealDir, err := filepath.EvalSymlinks(sessionLatestDir)
 	if err != nil {
 		logrus.Errorf("Failed to resolve session_latest symlink: %v", err)
@@ -124,7 +124,7 @@ func (r *RayLogHandler) processSessionLatestLogs() {
 	}
 
 	// Read node ID from /tmp/ray/raylet_node_id
-	nodeIDBytes, err := os.ReadFile("/tmp/ray/raylet_node_id")
+	nodeIDBytes, err := os.ReadFile(filepath.Join("/tmp", "ray", "raylet_node_id"))
 	if err != nil {
 		logrus.Errorf("Failed to read raylet_node_id: %v", err)
 		return
@@ -178,7 +178,8 @@ func (r *RayLogHandler) processSessionLatestLogs() {
 func (r *RayLogHandler) processSessionLatestLogFile(absoluteLogPathName, sessionID, nodeID string) error {
 	// Calculate relative path within logs directory
 	// The logsDir is /tmp/ray/session_latest/logs
-	logsDir := filepath.Join("/tmp/ray/session_latest", "logs")
+	sessionLatestDir := filepath.Join("/tmp", "ray", "session_latest")
+	logsDir := filepath.Join(sessionLatestDir, "logs")
 	relativePath, err := filepath.Rel(logsDir, absoluteLogPathName)
 	if err != nil {
 		return fmt.Errorf("failed to get relative path for %s: %w", absoluteLogPathName, err)
@@ -709,7 +710,8 @@ func (r *RayLogHandler) processPrevLogFile(absoluteLogPathName, localLogDir, ses
 //		session_2024-12-15_10-30-45_123456    ← Empty file! The path itself is the information
 //		session_2024-12-15_14-20-10_789012
 func (r *RayLogHandler) WatchSessionLatestLoops() {
-	sessionLatestDir := "/tmp/ray"
+	sessionLatestDir := filepath.Join("/tmp", "ray")
+	sessionLatestSymlink := filepath.Join(sessionLatestDir, "session_latest")
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
 		logrus.Errorf("Failed to create fsnotify watcher for session_latest: %v", err)
@@ -736,7 +738,7 @@ func (r *RayLogHandler) WatchSessionLatestLoops() {
 			}
 
 			logrus.Infof("File system event in session_latest: %s %s", event.Op, event.Name)
-			if event.Name == "/tmp/ray/session_latest" {
+			if event.Name == sessionLatestSymlink {
 				continue
 			}
 			rel, err := filepath.Rel(sessionLatestDir, event.Name)

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

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net/http"
 	"path"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -38,7 +39,7 @@ func NewCollector(config *types.RayCollectorConfig, writer storage.StorageWriter
 		Writer:       writer,
 		ShutdownChan: make(chan struct{}),
 	}
-	logDir := strings.TrimSpace(path.Join(config.SessionDir, utils.RAY_SESSIONDIR_LOGDIR_NAME))
+	logDir := strings.TrimSpace(filepath.Join(config.SessionDir, utils.RAY_SESSIONDIR_LOGDIR_NAME))
 	handler.LogDir = logDir
 	// rootMetaDir uses flat key format (name_id) for S3/OSS performance optimization.
 	// See utils.connector for the design rationale.

historyserver/pkg/eventserver/eventserver.go

@@ -22,6 +22,7 @@ type EventHandler struct {
 
 	ClusterTaskMap  *types.ClusterTaskMap
 	ClusterActorMap *types.ClusterActorMap
+	ClusterJobMap   *types.ClusterJobMap
 }
 
 var eventFilePattern = regexp.MustCompile(`-\d{4}-\d{2}-\d{2}-\d{2}$`)
@@ -44,6 +45,9 @@ func NewEventHandler(reader storage.StorageReader) *EventHandler {
 		ClusterActorMap: &types.ClusterActorMap{
 			ClusterActorMap: make(map[string]*types.ActorMap),
 		},
+		ClusterJobMap: &types.ClusterJobMap{
+			ClusterJobMap: make(map[string]*types.JobMap),
+		},
 	}
 }
 
@@ -115,6 +119,7 @@ func (h *EventHandler) Run(stop chan struct{}, numOfEventProcessors int) error {
 			clusterList := h.reader.List()
 			for _, clusterInfo := range clusterList {
 				clusterNameNamespace := clusterInfo.Name + "_" + clusterInfo.Namespace
+				clusterSessionKey := utils.BuildClusterSessionKey(clusterInfo.Name, clusterInfo.Namespace, clusterInfo.SessionName)
 				eventFileList := append(h.getAllJobEventFiles(clusterInfo), h.getAllNodeEventFiles(clusterInfo)...)
 
 				logrus.Infof("current eventFileList for cluster %s is: %v", clusterInfo.Name, eventFileList)
@@ -145,7 +150,7 @@ func (h *EventHandler) Run(stop chan struct{}, numOfEventProcessors int) error {
 						if curr == nil {
 							continue
 						}
-						curr["clusterName"] = clusterInfo.Name + "_" + clusterInfo.Namespace
+						curr["clusterName"] = clusterSessionKey
 						eventProcessorChannels[i%numOfEventProcessors] <- curr
 					}
 				}
@@ -525,6 +530,170 @@ func (h *EventHandler) storeEvent(eventMap map[string]any) error {
 		// TODO: Handle actor task definition event
 		// This is related to GET /api/v0/tasks (type=ACTOR_TASK)
 		logrus.Debugf("ACTOR_TASK_DEFINITION_EVENT received, not yet implemented")
+
+	case types.DRIVER_JOB_DEFINITION_EVENT:
+		// NOTE: When event comes in, JobID will be in base64, processing will convert it to Hex
+		jobDef, ok := eventMap["driverJobDefinitionEvent"]
+		if !ok {
+			return fmt.Errorf("event does not have 'driverJobDefinitionEvent'")
+		}
+
+		jsonDriverJobDefinition, err := json.Marshal(jobDef)
+		if err != nil {
+			return err
+		}
+
+		var currJob types.Job
+		if err := json.Unmarshal(jsonDriverJobDefinition, &currJob); err != nil {
+			return err
+		}
+
+		// Convert JobID from base64 to hex
+		currJob.JobID, err = utils.ConvertBase64ToHex(currJob.JobID)
+		if err != nil {
+			logrus.Errorf("Failed to convert JobID from base64 to Hex, will keep JobID in base64: %v", err)
+		}
+
+		// Convert DriverNodeID from base64 to hex
+		currJob.DriverNodeID, err = utils.ConvertBase64ToHex(currJob.DriverNodeID)
+		if err != nil {
+			logrus.Errorf("Failed to convert DriverNodeID from base64 to hex, will keep DriverNodeID in base64: %v", err)
+		}
+
+		jobMap := h.ClusterJobMap.GetOrCreateJobMap(currentClusterName)
+		jobMap.CreateOrMergeJob(currJob.JobID, func(j *types.Job) {
+			// If for some reason jobID is empty, we will keep whatever is in 'j'
+			var existingJobID string
+			if currJob.JobID == "" {
+				existingJobID = j.JobID
+			}
+
+			// ========== Lifecycle-derived fields ==========
+			// These fields are set by DRIVER_JOB_LIFECYCLE_EVENT
+			// We need to preserve them if lifecycle event arrived before definition event
+			existingStateTransitions := j.StateTransitions
+			existingState := j.State
+			existingStartTime := j.StartTime
+			existingEndTime := j.EndTime
+
+			// Overwrite with definition fields
+			*j = currJob
+
+			// Restore lifecycle-derived fields if they existed
+			if len(existingStateTransitions) > 0 {
+				if existingJobID != "" {
+					// This means that jobID was somehow empty.
+					j.JobID = existingJobID
+				}
+				j.StateTransitions = existingStateTransitions
+				j.State = existingState
+				j.StartTime = existingStartTime
+				j.EndTime = existingEndTime
+			}
+
+			// ========== Definition-only fields ==========
+			// Status, StatusTransitions, Message, DriverExitCode
+			// are ONLY set by DRIVER_JOB_DEFINITION_EVENT
+			// They are already in currJob, no need to restore
+		})
+	case types.DRIVER_JOB_LIFECYCLE_EVENT:
+		// NOTE: When event comes in, JobID will be in base64, processing will convert it to Hex
+		jobLifecycleEvent, ok := eventMap["driverJobLifecycleEvent"].(map[string]any)
+		if !ok {
+			return fmt.Errorf("invalid driverJobLifecycleEvent format")
+		}
+
+		// Get JobID and also convert JobID to hex from base64
+		// Will leave it as it if it's empty or somehow not a string
+		jobId, ok := jobLifecycleEvent["jobId"].(string)
+		if !ok {
+			logrus.Errorf("jobID is missing or is not a string, leaving it as is")
+		} else {
+			jobId, _ = utils.ConvertBase64ToHex(jobId)
+		}
+		stateTransitionUnstructed, _ := jobLifecycleEvent["stateTransitions"].([]any)
+
+		if len(stateTransitionUnstructed) == 0 || jobId == "" {
+			return nil
+		}
+
+		// TODO(chiayi): Will need to convert status timeline once it is added as well
+		// Following fields are related to status transition:
+		//  - status
+		//  - message
+		//  - errorType
+		//  - driverExitCode
+		var stateTransitions []types.JobStateTransition
+		for _, transition := range stateTransitionUnstructed {
+			tr, ok := transition.(map[string]any)
+			if !ok {
+				continue
+			}
+			state, _ := tr["state"].(string)
+			timestampStr, _ := tr["timestamp"].(string)
+
+			var timestamp time.Time
+			if timestampStr != "" {
+				timestamp, _ = time.Parse(time.RFC3339Nano, timestampStr)
+			}
+
+			stateTransitions = append(stateTransitions, types.JobStateTransition{
+				State:     types.JobState(state),
+				Timestamp: timestamp,
+			})
+		}
+
+		if len(stateTransitions) == 0 {
+			return nil
+		}
+
+		jobMap := h.ClusterJobMap.GetOrCreateJobMap(currentClusterName)
+		jobMap.CreateOrMergeJob(jobId, func(j *types.Job) {
+			// TODO(chiayi): take care of status (job progress) state if part of DriverJobLifecycleEvent
+			j.JobID = jobId
+
+			type stateTransitionKey struct {
+				State     string
+				Timestamp int64
+			}
+
+			existingStateKeys := make(map[stateTransitionKey]bool)
+			for _, t := range j.StateTransitions {
+				existingStateKeys[stateTransitionKey{string(t.State), t.Timestamp.UnixNano()}] = true
+			}
+
+			for _, t := range stateTransitions {
+				key := stateTransitionKey{string(t.State), t.Timestamp.UnixNano()}
+				if !existingStateKeys[key] {
+					j.StateTransitions = append(j.StateTransitions, t)
+					existingStateKeys[key] = true
+				}
+			}
+
+			sort.Slice(j.StateTransitions, func(i, k int) bool {
+				return j.StateTransitions[i].Timestamp.Before(j.StateTransitions[k].Timestamp)
+			})
+
+			if len(j.StateTransitions) == 0 {
+				return
+			}
+
+			lastStateTransition := j.StateTransitions[len(j.StateTransitions)-1]
+			j.State = lastStateTransition.State
+
+			if j.StartTime.IsZero() {
+				for _, t := range j.StateTransitions {
+					if t.State == types.CREATED {
+						j.StartTime = t.Timestamp
+						break
+					}
+				}
+			}
+
+			if lastStateTransition.State == types.JOBFINISHED {
+				j.EndTime = lastStateTransition.Timestamp
+			}
+		})
 	default:
 		logrus.Infof("Event not supported, skipping: %v", eventMap)
 	}
@@ -708,3 +877,41 @@ func (h *EventHandler) GetActorsMap(clusterName string) map[string]types.Actor {
 	}
 	return actors
 }
+
+func (h *EventHandler) GetJobsMap(clusterName string) map[string]types.Job {
+	h.ClusterJobMap.RLock()
+	defer h.ClusterJobMap.RUnlock()
+
+	jobMap, ok := h.ClusterJobMap.ClusterJobMap[clusterName]
+	if !ok {
+		return map[string]types.Job{}
+	}
+
+	jobMap.Lock()
+	defer jobMap.Unlock()
+
+	jobs := make(map[string]types.Job, len(jobMap.JobMap))
+	for id, job := range jobMap.JobMap {
+		jobs[id] = job.DeepCopy()
+	}
+	return jobs
+}
+
+func (h *EventHandler) GetJobByJobID(clusterName, jobID string) (types.Job, bool) {
+	h.ClusterJobMap.RLock()
+	defer h.ClusterJobMap.RUnlock()
+
+	jobMap, ok := h.ClusterJobMap.ClusterJobMap[clusterName]
+	if !ok {
+		return types.Job{}, false
+	}
+
+	jobMap.Lock()
+	defer jobMap.Unlock()
+
+	job, ok := jobMap.JobMap[jobID]
+	if !ok {
+		return types.Job{}, false
+	}
+	return job.DeepCopy(), true
+}