[history server] Improve api/v0/logs/file

historyserver/pkg/historyserver/reader.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"path"
+	"regexp"
 	"sort"
 	"strings"
 
@@ -26,6 +27,15 @@ const (
 	MAX_LOG_LIMIT = 10000
 )
 
+// ANSI escape code pattern for filtering colored output from logs
+// Matches patterns like: \x1b[31m (red color), \x1b[0m (reset), etc.
+var ansiEscapePattern = regexp.MustCompile(`\x1b\[[0-9;]+m`)
+
+// filterAnsiEscapeCodes removes ANSI escape sequences from log content
+func filterAnsiEscapeCodes(content []byte) []byte {
+	return ansiEscapePattern.ReplaceAll(content, []byte(""))
+}
+
 func (s *ServerHandler) listClusters(limit int) []utils.ClusterInfo {
 	// Initial continuation marker
 	logrus.Debugf("Prepare to get list clusters info ...")
@@ -73,18 +83,31 @@ func (s *ServerHandler) _getNodeLogs(rayClusterNameID, sessionId, nodeId, dir st
 	return json.Marshal(ret)
 }
 
-func (s *ServerHandler) _getNodeLogFile(rayClusterNameID, sessionID, nodeID, filename string, maxLines int) ([]byte, error) {
-	logPath := path.Join(sessionID, "logs", nodeID, filename)
+func (s *ServerHandler) _getNodeLogFile(rayClusterNameID, sessionID string, options GetLogFileOptions) ([]byte, error) {
+	// Build log path with attempt_number if specified
+	logPath := path.Join(sessionID, "logs", options.NodeID, options.Filename)
+	if options.AttemptNumber > 0 {
+		// Append attempt number to filename (e.g., worker.log.1)
+		logPath = fmt.Sprintf("%s.%d", logPath, options.AttemptNumber)
+	}
 
 	reader := s.reader.GetContent(rayClusterNameID, logPath)
 
 	if reader == nil {
 		return nil, utils.NewHTTPError(fmt.Errorf("log file not found: %s", logPath), http.StatusNotFound)
 	}
 
+	maxLines := options.Lines
 	if maxLines < 0 {
-		// -1 means read all lines, match Ray Dashboard API behavior
-		return io.ReadAll(reader)
+		// -1 means read all lines
+		content, err := io.ReadAll(reader)
+		if err != nil {
+			return nil, err
+		}
+		if options.FilterAnsiCode {
+			content = filterAnsiEscapeCodes(content)
+		}
+		return content, nil
 	}
 
 	if maxLines == 0 {
@@ -132,7 +155,12 @@ func (s *ServerHandler) _getNodeLogFile(rayClusterNameID, sessionID, nodeID, fil
 		lines = append(buffer[start:], buffer[:start]...)
 	}
 
-	return []byte(strings.Join(lines, "\n")), nil
+	result := []byte(strings.Join(lines, "\n"))
+	if options.FilterAnsiCode {
+		result = filterAnsiEscapeCodes(result)
+	}
+
+	return result, nil
 }
 
 func (s *ServerHandler) GetNodes(rayClusterNameID, sessionId string) ([]byte, error) {

historyserver/pkg/historyserver/router.go

@@ -109,6 +109,10 @@ func routerAPI(s *ServerHandler) {
 		Doc("get logfile").Param(ws.QueryParameter("node_id", "node_id")).
 		Param(ws.QueryParameter("filename", "filename")).
 		Param(ws.QueryParameter("lines", "lines")).
+		Param(ws.QueryParameter("timeout", "timeout")).
+		Param(ws.QueryParameter("attempt_number", "attempt_number")).
+		Param(ws.QueryParameter("download_file", "download_file")).
+		Param(ws.QueryParameter("filter_ansi_code", "filter_ansi_code")).
 		Produces("text/plain").
 		Writes("")) // Placeholder for specific return type
 
@@ -573,24 +577,26 @@ func (s *ServerHandler) getNodeLogFile(req *restful.Request, resp *restful.Respo
 	clusterNamespace := req.Attribute(COOKIE_CLUSTER_NAMESPACE_KEY).(string)
 	sessionName := req.Attribute(COOKIE_SESSION_NAME_KEY).(string)
 
-	// Parse query parameters
-	nodeID := req.QueryParameter("node_id")
-	filename := req.QueryParameter("filename")
-	lines := req.QueryParameter("lines")
+	// Parse query parameters into GetLogFileOptions struct
+	options, err := parseGetLogFileOptions(req)
+	if err != nil {
+		resp.WriteErrorString(http.StatusBadRequest, err.Error())
+		return
+	}
 
 	// Validate required parameters
-	if nodeID == "" {
+	if options.NodeID == "" {
 		resp.WriteErrorString(http.StatusBadRequest, "Missing required parameter: node_id")
 		return
 	}
-	if filename == "" {
+	if options.Filename == "" {
 		resp.WriteErrorString(http.StatusBadRequest, "Missing required parameter: filename")
 		return
 	}
 
 	// Prevent path traversal attacks (e.g., ../../etc/passwd)
-	if !fs.ValidPath(nodeID) || !fs.ValidPath(filename) {
-		resp.WriteErrorString(http.StatusBadRequest, fmt.Sprintf("invalid path: path traversal not allowed (node_id=%s, filename=%s)", nodeID, filename))
+	if !fs.ValidPath(options.NodeID) || !fs.ValidPath(options.Filename) {
+		resp.WriteErrorString(http.StatusBadRequest, fmt.Sprintf("invalid path: path traversal not allowed (node_id=%s, filename=%s)", options.NodeID, options.Filename))
 		return
 	}
 
@@ -599,18 +605,7 @@ func (s *ServerHandler) getNodeLogFile(req *restful.Request, resp *restful.Respo
 		return
 	}
 
-	// Convert lines parameter to int
-	maxLines := 0
-	if lines != "" {
-		parsedLines, err := strconv.Atoi(lines)
-		if err != nil {
-			resp.WriteErrorString(http.StatusBadRequest, fmt.Sprintf("invalid lines parameter: %s", lines))
-			return
-		}
-		maxLines = parsedLines
-	}
-
-	content, err := s._getNodeLogFile(clusterNameID+"_"+clusterNamespace, sessionName, nodeID, filename, maxLines)
+	content, err := s._getNodeLogFile(clusterNameID+"_"+clusterNamespace, sessionName, options)
 	if err != nil {
 		var httpErr *utils.HTTPError
 		if errors.As(err, &httpErr) {
@@ -622,9 +617,62 @@ func (s *ServerHandler) getNodeLogFile(req *restful.Request, resp *restful.Respo
 		}
 		return
 	}
+
+	// Set Content-Disposition header if download_file is requested
+	if options.DownloadFile {
+		resp.AddHeader("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", options.Filename))
+	}
+
 	resp.Write(content)
 }
 
+// parseGetLogFileOptions parses query parameters into GetLogFileOptions struct
+func parseGetLogFileOptions(req *restful.Request) (GetLogFileOptions, error) {
+	options := GetLogFileOptions{
+		NodeID:   req.QueryParameter("node_id"),
+		Filename: req.QueryParameter("filename"),
+	}
+
+	// Parse lines parameter
+	if linesStr := req.QueryParameter("lines"); linesStr != "" {
+		lines, err := strconv.Atoi(linesStr)
+		if err != nil {
+			return options, fmt.Errorf("invalid lines parameter: %s", linesStr)
+		}
+		options.Lines = lines
+	}
+
+	// Parse timeout parameter
+	if timeoutStr := req.QueryParameter("timeout"); timeoutStr != "" {
+		timeout, err := strconv.Atoi(timeoutStr)
+		if err != nil {
+			return options, fmt.Errorf("invalid timeout parameter: %s", timeoutStr)
+		}
+		options.Timeout = timeout
+	}
+
+	// Parse attempt_number parameter
+	if attemptStr := req.QueryParameter("attempt_number"); attemptStr != "" {
+		attemptNumber, err := strconv.Atoi(attemptStr)
+		if err != nil {
+			return options, fmt.Errorf("invalid attempt_number parameter: %s", attemptStr)
+		}
+		options.AttemptNumber = attemptNumber
+	}
+
+	// Parse filter_ansi_code parameter (boolean)
+	if filterStr := req.QueryParameter("filter_ansi_code"); filterStr != "" {
+		options.FilterAnsiCode = strings.ToLower(filterStr) == "true"
+	}
+
+	// Parse download_file parameter (boolean)
+	if downloadStr := req.QueryParameter("download_file"); downloadStr != "" {
+		options.DownloadFile = strings.ToLower(downloadStr) == "true"
+	}
+
+	return options, nil
+}
+
 func (s *ServerHandler) getTaskSummarize(req *restful.Request, resp *restful.Response) {
 	clusterName := req.Attribute(COOKIE_CLUSTER_NAME_KEY).(string)
 	clusterNamespace := req.Attribute(COOKIE_CLUSTER_NAMESPACE_KEY).(string)

historyserver/pkg/historyserver/types.go

@@ -1,5 +1,25 @@
 package historyserver
 
+// GetLogFileOptions contains all options for fetching log files
+type GetLogFileOptions struct {
+	// Required parameters
+	NodeID   string // The node id where the log file is located
+	Filename string // The log file name
+
+	// Optional parameters with defaults
+	// Number of lines to return, default to DEFAULT_LOG_LIMIT (1000)
+	// -1 = all lines
+	Lines int
+	// Timeout in seconds for the request, default to 0 (no timeout)
+	Timeout int
+	// Attempt number for task retries, default to 0 (first attempt)
+	AttemptNumber int
+	// Whether to filter ANSI escape codes from logs, default to False
+	FilterAnsiCode bool
+	// Whether to set Content-Disposition header for file download, default to false
+	DownloadFile bool
+}
+
 type ReplyTaskInfo struct {
 	Data   TaskInfoData `json:"data"`
 	Msg    string       `json:"msg"`