[history server] Web Server + Event Processor

historyserver/Dockerfile.historyserver

@@ -0,0 +1,29 @@
+FROM golang:1.25.1 AS builder
+
+ENV GOPROXY=https://proxy.golang.org,direct
+WORKDIR /historyserver
+
+# Copy the go modules and manifests.
+COPY go.mod go.mod
+COPY go.sum go.sum
+# Cache dependencies to avoid re-downloading when only sources change.
+RUN go mod download
+
+# Copy the go source.
+COPY cmd/historyserver/main.go cmd/historyserver/main.go
+# need collector because storage's interface is put in here, will change
+# after this is merged
+# https://github.com/ray-project/kuberay/pull/4302
+COPY pkg/collector/ pkg/collector/
+COPY pkg/historyserver/ pkg/historyserver/
+COPY pkg/storage/ pkg/storage/
+COPY pkg/utils/ pkg/utils/
+COPY pkg/eventserver/ pkg/eventserver/
+
+# Build the historyserver binary.
+COPY Makefile Makefile
+RUN make buildhistoryserver GOOS=linux GOARCH=amd64
+
+FROM ubuntu:22.04
+RUN apt-get update && apt-get upgrade -y && rm -rf /var/cache/apt/ && apt-get install -y ca-certificates
+COPY --from=builder /historyserver/output/bin/historyserver /usr/local/bin/historyserver

historyserver/Makefile

@@ -9,6 +9,7 @@ BIN_DIR=$(OUT_DIR)/bin
 BINARY_NAME=historyserver
 BINARY_NAME_COLLECTOR=collector
 COLLECTOR_IMG ?= collector:v0.1.0
+HISTORYSERVER_IMG ?= historyserver:v0.1.0
 
 # Setting SHELL to bash allows bash commands to be executed by recipes.
 # Options are set to exit when a recipe line exits non-zero or a piped command fails.
@@ -18,7 +19,7 @@ COMMIT_SHORT ?= $(shell git rev-parse --short HEAD)
 BRANCH ?= $(shell git branch --show-current)
 VERSION ?= $(shell git describe --tags --long|awk -F '-' '{print $$1"."$$2"-"$$3""}')
 
-PACKAGE = gitlab.alibaba-inc.com/eml/historyserver
+PACKAGE = github.com/ray-project/kuberay/historyserver
 
 GO_LDFLAGS := -extldflags "-static"
 # GO_LDFLAGS += -w -s # Drop debugging symbols.
@@ -66,13 +67,16 @@ simplebuild:
 mod:
 	go mod tidy
 
+.PHONY: localimage-build
+localimage-build: localimage-collector localimage-historyserver
+
 .PHONY: localimage-collector
 localimage-collector:
 	docker build -t $(COLLECTOR_IMG) -f Dockerfile.collector .
 
-.PHONY: localimage
-localimage: dockerbuilder_instance
-	docker buildx build -t historyserver:laster --platform linux/amd64 . --load
+.PHONY: localimage-historyserver
+localimage-historyserver: dockerbuilder_instance
+	docker build -t $(HISTORYSERVER_IMG) -f Dockerfile.historyserver .
 
 .PHONY: dockerbuilder_instance
 dockerbuilder_instance:

historyserver/cmd/historyserver/main.go

@@ -1 +1,101 @@
 package main
