add the implementation of historyserver beta version

Signed-off-by: KunWuLuan <kunwuluan@gmail.com>

Author: KunWuLuan

historyserver/backend/historyserver/clientmanager.go

@@ -0,0 +1,91 @@
+package historyserver
+
+import (
+	"context"
+	"strings"
+
+	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	"github.com/sirupsen/logrus"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type ClientManager struct {
+	configs []*rest.Config
+	clients []client.Client
+}
+
+func (c *ClientManager) ListRayClusters(ctx context.Context) ([]*rayv1.RayCluster, error) {
+	list := []*rayv1.RayCluster{}
+	for _, c := range c.clients {
+		listOfRayCluster := rayv1.RayClusterList{}
+		err := c.List(ctx, &listOfRayCluster)
+		if err != nil {
+			logrus.Errorf("Failed to list RayClusters: %v", err)
+			continue
+		}
+		for _, rayCluster := range listOfRayCluster.Items {
+			list = append(list, &rayCluster)
+		}
+	}
+	return list, nil
+}
+
+func NewClientManager(kubeconfigs string) *ClientManager {
+	kubeconfigList := []*rest.Config{}
+	if len(kubeconfigs) > 0 {
+		stringList := strings.Split(kubeconfigs, ",")
+		if len(stringList) > 1 {
+			// historyserver is able to get query from live gcs, which is not safe.
+			// we hope to replace these apis with one events.
+			logrus.Errorf("Only one kubeconfig is supported.")
+		}
+		for _, kubeconfig := range stringList {
+			if kubeconfig != "" {
+				c, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
+				if err != nil {
+					logrus.Errorf("Failed to build config from kubeconfig: %v", err)
+					continue
+				}
+				c.QPS = 50
+				c.Burst = 100
+				kubeconfigList = append(kubeconfigList, c)
+				logrus.Infof("add config from path: %v", kubeconfig)
+				break
+			}
+		}
+	} else {
+		c, err := rest.InClusterConfig()
+		c.QPS = 50
+		c.Burst = 100
+		if err == nil {
+			kubeconfigList = append(kubeconfigList, c)
+			logrus.Infof("add config from in cluster config")
+		} else {
+			logrus.Errorf("Failed to build config from kubeconfig: %v", err)
+		}
+	}
+	scheme := runtime.NewScheme()
+	utilruntime.Must(rayv1.AddToScheme(scheme))
+	clientList := []client.Client{}
+	for _, config := range kubeconfigList {
+		c, err := client.New(config, client.Options{
+			Scheme: scheme,
+		})
+		if err != nil {
+			logrus.Errorf("Failed to create client: %v", err)
+			continue
+		}
+		clientList = append(clientList, c)
+	}
+	logrus.Infof("create client manager successfully, clients: %v", len(clientList))
+	return &ClientManager{
+		configs: kubeconfigList,
+		clients: clientList,
+	}
+}

historyserver/backend/historyserver/reader.go

@@ -1,13 +1,16 @@
 package historyserver
 
 import (
+	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"path"
 	"path/filepath"
 	"sort"
+	"time"
 
 	"github.com/emicklei/go-restful/v3"
 	"github.com/ray-project/kuberay/historyserver/utils"
@@ -17,14 +20,83 @@ import (
 func (s *ServerHandler) listClusters(limit int) []utils.ClusterInfo {
 	// 初始的继续标记
 	logrus.Debugf("Prepare to get list clusters info ...")
+	ctx := context.Background()
+	liveClusters, _ := s.clientManager.ListRayClusters(ctx)
+	liveClusterNames := []string{}
+	liveClusterInfos := []utils.ClusterInfo{}
+	for _, liveCluster := range liveClusters {
+		liveClusterInfo := utils.ClusterInfo{
+			Name:            liveCluster.Name,
+			Namespace:       liveCluster.Namespace,
+			CreateTime:      liveCluster.CreationTimestamp.String(),
+			CreateTimeStamp: liveCluster.CreationTimestamp.Unix(),
+			SessionName:     "live",
+		}
+		liveClusterInfos = append(liveClusterInfos, liveClusterInfo)
+		liveClusterNames = append(liveClusterNames, liveCluster.Name)
+	}
+	logrus.Infof("live clusters: %v", liveClusterNames)
 	clusters := s.reader.List()
 	sort.Sort(utils.ClusterInfoList(clusters))
 	if limit > 0 {
 		clusters = clusters[:limit]
 	}
+	clusters = append(liveClusterInfos, clusters...)
 	return clusters
 }
 
+func (s *ServerHandler) _getNodeLogs(rayClusterNameID, sessionId, nodeId, dir string) ([]byte, error) {
+	logPath := path.Join(sessionId, "logs", nodeId)
+	if dir != "" {
+		logPath = path.Join(logPath, dir)
+	}
+	files := s.reader.ListFiles(rayClusterNameID, logPath)
+	ret := map[string]interface{}{
+		"data": map[string]interface{}{
+			"result": map[string]interface{}{
+				"padding": files,
+			},
+		},
+	}
+	return json.Marshal(ret)
+}
+
+func (s *ServerHandler) GetNodes(rayClusterNameID, sessionId string) ([]byte, error) {
+	logPath := path.Join(sessionId, "logs")
+	nodes := s.reader.ListFiles(rayClusterNameID, logPath)
+	templ := map[string]interface{}{
+		"result": true,
+		"msg":    "Node summary fetched.",
+		"data": map[string]interface{}{
+			"summary": []map[string]interface{}{},
+		},
+	}
+	nodeSummary := []map[string]interface{}{}
+	for _, node := range nodes {
+		nodeSummary = append(nodeSummary, map[string]interface{}{
+			"raylet": map[string]interface{}{
+				"nodeId": path.Clean(node),
+				"state":  "ALIVE",
+			},
+			"ip": "UNKNOWN",
+		})
+	}
+	templ["data"].(map[string]interface{})["summary"] = nodeSummary
+	return json.Marshal(templ)
+}
+
+func (s *ServerHandler) ClusterInfo(rayClusterNameID string) []byte {
+	templ := `{
+    "result": true,
+    "msg": "Got formatted cluster status.",
+    "data": {
+        "clusterStatus": "======== Autoscaler status: %f ========\nNode status\n---------------------------------------------------------------\nActive:\n (no active nodes)\nIdle:\n 0 headgroup\nPending:\n (no pending nodes)\nRecent failures:\n (no failures)\n\nResources\n---------------------------------------------------------------\nTotal Usage:\n 0B/0B memory\n 0B/0B object_store_memory\n\nFrom request_resources:\n (none)\nPending Demands:\n (no resource demands)"
+    }
+}`
+	afterRender := fmt.Sprintf(templ, time.Now().Format("2006-01-02 15:04:05.000000"))
+	return []byte(afterRender)
+}
+
 func (s *ServerHandler) MetaKeyInfo(rayClusterNameID, key string) []byte {
 	baseObject := path.Join(utils.GetMetaDirByNameID(s.rootDir, rayClusterNameID), key)
 	logrus.Infof("Prepare to get object %s info ...", baseObject)
@@ -37,8 +109,8 @@ func (s *ServerHandler) MetaKeyInfo(rayClusterNameID, key string) []byte {
 	return data
 }
 
-func (s *ServerHandler) LogKeyInfo(rayClusterNameID, nodeID, key string, lines int64) []byte {
-	baseObject := path.Join(utils.GetLogDirByNameID(s.rootDir, rayClusterNameID, nodeID), key)
+func (s *ServerHandler) LogKeyInfo(rayClusterNameID, nodeID, sessionId, key string, lines int64) []byte {
+	baseObject := path.Join(utils.GetLogDirByNameID(s.rootDir, rayClusterNameID, nodeID, sessionId), key)
 	logrus.Infof("Prepare to get object %s info ...", baseObject)
 	body := s.reader.GetContent(rayClusterNameID, baseObject)
 	data, err := io.ReadAll(body)
@@ -56,8 +128,22 @@ func (s *ServerHandler) staticFileHandler(req *restful.Request, resp *restful.Re
 	// Get the path parameter
 	path := req.PathParameter("path")
 
+	isHomePage := true
+	_, err := req.Request.Cookie(COOKIE_CLUSTER_NAME_KEY)
+	isHomePage = err != nil
+	prefix := ""
+	if isHomePage {
+		prefix = "homepage"
+	} else {
+		version := "v2.51.0"
+		if versionCookie, err := req.Request.Cookie(COOKIE_DASHBOARD_VERSION_KEY); err == nil {
+			version = versionCookie.Value
+		}
+		prefix = version + "/client/build"
+	}
+
 	// Construct the full path to the static directory
-	fullPath := filepath.Join(s.dashboardDir, "static", path)
+	fullPath := filepath.Join(s.dashboardDir, prefix, "static", path)
 	logrus.Infof("staticFileHandler fullpath %s", fullPath)
 
 	// Check if the full path exists