+
+import (
+	"encoding/json"
+	"flag"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"github.com/ray-project/kuberay/historyserver/pkg/collector"
+	"github.com/ray-project/kuberay/historyserver/pkg/collector/types"
+	"github.com/ray-project/kuberay/historyserver/pkg/eventserver"
+	"github.com/ray-project/kuberay/historyserver/pkg/historyserver"
+	"github.com/sirupsen/logrus"
+)
+
+func main() {
+	runtimeClassName := ""
+	rayRootDir := ""
+	kubeconfigs := ""
+	runtimeClassConfigPath := "/var/collector-config/data"
+	dashboardDir := ""
+	flag.StringVar(&runtimeClassName, "runtime-class-name", "", "")
+	flag.StringVar(&rayRootDir, "ray-root-dir", "", "")
+	flag.StringVar(&kubeconfigs, "kubeconfigs", "", "")
+	flag.StringVar(&dashboardDir, "dashboard-dir", "/dashboard", "")
+	flag.StringVar(&runtimeClassConfigPath, "runtime-class-config-path", "", "") //"/var/collector-config/data"
+	flag.Parse()
+
+	cliMgr := historyserver.NewClientManager(kubeconfigs)
+
+	jsonData := make(map[string]interface{})
+	if runtimeClassConfigPath != "" {
+		data, err := os.ReadFile(runtimeClassConfigPath)
+		if err != nil {
+			panic("Failed to read runtime class config " + err.Error())
+		}
+		err = json.Unmarshal(data, &jsonData)
+		if err != nil {
+			panic("Failed to parse runtime class config: " + err.Error())
+		}
+	}
+
+	registry := collector.GetReaderRegistry()
+	factory, ok := registry[runtimeClassName]
+	if !ok {
+		panic("Not supported runtime class name: " + runtimeClassName + ".")
+	}
+
+	globalConfig := types.RayHistoryServerConfig{
+		RootDir: rayRootDir,
+	}
+
+	reader, err := factory(&globalConfig, jsonData)
+	if err != nil {
+		panic("Failed to create reader for runtime class name: " + runtimeClassName + ".")
+	}
+
+	// Create EventHandler with storage reader
+	eventHandler := eventserver.NewEventHandler(reader)
+
+	// WaitGroup to track goroutine completion
+	var wg sync.WaitGroup
+
+	// Start EventHandler in background goroutine
+	eventStop := make(chan struct{}, 1)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		logrus.Info("Starting EventHandler in background...")
+		if err := eventHandler.Run(eventStop, 2); err != nil {
+			logrus.Errorf("EventHandler stopped with error: %v", err)
+		}
+		logrus.Info("EventHandler shutdown complete")
+	}()
+
+	handler := historyserver.NewServerHandler(&globalConfig, dashboardDir, reader, cliMgr, eventHandler)
+
+	sigChan := make(chan os.Signal, 1)
+	stop := make(chan struct{}, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		handler.Run(stop)
+		logrus.Info("HTTP server shutdown complete")
+	}()
+
+	<-sigChan
+	logrus.Info("Received shutdown signal, initiating graceful shutdown...")
+
+	// Stop both the server and the event handler
+	stop <- struct{}{}
+	eventStop <- struct{}{}
+
+	// Wait for both goroutines to complete
+	wg.Wait()
+	logrus.Info("Graceful shutdown complete")
+}

historyserver/config/historyserver.yaml

@@ -0,0 +1,63 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: historyserver #TODO: to specify your service name
+  labels:
+    app: historyserver
+spec:
+  selector:
+    app: historyserver
+  ports:
+  - protocol: TCP
+    name: http
+    port: 30080
+    targetPort: 8080
+  type: ClusterIP
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: historyserver-demo
+  labels:
+    app: historyserver
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: historyserver
+  template:
+    metadata:
+      labels:
+        app: historyserver
+    spec:
+      serviceAccountName: historyserver
+      containers:
+      - name: historyserver
+        env:
+          - name: S3DISABLE_SSL
+            value: "true"
+          - name: AWS_S3ID
+            value: minioadmin
+          - name: AWS_S3SECRET
+            value: minioadmin
+          - name: AWS_S3TOKEN
+            value: ""
+          - name: S3_BUCKET
+            value: "ray-historyserver"
+          - name: S3_ENDPOINT
+            value: "minio-service.minio-dev:9000"
+          - name: S3_REGION
+            value: "test"
+          - name: S3FORCE_PATH_STYLE
+            value: "true"
+        image: historyserver:v0.1.0
+        imagePullPolicy: IfNotPresent
+        command:
+        - historyserver
+        - --runtime-class-name=s3
+        - --ray-root-dir=log
+        ports:
+        - containerPort: 8080
+        resources:
+          limits:
+            cpu: "500m"

historyserver/config/rayjob.yaml

@@ -8,11 +8,11 @@ spec:
     import ray
     ray.init()
 
-    @ray.remote
+    @ray.remote(num_cpus=0.5)
     def my_task(x):
         return x * 2
 
-    @ray.remote
+    @ray.remote(num_cpus=0.5)
     class Counter:
         def __init__(self):
             self.count = 0

historyserver/config/service_account.yaml

@@ -0,0 +1,27 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: historyserver
+automountServiceAccountToken: true
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: raycluster-reader
+rules:
+- apiGroups: ["ray.io"]
+  resources: ["rayclusters"]
+  verbs: ["list", "get"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: historyserver
+  namespace: default
+subjects:
+- kind: ServiceAccount
+  name: historyserver
+  namespace: default
+roleRef:
+  kind: ClusterRole
+  name: raycluster-reader