diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json new file mode 100644 index 0000000..5bd6144 --- /dev/null +++ b/.devcontainer/devcontainer.json @@ -0,0 +1,12 @@ +{ + "name": "RLark Development", + "image": "harbor.infini-ai.com/share/devpods/universal:alpha1", + "containerEnv": { + "GOPROXY": "https://goproxy.cn,direct", + "GO111MODULE": "on", + "GOPRIVATE": "gitlab.infini-ai.com", + "GONOSUMDB": "gitlab.infini-ai.com", + "GONOPROXY": "gitlab.infini-ai.com", + "GOAUTH": "${containerWorkspaceFolder}/.devcontainer/goauth.sh" + } +} diff --git a/.devcontainer/goauth.sh b/.devcontainer/goauth.sh new file mode 100755 index 0000000..34abcca --- /dev/null +++ b/.devcontainer/goauth.sh @@ -0,0 +1,14 @@ +#!/bin/bash + +host=gitlab.infini-ai.com +result=$(echo -e "protocol=https\nhost=$host" | git credential fill 2>/dev/null) +username=$(echo "$result" | grep "^username=" | cut -d'=' -f2) +password=$(echo "$result" | grep "^password=" | cut -d'=' -f2) +if [ -z "$username" ] || [ -z "$password" ]; then + exit 1 +fi + +echo "https://$host" +echo "" +echo "Authorization: Basic $(echo -n "$username:$password" | base64)" +echo "" diff --git a/.devcontainer/init-vscode.sh b/.devcontainer/init-vscode.sh new file mode 100755 index 0000000..994e1db --- /dev/null +++ b/.devcontainer/init-vscode.sh @@ -0,0 +1,25 @@ +#!/bin/bash + +scriptFullPath="$(realpath "$0")" +projectPath="$(dirname "$(dirname "$scriptFullPath")")" + +echo "Script full path: $projectPath" + +mkdir -p "$projectPath/.vscode/data/zsh" +echo "*" > "$projectPath/.vscode/.gitignore" +cat < "$projectPath/.vscode/settings.json" +{ + "terminal.integrated.env.linux": { + "XDG_DATA_HOME": "$projectPath/.vscode/data", + "ZDOTDIR": "$projectPath/.vscode/data/zsh" + }, + "terminal.integrated.defaultProfile.linux": "zsh" +} +EOF +cat <<'EOF' > "$projectPath/.vscode/data/zsh/.zshrc" +export ZSH="$HOME/.oh-my-zsh" +ZSH_THEME="ys" +plugins=(git docker kubectl) +source $ZSH/oh-my-zsh.sh +zstyle ':omz:update' mode disabled +EOF diff --git a/.gitignore b/.gitignore index 4270f5e..918a86f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ # IDE .idea/ .vscode/ +.codex/ *.swp # Build output diff --git a/api/config/crd/bases/rlinf.io_jobs.yaml b/api/config/crd/bases/rlinf.io_jobs.yaml index 6651f9f..d081a67 100644 --- a/api/config/crd/bases/rlinf.io_jobs.yaml +++ b/api/config/crd/bases/rlinf.io_jobs.yaml @@ -107,6 +107,11 @@ spec: properties: kind: type: string + pvcSizeGbMap: + additionalProperties: + format: int32 + type: integer + type: object pvcStorageMap: additionalProperties: type: string diff --git a/api/config/crd/bases/rlinf.io_nodes.yaml b/api/config/crd/bases/rlinf.io_nodes.yaml index 0568924..0b8d3bc 100644 --- a/api/config/crd/bases/rlinf.io_nodes.yaml +++ b/api/config/crd/bases/rlinf.io_nodes.yaml @@ -88,6 +88,11 @@ spec: 等 Warning 事件及镜像拉取/调度相关事件）。控制面 Task reconciler 在 Task 处于 Pending 期间聚合各节点事件到 Task.status.events，供前端展示。 items: + description: |- + NodeEvent represents a Kubernetes Event observed on a node that is relevant + for surfacing to operators (e.g. DiskPressure warnings, FailedScheduling, + image pull failures). The node-agent collects Warning events plus a small + set of Normal scheduling/pulling events and writes them to Node.status.events. properties: count: format: int32 @@ -127,6 +132,8 @@ spec: type: string pullProgress: items: + description: PullProgress captures the progress of an in-flight + image pull on a node. properties: downloaded: format: int64 diff --git a/api/config/crd/bases/rlinf.io_tasks.yaml b/api/config/crd/bases/rlinf.io_tasks.yaml index 5f77d0b..bbb01ae 100644 --- a/api/config/crd/bases/rlinf.io_tasks.yaml +++ b/api/config/crd/bases/rlinf.io_tasks.yaml @@ -99,6 +99,11 @@ spec: properties: kind: type: string + pvcSizeGbMap: + additionalProperties: + format: int32 + type: integer + type: object pvcStorageMap: additionalProperties: type: string diff --git a/api/config/crd/bases/rlinf.io_workflows.yaml b/api/config/crd/bases/rlinf.io_workflows.yaml index 01ce4da..b5caedf 100644 --- a/api/config/crd/bases/rlinf.io_workflows.yaml +++ b/api/config/crd/bases/rlinf.io_workflows.yaml @@ -118,6 +118,11 @@ spec: properties: kind: type: string + pvcSizeGbMap: + additionalProperties: + format: int32 + type: integer + type: object pvcStorageMap: additionalProperties: type: string diff --git a/api/kubeclients/applyconfiguration/rlark.io/v1alpha1/kubernetesworkloadspec.go b/api/kubeclients/applyconfiguration/rlark.io/v1alpha1/kubernetesworkloadspec.go index 167b5aa..b3446f1 100644 --- a/api/kubeclients/applyconfiguration/rlark.io/v1alpha1/kubernetesworkloadspec.go +++ b/api/kubeclients/applyconfiguration/rlark.io/v1alpha1/kubernetesworkloadspec.go @@ -29,6 +29,7 @@ type KubernetesWorkloadSpecApplyConfiguration struct { Replicas *int32 `json:"replicas,omitempty"` Template *v1.PodTemplateSpec `json:"template,omitempty"` PvcStorageMap map[string]string `json:"pvcStorageMap,omitempty"` + PvcSizeGbMap map[string]int32 `json:"pvcSizeGbMap,omitempty"` } // KubernetesWorkloadSpecApplyConfiguration constructs a declarative configuration of the KubernetesWorkloadSpec type for use with @@ -74,3 +75,17 @@ func (b *KubernetesWorkloadSpecApplyConfiguration) WithPvcStorageMap(entries map } return b } + +// WithPvcSizeGbMap puts the entries into the PvcSizeGbMap field in the declarative configuration +// and returns the receiver, so that objects can be build by chaining "With" function invocations. +// If called multiple times, the entries provided by each call will be put on the PvcSizeGbMap field, +// overwriting an existing map entries in PvcSizeGbMap field with the same key. +func (b *KubernetesWorkloadSpecApplyConfiguration) WithPvcSizeGbMap(entries map[string]int32) *KubernetesWorkloadSpecApplyConfiguration { + if b.PvcSizeGbMap == nil && len(entries) > 0 { + b.PvcSizeGbMap = make(map[string]int32, len(entries)) + } + for k, v := range entries { + b.PvcSizeGbMap[k] = v + } + return b +} diff --git a/api/rlark.io/v1alpha1/task_types.go b/api/rlark.io/v1alpha1/task_types.go index 81c3fdb..118932e 100644 --- a/api/rlark.io/v1alpha1/task_types.go +++ b/api/rlark.io/v1alpha1/task_types.go @@ -41,6 +41,7 @@ type KubernetesWorkloadSpec struct { Replicas *int32 `json:"replicas,omitempty"` Template corev1.PodTemplateSpec `json:"template,omitempty"` PvcStorageMap map[string]string `json:"pvcStorageMap,omitempty"` + PvcSizeGbMap map[string]int32 `json:"pvcSizeGbMap,omitempty"` } type DockerTaskSpec struct { diff --git a/api/rlark.io/v1alpha1/zz_generated.deepcopy.go b/api/rlark.io/v1alpha1/zz_generated.deepcopy.go index 660a4f6..e41e5f8 100644 --- a/api/rlark.io/v1alpha1/zz_generated.deepcopy.go +++ b/api/rlark.io/v1alpha1/zz_generated.deepcopy.go @@ -601,6 +601,13 @@ func (in *KubernetesWorkloadSpec) DeepCopyInto(out *KubernetesWorkloadSpec) { (*out)[key] = val } } + if in.PvcSizeGbMap != nil { + in, out := &in.PvcSizeGbMap, &out.PvcSizeGbMap + *out = make(map[string]int32, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KubernetesWorkloadSpec. diff --git a/apps/embodied-runtime/README.md b/apps/embodied-runtime/README.md index 3c327d0..e8b4cea 100644 --- a/apps/embodied-runtime/README.md +++ b/apps/embodied-runtime/README.md @@ -240,11 +240,11 @@ Where host device passthrough mounts existing `/dev/*` nodes, **host macvlan** c ```yaml initContainers: - name: devinit - image: rlinf/embodied-runtime:v0.1.0 - command: ["devinit", "setup"] # reads RLINF_EMBODIED_DEVINIT_SOCKET_PATH + image: busybox:latest + command: ["/opt/rlinf/bin/devinit", "setup"] # reads RLINF_EMBODIED_DEVINIT_SOCKET_PATH resources: requests: - rlinf.io/device: 1 # triggers Allocate → RunDir mount + env vars + rlinf.io/device: 1 # triggers Allocate → RunDir mount + BinDir mount + env vars limits: # required: LimitRanger/ResourceQuota rejects init containers without limits rlinf.io/device: 1 # extended-resource limits must equal requests ``` @@ -271,7 +271,7 @@ The webhook has **automatic CA management**: 2. It reads the `MutatingWebhookConfiguration` (named via `--webhook-mutating-config`); when a webhook's `caBundle` is empty it patches in the CA certificate. 3. It signs a serving certificate with the CA and starts the HTTPS server. -When the `caBundle` is non-empty the webhook leaves it alone (assumed managed); a mismatch logs a warning. The init image defaults to the auto-discovered device-plugin image (downward API), which ships `devinit` — so it usually needs no configuration. +When the `caBundle` is non-empty the webhook leaves it alone (assumed managed); a mismatch logs a warning. The init image defaults to the auto-discovered device-plugin image (downward API), falling back to `busybox:latest`. The devinit binary is mounted from the host via BinDir, so the image does not need to contain it. Device-plugin CLI flags: @@ -283,7 +283,7 @@ Device-plugin CLI flags: | `--webhook-mutating-config` | `MutatingWebhookConfiguration` name whose `caBundle` is auto-managed. | | `--webhook-service-name` / `--webhook-service-namespace` | Service fronting the webhook (forms the serving cert DNS SAN). | | `--webhook-ca-secret-name` / `--webhook-ca-secret-namespace` | Secret persisting the CA (empty = in-memory). | -| `--webhook-devinit-image` | Injected init container image (default: auto-discovered device-plugin image). | +| `--webhook-devinit-image` | Injected init container image (default: auto-discovered device-plugin image, then `busybox:latest`). The binary is mounted from the host, so the image does not need to contain devinit. | See the [Helm chart](./charts/embodied-runtime) `webhook:` values for a turnkey deployment: it renders the webhook Service, the `MutatingWebhookConfiguration` (empty `caBundle`), the required RBAC (cluster-scoped `mutatingwebhookconfigurations` get/patch + namespaced `secrets`), and wires all the flags above into the device-plugin DaemonSet. Enable it with: @@ -295,7 +295,7 @@ webhook: caSecret: # persist the CA across restarts (recommended) name: devinit-ca namespace: rlark-system # defaults to the release namespace - # devinitImage: "" # defaults to .Values.devicePlugin.image + # devinitImage: "" # defaults to .Values.devicePlugin.image, then busybox:latest (binary is mounted from host) ``` The webhook only renders when **both** `webhook.enabled` and `config.hostMacvlans` are set; enabling it without macvlans is a no-op (the handler injects nothing). diff --git a/apps/embodied-runtime/README.zh-CN.md b/apps/embodied-runtime/README.zh-CN.md index 2d98f99..75bb622 100644 --- a/apps/embodied-runtime/README.zh-CN.md +++ b/apps/embodied-runtime/README.zh-CN.md @@ -238,11 +238,11 @@ host_devices: ```yaml initContainers: - name: devinit - image: rlinf/embodied-runtime:v0.1.0 - command: ["devinit", "setup"] # 读取 RLINF_EMBODIED_DEVINIT_SOCKET_PATH + image: busybox:latest + command: ["/opt/rlinf/bin/devinit", "setup"] # 读取 RLINF_EMBODIED_DEVINIT_SOCKET_PATH resources: requests: - rlinf.io/device: 1 # 触发 Allocate → RunDir 挂载 + 环境变量 + rlinf.io/device: 1 # 触发 Allocate → RunDir 挂载 + BinDir 挂载 + 环境变量 limits: # 必填：LimitRanger/ResourceQuota 会拒绝未设置 limits 的 init 容器 rlinf.io/device: 1 # 扩展资源的 limits 必须等于 requests ``` @@ -269,7 +269,7 @@ host_macvlans: 2. 读取 `MutatingWebhookConfiguration`（由 `--webhook-mutating-config` 指定）；当某 webhook 的 `caBundle` 为空时，把 CA 证书 patch 进去。 3. 用该 CA 签发服务证书并启动 HTTPS 服务。 -`caBundle` 非空时 webhook 不动它（视为已托管）；不匹配时打印告警。init 镜像默认取自动发现的 device plugin 镜像（downward API），其中已包含 `devinit`，通常无需配置。 +`caBundle` 非空时 webhook 不动它（视为已托管）；不匹配时打印告警。init 镜像默认取自动发现的 device plugin 镜像（downward API），回退到 `busybox:latest`。devinit 二进制从宿主通过 BinDir 挂载，镜像无需包含它。 device-plugin CLI 参数： @@ -281,7 +281,7 @@ device-plugin CLI 参数： | `--webhook-mutating-config` | 待自动管理 `caBundle` 的 `MutatingWebhookConfiguration` 名称。 | | `--webhook-service-name` / `--webhook-service-namespace` | 前置 webhook 的 Service（构成服务证书 DNS SAN）。 | | `--webhook-ca-secret-name` / `--webhook-ca-secret-namespace` | 持久化 CA 的 Secret（留空 = 内存中生成）。 | -| `--webhook-devinit-image` | 注入的 init 容器镜像（默认：自动发现的 device plugin 镜像）。 | +| `--webhook-devinit-image` | 注入的 init 容器镜像（默认：自动发现的 device plugin 镜像，回退到 `busybox:latest`）。二进制从宿主挂载，镜像无需包含 devinit。 | 参见 [Helm chart](./charts/embodied-runtime) 的 `webhook:` 值，提供一键式部署：渲染 webhook Service、`MutatingWebhookConfiguration`（`caBundle` 留空）、所需 RBAC（集群级 `mutatingwebhookconfigurations` 的 get/patch + 命名空间级 `secrets`），并把上述参数全部接进 device-plugin DaemonSet。启用方式： @@ -293,7 +293,7 @@ webhook: caSecret: # 持久化 CA，跨重启复用（推荐） name: devinit-ca namespace: rlark-system # 默认取发布命名空间 - # devinitImage: "" # 默认取 .Values.devicePlugin.image + # devinitImage: "" # 默认取 .Values.devicePlugin.image，回退到 busybox:latest（二进制从宿主挂载） ``` webhook 仅在 `webhook.enabled` 与 `config.hostMacvlans` **同时**设置时渲染；未配置 macvlan 时启用它无效（handler 不注入任何内容）。 diff --git a/apps/embodied-runtime/charts/embodied-runtime/templates/daemonset.yaml b/apps/embodied-runtime/charts/embodied-runtime/templates/daemonset.yaml index 940f0f2..54bc189 100644 --- a/apps/embodied-runtime/charts/embodied-runtime/templates/daemonset.yaml +++ b/apps/embodied-runtime/charts/embodied-runtime/templates/daemonset.yaml @@ -102,6 +102,8 @@ spec: mountPath: /var/lib/kubelet/device-plugins - name: socket-dir mountPath: /var/run/rlark + - name: bin-dir + mountPath: /opt/rlinf/bin securityContext: privileged: true @@ -118,3 +120,7 @@ spec: hostPath: path: /var/run/rlark type: DirectoryOrCreate + - name: bin-dir + hostPath: + path: /opt/rlinf/bin + type: DirectoryOrCreate diff --git a/apps/embodied-runtime/charts/embodied-runtime/values.yaml b/apps/embodied-runtime/charts/embodied-runtime/values.yaml index d355914..6510d32 100644 --- a/apps/embodied-runtime/charts/embodied-runtime/values.yaml +++ b/apps/embodied-runtime/charts/embodied-runtime/values.yaml @@ -48,10 +48,11 @@ webhook: caSecret: name: "" namespace: "" - # devinitImage overrides the injected init container image. It must contain - # the devinit binary at /usr/local/bin/devinit. When empty, the device - # plugin auto-discovers its own image (.Values.devicePlugin.image, via the - # downward API) which ships devinit — so this is rarely needed. + # devinitImage overrides the injected init container image. The devinit + # binary is mounted from the host (/opt/rlinf/bin/devinit) via Allocate, so + # the image does not need to contain it. When empty, the device plugin + # auto-discovers its own image (.Values.devicePlugin.image, via the + # downward API) or falls back to "busybox:latest". devinitImage: "" # device-plugin configuration file. Keys are snake_case to match the binary's diff --git a/apps/embodied-runtime/cmd/device-plugin/main.go b/apps/embodied-runtime/cmd/device-plugin/main.go index 6483a40..6156c7c 100644 --- a/apps/embodied-runtime/cmd/device-plugin/main.go +++ b/apps/embodied-runtime/cmd/device-plugin/main.go @@ -78,7 +78,7 @@ defaults are used. See examples/device-plugin-config.yaml for a template.`, cmd.Flags().StringVar(&wh.CASecretNamespace, "webhook-ca-secret-namespace", "", "Namespace of the CA Secret (required when --webhook-ca-secret-name is set)") cmd.Flags().StringVar(&wh.DevinitImage, "webhook-devinit-image", "", - "Image for the injected init container; must contain devinit at /usr/local/bin/devinit (default: auto-discovered device-plugin image)") + "Image for the injected init container; the binary is mounted from the host, so the image does not need to contain devinit (default: auto-discovered device-plugin image, then busybox:latest)") if err := cmd.Execute(); err != nil { log.Fatalf("[device-plugin] fatal: %v", err) diff --git a/apps/embodied-runtime/docs/examples.md b/apps/embodied-runtime/docs/examples.md index 51c61fa..245d0b5 100644 --- a/apps/embodied-runtime/docs/examples.md +++ b/apps/embodied-runtime/docs/examples.md @@ -361,16 +361,16 @@ ip route curl -k https://172.16.0.2/ # reach the robot directly on its subnet ``` -If you prefer not to use the webhook, author the init container yourself (it just requests the resource and runs `devinit setup`): +If you prefer not to use the webhook, author the init container yourself (it just requests the resource and runs `devinit setup`). The devinit binary is mounted from the host via BinDir, so the image does not need to contain it: ```yaml initContainers: - name: devinit - image: rlinf/embodied-runtime:v0.1.0 - command: ["devinit", "setup"] # reads RLINF_EMBODIED_DEVINIT_SOCKET_PATH + image: busybox:latest + command: ["/opt/rlinf/bin/devinit", "setup"] # reads RLINF_EMBODIED_DEVINIT_SOCKET_PATH resources: requests: - rlinf.io/device: 1 # triggers Allocate → RunDir mount + env vars + rlinf.io/device: 1 # triggers Allocate → RunDir mount + BinDir mount + env vars limits: # required: LimitRanger/ResourceQuota rejects init containers without limits rlinf.io/device: 1 # extended-resource limits must equal requests ``` diff --git a/apps/embodied-runtime/docs/examples.zh-CN.md b/apps/embodied-runtime/docs/examples.zh-CN.md index 8e47857..90b7707 100644 --- a/apps/embodied-runtime/docs/examples.zh-CN.md +++ b/apps/embodied-runtime/docs/examples.zh-CN.md @@ -361,16 +361,16 @@ ip route curl -k https://172.16.0.2/ # 直接在机器人子网上访问机器人 ``` -若不想用 webhook，可自行编写 init 容器（只需申请资源并运行 `devinit setup`）： +若不想用 webhook，可自行编写 init 容器（只需申请资源并运行 `devinit setup`）。devinit 二进制从宿主通过 BinDir 挂载，镜像无需包含它： ```yaml initContainers: - name: devinit - image: rlinf/embodied-runtime:v0.1.0 - command: ["devinit", "setup"] # 读取 RLINF_EMBODIED_DEVINIT_SOCKET_PATH + image: busybox:latest + command: ["/opt/rlinf/bin/devinit", "setup"] # 读取 RLINF_EMBODIED_DEVINIT_SOCKET_PATH resources: requests: - rlinf.io/device: 1 # 触发 Allocate → RunDir 挂载 + 环境变量 + rlinf.io/device: 1 # 触发 Allocate → RunDir 挂载 + BinDir 挂载 + 环境变量 limits: # 必填：LimitRanger/ResourceQuota 会拒绝未设置 limits 的 init 容器 rlinf.io/device: 1 # 扩展资源的 limits 必须等于 requests ``` diff --git a/apps/embodied-runtime/pkg/deviceplugin/plugin.go b/apps/embodied-runtime/pkg/deviceplugin/plugin.go index 1b64e13..ce1e654 100644 --- a/apps/embodied-runtime/pkg/deviceplugin/plugin.go +++ b/apps/embodied-runtime/pkg/deviceplugin/plugin.go @@ -95,6 +95,35 @@ func envOrDefault(key, defaultVal string) string { return defaultVal } +// ensureDevinitBinary copies the devinit CLI binary from the device-plugin +// image (/usr/local/bin/devinit) to the host-mounted BinDir so it can be +// mounted into workload pods via Allocate. The init container then runs the +// binary from the mounted path without needing the device-plugin image. +// Idempotent: if the destination already exists it is silently skipped. +func ensureDevinitBinary() { + src := "/usr/local/bin/devinit" + dst := filepath.Join(BinDir, "devinit") + + if _, err := os.Stat(dst); err == nil { + return + } + + data, err := os.ReadFile(src) + if err != nil { + log.Printf("[device-plugin] WARNING: devinit binary not found at %s — init container will be skipped: %v", src, err) + return + } + if err := os.MkdirAll(BinDir, 0755); err != nil { + log.Printf("[device-plugin] WARNING: create %s: %v — init container will be skipped", BinDir, err) + return + } + if err := os.WriteFile(dst, data, 0755); err != nil { + log.Printf("[device-plugin] WARNING: write %s: %v — init container will be skipped", dst, err) + return + } + log.Printf("[device-plugin] devinit binary copied to %s", dst) +} + // PluginSocketPath returns the full path to the plugin's gRPC socket. func PluginSocketPath() string { return pluginapi.DevicePluginPath + PluginSocketName @@ -412,6 +441,11 @@ func NewPlugin(cfg PluginConfig, whcfg WebhookConfig) *Plugin { // Detect devices (after managers are started). p.devices = p.detectDevices() + // Copy the devinit binary to the host-mounted BinDir so the webhook can + // inject an init container that uses the mounted binary (no image + // dependency). Best-effort: a failure is logged but does not block startup. + ensureDevinitBinary() + return p } diff --git a/apps/embodied-runtime/pkg/deviceplugin/webhook.go b/apps/embodied-runtime/pkg/deviceplugin/webhook.go index 7f757f2..a59c6ed 100644 --- a/apps/embodied-runtime/pkg/deviceplugin/webhook.go +++ b/apps/embodied-runtime/pkg/deviceplugin/webhook.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "log" + "os" admissionv1 "k8s.io/api/admission/v1" corev1 "k8s.io/api/core/v1" @@ -56,9 +57,10 @@ type WebhookConfig struct { CASecretName string CASecretNamespace string - // DevinitImage is the init container image. It must contain the devinit - // binary at devinitBinaryPath. When empty, NewPlugin fills it from the - // auto-discovered device-plugin image (downward API). + // DevinitImage is the init container image. The binary is mounted from + // the host via Allocate, so the image does not need to contain devinit. + // When empty, the auto-discovered device-plugin image is used; falling + // back to "busybox:latest" if auto-discovery is unavailable. DevinitImage string } @@ -110,11 +112,26 @@ const ( // webhook. Reused to detect an existing injection (idempotency). devinitContainerName = "rlark-devinit" - // devinitBinaryPath is the devinit CLI path inside the init container - // image. The device-plugin image ships it here (see the Dockerfile). - devinitBinaryPath = "/usr/local/bin/devinit" + // devinitBinaryPath is the path to the devinit CLI binary, mounted from + // the host via Allocate (BinDir). The device plugin copies the binary + // there at startup; the init container runs it from this mounted path. + devinitBinaryPath = BinDir + "/devinit" + + // defaultDevinitImage is the fallback image for the injected init + // container when no DevinitImage is configured. The binary is mounted + // from the host, so the image does not need to contain it — any image + // with a shell works. + defaultDevinitImage = "busybox:latest" ) +// devinitBinaryExists is a function variable that reports whether the devinit +// binary is present at devinitBinaryPath. Extracted so tests can override it +// without creating the file on the actual host filesystem. +var devinitBinaryExists = func() bool { + _, err := os.Stat(devinitBinaryPath) + return err == nil +} + // --------------------------------------------------------------------------- // Handler // --------------------------------------------------------------------------- @@ -127,8 +144,9 @@ const ( // (called for every container that requests the resource) injects the // RunDir mount (which exposes the devinit service socket) and the // RLINF_EMBODIED_DEVINIT_SOCKET_PATH env var, so the init container can -// locate the socket and run `devinit setup`. No extra volumes or mounts are -// needed from the webhook. The devinit binary itself lives in the image. +// locate the socket and run `devinit setup`. The devinit binary is mounted +// from the host via BinDir (see ensureDevinitBinary in plugin.go), so the +// init container image does not need to contain it. // // The init container runs `devinit setup`, which dials the device plugin's // init service Unix socket; the service reads the caller's PID from the @@ -136,7 +154,7 @@ const ( // network namespace (skipped for hostNetwork pods). type devinitHandler struct { resourceName string // extended resource advertised by this plugin - image string // init container image (contains devinit binary) + image string // init container image (binary is mounted from host) } // newDevinitHandler builds a handler from the plugin's resolved configuration. @@ -230,7 +248,9 @@ func buildDevinitPatch(pod *corev1.Pod, resourceName, image string) ([]byte, err // buildDevinitContainer constructs the init container spec. It requests one // unit of the plugin's extended resource (so Allocate injects the RunDir // socket mount and RLINF_EMBODIED_DEVINIT_SOCKET_PATH env var) and runs the -// devinit CLI, which lives in the image at devinitBinaryPath. +// devinit CLI from the host-mounted BinDir at devinitBinaryPath. The image +// does not need to contain the binary — it is mounted from the host via +// Allocate's BinDir mount. // // The extended resource is mirrored in limits: Kubernetes requires // extended-resource limits to equal their requests, and clusters enforcing a @@ -298,8 +318,12 @@ var _ mutatingwebhook.Handler = (*devinitHandler)(nil) // // The devinit image is resolved in this order: WebhookConfig.DevinitImage // (CLI flag), then the auto-discovered device-plugin image (downward API), -// which is the natural default since the device-plugin image ships the -// devinit binary. +// then the defaultDevinitImage constant. The binary is mounted from the host +// via BinDir, so the image does not need to contain the devinit binary. +// +// If the devinit binary is not found at devinitBinaryPath on the host, the +// webhook is skipped with a warning — the init container would have nothing +// to run. func newWebhookServer(p *Plugin) (*mutatingwebhook.Server, error) { wh := p.webhookCfg if !wh.Enabled { @@ -315,12 +339,22 @@ func newWebhookServer(p *Plugin) (*mutatingwebhook.Server, error) { if wh.MutatingWebhookConfigName == "" { return nil, fmt.Errorf("mutating webhook config name not set") } + + // Check that the devinit binary exists on the host (in BinDir, mounted + // into the device-plugin container). The binary is copied there by + // ensureDevinitBinary in plugin.go. If it is missing, the webhook has + // nothing to inject — skip with a warning. + if !devinitBinaryExists() { + log.Printf("[device-plugin/webhook] WARNING: devinit binary not found at %s — skipping webhook", devinitBinaryPath) + return nil, nil + } + image := wh.DevinitImage if image == "" { image = p.disc.initImage } if image == "" { - return nil, fmt.Errorf("devinit image not set and device-plugin image could not be auto-discovered") + image = defaultDevinitImage } cfg := mutatingwebhook.ServerConfig{ Addr: wh.EffectiveAddr(), diff --git a/apps/embodied-runtime/pkg/deviceplugin/webhook_test.go b/apps/embodied-runtime/pkg/deviceplugin/webhook_test.go index 3ac741c..2b902a8 100644 --- a/apps/embodied-runtime/pkg/deviceplugin/webhook_test.go +++ b/apps/embodied-runtime/pkg/deviceplugin/webhook_test.go @@ -3,6 +3,7 @@ package deviceplugin import ( "context" "encoding/json" + "os" "testing" admissionv1 "k8s.io/api/admission/v1" @@ -109,8 +110,8 @@ func TestBuildDevinitContainer(t *testing.T) { if c.Image != "rlinf/device-plugin:v1" { t.Errorf("Image = %q", c.Image) } - if len(c.Command) != 2 || c.Command[0] != devinitBinaryPath || c.Command[1] != "setup" { - t.Errorf("Command = %v, want [%q setup]", c.Command, devinitBinaryPath) + if len(c.Command) != 2 || c.Command[0] != BinDir+"/devinit" || c.Command[1] != "setup" { + t.Errorf("Command = %v, want [%s setup]", c.Command, BinDir+"/devinit") } got := c.Resources.Requests[corev1.ResourceName(testResource)] if got.Value() != 1 { @@ -369,14 +370,24 @@ func TestNewWebhookServer_NoConfigName(t *testing.T) { } func TestNewWebhookServer_NoImage(t *testing.T) { + devinitBinaryExists = func() bool { return true } + t.Cleanup(func() { devinitBinaryExists = func() bool { _, err := os.Stat(devinitBinaryPath); return err == nil } }) + wh := WebhookConfig{Enabled: true, MutatingWebhookConfigName: "wh", ServiceName: "svc", Namespace: "ns"} p := pluginForWebhookTest(t, wh, true) // no DevinitImage, no discovered initImage - if _, err := newWebhookServer(p); err == nil { - t.Fatal("expected error when no devinit image, got nil") + s, err := newWebhookServer(p) + if err != nil { + t.Fatalf("expected no error when no devinit image (falls back to busybox), got %v", err) + } + if s == nil { + t.Fatal("expected non-nil server when no devinit image (falls back to busybox)") } } func TestNewWebhookServer_OK(t *testing.T) { + devinitBinaryExists = func() bool { return true } + t.Cleanup(func() { devinitBinaryExists = func() bool { _, err := os.Stat(devinitBinaryPath); return err == nil } }) + wh := WebhookConfig{ Enabled: true, MutatingWebhookConfigName: "wh.example.com", @@ -397,6 +408,9 @@ func TestNewWebhookServer_OK(t *testing.T) { // TestNewWebhookServer_AutoDiscoversImage confirms that when DevinitImage is // empty, the discovered device-plugin image is used as the devinit image. func TestNewWebhookServer_AutoDiscoversImage(t *testing.T) { + devinitBinaryExists = func() bool { return true } + t.Cleanup(func() { devinitBinaryExists = func() bool { _, err := os.Stat(devinitBinaryPath); return err == nil } }) + wh := WebhookConfig{ Enabled: true, MutatingWebhookConfigName: "wh.example.com", diff --git a/apps/rlark-ui/README.md b/apps/rlark-ui/README.md index ee68da8..74f0f50 100644 --- a/apps/rlark-ui/README.md +++ b/apps/rlark-ui/README.md @@ -44,7 +44,7 @@ The frontend uses exactly one data mode so pages never mix mock and backend data - Nodes：节点默认按集群和节点名称排序，展示物理位置、GPU/具身设备总量、空闲量与型号；从总览地图点击城市可直接进入对应位置筛选结果 - Cluster detail：集群详情中的节点表与 Nodes 使用相同的位置、资源数量、空闲状态和型号口径 - Node metadata：容量与可分配量由 Agent 从 Kubernetes Node 自动上报；管理员维护的位置、节点分类、GPU 型号和具身设备型号存储在 KCP Node CR，并在 Agent 状态同步时保留。节点详情会同时列出 CPU、内存、GPU 以及所有 `rlinf.io/device*` 端侧设备资源 -- Jobs：列表直接展示并支持复制 Kubernetes `metadata.name` 任务 ID；仅在配置 `rlark.io/display-name` 时补充显示名称，并展示去重节点数、创建时间和停止时间 +- Jobs：列表以 `rlark.io/display-name` 展示名为主，并展示和支持复制系统生成的 Kubernetes `metadata.name` 任务 ID；旧任务未配置展示名时回退到资源 ID - Lists：集群、节点、任务、工作流、存储和 SSH 公钥主列表支持点击表头切换升序与降序；分页基于排序后的完整筛选结果 - Time：创建时间、停止时间等统一转换为中国标准时间（`Asia/Shanghai`），格式为 `YYYY-MM-DD HH:mm:ss` - Overview：核心指标聚焦具身集群数量、具身节点数量、具身设备种类，以及正在运行/全部任务数量 @@ -66,7 +66,7 @@ Admin task management reuses the business-platform list and detail views, but in Admin node management exposes Kubernetes scheduling state directly in the node list and detail view. Cordon prevents new workloads from being scheduled without interrupting running workloads; uncordon restores scheduling. -Job IDs use the Kubernetes resource `metadata.name` and can be copied directly. User-facing timestamps use China Standard Time (`Asia/Shanghai`) in `YYYY-MM-DD HH:mm:ss` format. The overview emphasizes embodied clusters, embodied nodes, unique device models, and running/total jobs. +Job lists primarily show the `rlark.io/display-name` annotation and expose the system-generated Kubernetes `metadata.name` resource ID as copyable secondary information. Legacy Jobs without a display name fall back to the resource ID. User-facing timestamps use China Standard Time (`Asia/Shanghai`) in `YYYY-MM-DD HH:mm:ss` format. The overview emphasizes embodied clusters, embodied nodes, unique device models, and running/total jobs. - API Reference：后端资源 API 演示 diff --git a/apps/rlark-ui/nginx.conf b/apps/rlark-ui/nginx.conf index bf6bb7b..a3923e6 100644 --- a/apps/rlark-ui/nginx.conf +++ b/apps/rlark-ui/nginx.conf @@ -11,6 +11,18 @@ server { client_max_body_size 500m; # 对象存储：允许最大 500MB 的请求体 + # 压缩配置 + gzip on; + gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript; + gzip_min_length 1024; # 单位是字节，约为 1KB[reference:4] + gzip_comp_level 6; + # 启用 Vary: Accept-Encoding 响应头 + gzip_vary on; + # （重要）配置对代理请求的压缩行为 + # 当 Nginx 作为反向代理时，此指令控制是否压缩后端返回的响应[reference:8] + # 下面配置表示：仅当响应包含 no-cache/no-store/private/expired/auth 头时才压缩[reference:9] + gzip_proxied no-cache no-store private expired auth; + location / { try_files $uri $uri/ /index.html; } diff --git a/apps/rlark-ui/package.json b/apps/rlark-ui/package.json index 3aa6782..c032fab 100644 --- a/apps/rlark-ui/package.json +++ b/apps/rlark-ui/package.json @@ -15,7 +15,7 @@ "dev": "vite --host 0.0.0.0", "build": "tsc -b && vite build", "preview": "vite preview --host 0.0.0.0", - "test": "tsc src/utils/terminalKeyboard.ts src/utils/nodeVisibility.ts src/utils/imageReference.ts src/utils/nodeBatchMetadata.ts src/utils/nodeResources.ts src/utils/jobPhase.ts src/utils/resourceAvailability.ts --ignoreConfig --outDir dist/test --target ES2022 --module ES2022 --moduleResolution Bundler --lib ES2022,DOM && node --test tests/*.test.mjs" + "test": "tsc src/utils/terminalKeyboard.ts src/utils/nodeVisibility.ts src/utils/imageReference.ts src/utils/nodeBatchMetadata.ts src/utils/nodeResources.ts src/utils/jobPhase.ts src/utils/job.ts src/utils/crd.ts src/utils/resourceAvailability.ts --ignoreConfig --outDir dist/test --target ES2022 --module ES2022 --moduleResolution Bundler --lib ES2022,DOM && node --test tests/*.test.mjs" }, "dependencies": { "@vitejs/plugin-react": "^4.7.0", diff --git a/apps/rlark-ui/src/App.tsx b/apps/rlark-ui/src/App.tsx index 79e4e81..e26c496 100644 --- a/apps/rlark-ui/src/App.tsx +++ b/apps/rlark-ui/src/App.tsx @@ -373,13 +373,13 @@ export default function App() { setEditJob(null); setRestartAfterEdit(false); }} - onSuccess={(message) => { + onSuccess={(message, jobName) => { setCreateOpen(false); setCloneJob(null); setEditJob(null); setRestartAfterEdit(false); setJobSubmitNotice(message); - navigate("jobs", undefined, { replace: true }); + navigate("jobs", jobName, { replace: true }); }} copy={c} cloneJob={cloneJob} diff --git a/apps/rlark-ui/src/admin/AdminDashboard.tsx b/apps/rlark-ui/src/admin/AdminDashboard.tsx index abba4c2..143b18a 100644 --- a/apps/rlark-ui/src/admin/AdminDashboard.tsx +++ b/apps/rlark-ui/src/admin/AdminDashboard.tsx @@ -256,9 +256,10 @@ export function AdminDashboard({ className="secondary-button" onClick={() => fetchDashboard()} disabled={loading} + aria-busy={loading} > - {c.common.refresh} + {loading ? (zh ? "刷新中..." : "Refreshing...") : c.common.refresh} diff --git a/apps/rlark-ui/src/admin/AdminPage.tsx b/apps/rlark-ui/src/admin/AdminPage.tsx index 762d495..91f005c 100644 --- a/apps/rlark-ui/src/admin/AdminPage.tsx +++ b/apps/rlark-ui/src/admin/AdminPage.tsx @@ -49,6 +49,7 @@ export function ClustersOverviewAdminPage({ copy: c }: { copy: Copy }) { const zh = c.nav.overview === "总览"; const [nodes, setNodes] = useState([]); const [loading, setLoading] = useState(true); + const [refreshing, setRefreshing] = useState(false); const [error, setError] = useState(""); const [selectedClusterNs, setSelectedClusterNs] = useState( null, @@ -72,6 +73,16 @@ export function ClustersOverviewAdminPage({ copy: c }: { copy: Copy }) { useAutoRefresh(fetchNodes, 10000); + const handleRefresh = async () => { + if (refreshing) return; + setRefreshing(true); + try { + await fetchNodes(false); + } finally { + setRefreshing(false); + } + }; + const clustersList = useMemo(() => { const map = new Map(); for (const n of nodes) { @@ -142,9 +153,21 @@ export function ClustersOverviewAdminPage({ copy: c }: { copy: Copy }) { : "View all managed nodes grouped by namespace (cluster)."}

-
@@ -169,9 +192,17 @@ export function ClustersOverviewAdminPage({ copy: c }: { copy: Copy }) { : "View all managed nodes grouped by namespace (cluster)."}

-
@@ -891,6 +922,7 @@ export function AdminPage({ const zh = c.nav.overview === "总览"; const [nodes, setNodes] = useState([]); const [loading, setLoading] = useState(true); + const [refreshing, setRefreshing] = useState(false); const [error, setError] = useState(""); const [editingNode, setEditingNode] = useState(null); const [labelDraft, setLabelDraft] = useState>({}); @@ -933,6 +965,16 @@ export function AdminPage({ useAutoRefresh(fetchNodes, 10000); + const handleRefresh = async () => { + if (refreshing) return; + setRefreshing(true); + try { + await fetchNodes(false); + } finally { + setRefreshing(false); + } + }; + const startEdit = (node: CRDNode) => { const editableLabels = { ...(node.metadata.labels ?? {}) }; NODE_CATEGORY_KEYS.forEach((key) => delete editableLabels[key]); @@ -1402,9 +1444,21 @@ export function AdminPage({ ? `批量设置 (${selectedNodeKeys.size})` : `Batch edit (${selectedNodeKeys.size})`} - @@ -1781,7 +1835,8 @@ export function AdminPage({ fetchNodes()} + onRefresh={handleRefresh} + refreshing={refreshing} onSelectNode={(name) => onNavigate(name)} onToggleScheduling={toggleCordon} updatingNode={cordoningNode} diff --git a/apps/rlark-ui/src/components/NodeResourceBrowser.tsx b/apps/rlark-ui/src/components/NodeResourceBrowser.tsx index ae33434..a475e12 100644 --- a/apps/rlark-ui/src/components/NodeResourceBrowser.tsx +++ b/apps/rlark-ui/src/components/NodeResourceBrowser.tsx @@ -30,6 +30,7 @@ export function NodeResourceBrowser({ copy: c, onSelectNode, onRefresh, + refreshing, onToggleScheduling, updatingNode, initialCategory = "all", @@ -43,7 +44,8 @@ export function NodeResourceBrowser({ nodeWorkloads?: Record; copy: Copy; onSelectNode: (name: string) => void; - onRefresh?: () => void; + onRefresh?: () => void | Promise; + refreshing?: boolean; onToggleScheduling?: (node: CRDNode) => void; updatingNode?: string | null; initialCategory?: CategoryFilter; @@ -235,6 +237,7 @@ export function NodeResourceBrowser({ count={filteredNodes.length} copy={c} onRefresh={onRefresh} + refreshing={refreshing} filterValue={phaseFilter} onFilterChange={(value) => setPhaseFilter(value as "All" | Phase)} filterOptions={[ diff --git a/apps/rlark-ui/src/components/create.tsx b/apps/rlark-ui/src/components/create.tsx index 58a11c2..a1dfd37 100644 --- a/apps/rlark-ui/src/components/create.tsx +++ b/apps/rlark-ui/src/components/create.tsx @@ -265,6 +265,7 @@ export function RoleNameInput({ return ( e.stopPropagation()} onChange={(e) => setDraft(e.target.value)} onBlur={() => { diff --git a/apps/rlark-ui/src/components/shared.tsx b/apps/rlark-ui/src/components/shared.tsx index 620a5ec..bdca027 100644 --- a/apps/rlark-ui/src/components/shared.tsx +++ b/apps/rlark-ui/src/components/shared.tsx @@ -488,6 +488,7 @@ export function PageToolbar({ count, copy: c, onRefresh, + refreshing = false, filterValue, onFilterChange, filterOptions, @@ -497,11 +498,24 @@ export function PageToolbar({ onChange: (value: string) => void; count: number; copy: Copy; - onRefresh?: () => void; + onRefresh?: () => void | Promise; + refreshing?: boolean; filterValue?: string; onFilterChange?: (value: string) => void; filterOptions?: Array<{ value: string; label: string }>; }) { + const [localRefreshing, setLocalRefreshing] = useState(false); + const isRefreshing = refreshing || localRefreshing; + const handleRefresh = async () => { + if (!onRefresh || isRefreshing) return; + setLocalRefreshing(true); + try { + await onRefresh(); + } finally { + setLocalRefreshing(false); + } + }; + return (
@@ -530,9 +544,22 @@ export function PageToolbar({ )} {onRefresh && ( - )} diff --git a/apps/rlark-ui/src/data.ts b/apps/rlark-ui/src/data.ts index c1541c1..1c59ac9 100644 --- a/apps/rlark-ui/src/data.ts +++ b/apps/rlark-ui/src/data.ts @@ -135,8 +135,10 @@ export interface Job { objectStorage: string; mountPath: string; hostPath: string; + pvcSizeGb: number; }>; pvcStorageMap?: Record; + pvcSizeGbMap?: Record; }>; taskStatuses: Array<{ name: string; diff --git a/apps/rlark-ui/src/hooks.ts b/apps/rlark-ui/src/hooks.ts index b207212..b4bab08 100644 --- a/apps/rlark-ui/src/hooks.ts +++ b/apps/rlark-ui/src/hooks.ts @@ -40,7 +40,7 @@ export function useAutoRefresh( fetcher: (isInitial: boolean) => Promise, interval = 10000, deps: unknown[] = [], -): { refresh: () => void } { +): { refresh: () => Promise } { const fetcherRef = useRef(fetcher); const timerRef = useRef(undefined); const mountedRef = useRef(true); diff --git a/apps/rlark-ui/src/mockBackend.ts b/apps/rlark-ui/src/mockBackend.ts index b3ffa5d..5699ff1 100644 --- a/apps/rlark-ui/src/mockBackend.ts +++ b/apps/rlark-ui/src/mockBackend.ts @@ -255,6 +255,14 @@ export function installMockBackend() { } if (method === "GET" && path === "/api/v1/rlinf.io/v1alpha1/nodes") return json({ items: nodes }); + if ( + method === "GET" && + path.startsWith("/api/v1/rlinf.io/v1alpha1/nodes/") + ) { + const name = decodeURIComponent(path.split("/").pop()!); + const node = nodes.find((item) => item.metadata.name === name); + return node ? json(node) : json({ error: "not found" }, 404); + } if ( method === "PATCH" && path.startsWith("/api/v1/rlinf.io/v1alpha1/nodes/") @@ -391,7 +399,7 @@ export function installMockBackend() { const next: StorageClass = { id: name, name, - namespace: payload.namespace || "default", + namespace: "kube-system", provider: payload.provider || "MinIO", clusters: Array.isArray(payload.clusters) ? payload.clusters : [], endpoint: payload.endpoint || "", diff --git a/apps/rlark-ui/src/pages/ClusterManagement.tsx b/apps/rlark-ui/src/pages/ClusterManagement.tsx index efc6bf4..88a83b7 100644 --- a/apps/rlark-ui/src/pages/ClusterManagement.tsx +++ b/apps/rlark-ui/src/pages/ClusterManagement.tsx @@ -227,7 +227,17 @@ export function ClusterManagementPage({ if (isInitial) setLoading(true); let resolvedNodes: CRDNode[] = []; try { - const response = await fetch("/api/v1/rlinf.io/v1alpha1/nodes"); + const nodesURL = new URL( + "/api/v1/rlinf.io/v1alpha1/nodes", + window.location.origin, + ); + if (selectedClusterID) { + nodesURL.searchParams.set( + "labelSelector", + `rlark.io/cluster-id=${selectedClusterID}`, + ); + } + const response = await fetch(nodesURL); if (!response.ok) throw new Error(`HTTP ${response.status}`); const body = await response.json(); resolvedNodes = body.items ?? []; @@ -456,7 +466,7 @@ export function ClusterManagementPage({ fetchClusters()} + onRefresh={() => fetchClusters(false)} onSelectNode={onSelectNode} />
@@ -489,7 +499,7 @@ export function ClusterManagementPage({ onChange={setQuery} count={filteredClusters.length} copy={c} - onRefresh={() => fetchClusters()} + onRefresh={() => fetchClusters(false)} filterValue={phaseFilter} onFilterChange={(value) => setPhaseFilter(value as ClusterPhaseFilter)} filterOptions={[ diff --git a/apps/rlark-ui/src/pages/Clusters.tsx b/apps/rlark-ui/src/pages/Clusters.tsx index 0db5325..0ebecaa 100644 --- a/apps/rlark-ui/src/pages/Clusters.tsx +++ b/apps/rlark-ui/src/pages/Clusters.tsx @@ -104,6 +104,7 @@ export function ClustersPage({ Record >({}); const [loading, setLoading] = useState(true); + const [refreshing, setRefreshing] = useState(false); const [error, setError] = useState(""); const resourceView = initialView ?? "clusters"; const [selectedClusterNs, setSelectedClusterNs] = useState( @@ -175,6 +176,16 @@ export function ClustersPage({ useAutoRefresh(fetchNodes, 10000); + const handleRefresh = async () => { + if (refreshing) return; + setRefreshing(true); + try { + await fetchNodes(false); + } finally { + setRefreshing(false); + } + }; + const workerNodes = useMemo( () => realNodes.filter(isBusinessWorkerNode), [realNodes], @@ -223,9 +234,21 @@ export function ClustersPage({

{c.clusters.title}

{c.clusters.desc}

-
@@ -289,9 +312,21 @@ export function ClustersPage({

{resourceView === "clusters" && ( - )} @@ -365,7 +400,8 @@ export function ClustersPage({ copy={c} initialCategory={initialCategory} initialQuery={initialQuery} - onRefresh={() => fetchNodes()} + onRefresh={handleRefresh} + refreshing={refreshing} onSelectNode={(name) => onNavigate?.(name)} /> diff --git a/apps/rlark-ui/src/pages/CreateJob.tsx b/apps/rlark-ui/src/pages/CreateJob.tsx index e1c4291..6e51cd1 100644 --- a/apps/rlark-ui/src/pages/CreateJob.tsx +++ b/apps/rlark-ui/src/pages/CreateJob.tsx @@ -1,12 +1,45 @@ import { useEffect, useRef, useState } from "react"; + +type ImageUsage = { + image: string; + useCount: number; + lastUsedAt: string; +}; + +function formatImageUsage(usedAt: string, useCount: number, zh: boolean) { + const seconds = Math.max( + 0, + Math.floor((Date.now() - Date.parse(usedAt)) / 1000), + ); + const relative = + seconds < 60 + ? zh + ? "刚刚" + : "just now" + : seconds < 3600 + ? zh + ? `${Math.floor(seconds / 60)} 分钟前` + : `${Math.floor(seconds / 60)}m ago` + : seconds < 86400 + ? zh + ? `${Math.floor(seconds / 3600)} 小时前` + : `${Math.floor(seconds / 3600)}h ago` + : zh + ? `${Math.floor(seconds / 86400)} 天前` + : `${Math.floor(seconds / 86400)}d ago`; + return zh + ? `${relative}使用 · ${useCount} 次` + : `${relative} · used ${useCount} times`; +} import { Check, ChevronDown, Plus, Trash2, X } from "lucide-react"; import { type Cluster, clusters, type Job, type JobType } from "../data"; import type { Copy } from "../i18n"; import type { CRDTask, RoleResource } from "../types"; import { ROLE_TEMPLATES, - computePvcStorageMap, + automaticNetworkDomain, generateJobCRD, + generateJobResourceName, parseNodeSelectorStr, } from "../utils/job"; import { toYaml } from "../utils/yaml"; @@ -141,7 +174,7 @@ export function CreateJobModal({ restartAfterSave = false, }: { onClose: () => void; - onSuccess: (message: string) => void; + onSuccess: (message: string, jobName: string) => void; copy: Copy; cloneJob?: Job | null; editJob?: Job | null; @@ -180,10 +213,13 @@ export function CreateJobModal({ const [jobName, setJobName] = useState( sourceJob ? editJob - ? sourceJob.name - : sourceJob.name + "-copy" + ? sourceJob.displayName + : sourceJob.displayName + "-copy" : "robot-policy-training", ); + const [jobResourceName] = useState(() => + editJob ? editJob.name : generateJobResourceName(), + ); const [headerRole, setHeaderRole] = useState( sourceJob?.headerRole ?? roles[0], ); @@ -193,13 +229,15 @@ export function CreateJobModal({ sourceJob?.command ?? "python train.py --config /mnt/config/train.yaml --dataset /mnt/dataset --output /mnt/checkpoints", ); - const [domain, setDomain] = useState(sourceJob?.domain ?? ""); const [tensorBoardDir, setTensorBoardDir] = useState( sourceJob?.tensorBoardDir ?? "", ); - const [sshPublicKey, setSSHPublicKey] = useState( - sourceJob?.sshPublicKey ?? "", + const [sshPublicKeys, setSSHPublicKeys] = useState(() => + sourceJob?.sshPublicKey + ? sourceJob.sshPublicKey.split("\n").filter(Boolean) + : [], ); + const sshPublicKey = sshPublicKeys.join("\n"); const [sshKeys, setSShKeys] = useState< { index: number; user: string; public_key: string; added_at: string }[] >([]); @@ -221,9 +259,20 @@ export function CreateJobModal({ const [storageClassLoading, setStorageClassLoading] = useState(false); const [storageClassFetched, setStorageClassFetched] = useState(false); const [clustersLoaded, setClustersLoaded] = useState(false); + const [recentImages, setRecentImages] = useState([]); + const [imagePickerRole, setImagePickerRole] = useState(null); const lastFetchedStorageClusterRef = useRef(""); const inferenceDoneRef = useRef(false); + useEffect(() => { + fetch("/api/v1/images") + .then((r) => + r.ok ? r.json() : Promise.reject(new Error(`HTTP ${r.status}`)), + ) + .then((data) => setRecentImages(data.items ?? [])) + .catch(() => {}); + }, []); + useEffect(() => { fetch("/api/v1/rlinf.io/v1alpha1/domains") .then((r) => @@ -371,6 +420,7 @@ export function CreateJobModal({ objectStorage: m.objectStorage ?? "", mountPath: m.mountPath ?? "", hostPath: m.hostPath ?? m.objectStorage ?? "", + pvcSizeGb: m.pvcSizeGb ?? 10, })), }; }); @@ -413,6 +463,18 @@ export function CreateJobModal({ : {}, ); const [activeRoleTab, setActiveRoleTab] = useState(roles[0] ?? ""); + const roleConfigTopRef = useRef(null); + + const selectRole = (role: string, scrollToTop = false) => { + setActiveRoleTab(role); + if (scrollToTop) { + window.requestAnimationFrame(() => { + const modalBody = + roleConfigTopRef.current?.closest(".modal-body"); + if (modalBody) modalBody.scrollTop = 0; + }); + } + }; useEffect(() => { if (availableClusters.length === 0) return; @@ -561,8 +623,9 @@ export function CreateJobModal({ }; const renameRole = (oldName: string, newName: string) => { newName = newName.trim(); - if (!newName || oldName === newName) return; - if (roles.includes(newName)) return; + if (!newName || newName.length > 50 || oldName === newName) return; + if (roles.some((role) => role.toLowerCase() === newName.toLowerCase())) + return; setRoles((prev) => prev.map((r) => (r === oldName ? newName : r))); setRoleResources((prev) => { const rr = prev[oldName]; @@ -621,15 +684,14 @@ export function CreateJobModal({ const updateRRMount = ( role: string, i: number, - field: "objectStorage" | "mountPath" | "type" | "hostPath", - v: string, + field: "objectStorage" | "mountPath" | "type" | "hostPath" | "pvcSizeGb", + v: string | number, ) => { setRoleResources((prev) => { const rr = prev[role]; const next = [...rr.mounts]; next[i] = { ...next[i], [field]: v }; - const pvcStorageMap = computePvcStorageMap(role, next, jobName); - return { ...prev, [role]: { ...rr, mounts: next, pvcStorageMap } }; + return { ...prev, [role]: { ...rr, mounts: next } }; }); }; const addRRMount = (role: string) => { @@ -642,29 +704,30 @@ export function CreateJobModal({ objectStorage: "", mountPath: "", hostPath: "", + pvcSizeGb: 10, }, ]; - const pvcStorageMap = computePvcStorageMap(role, newMounts, jobName); - return { ...prev, [role]: { ...rr, mounts: newMounts, pvcStorageMap } }; + return { ...prev, [role]: { ...rr, mounts: newMounts } }; }); }; const removeRRMount = (role: string, i: number) => { setRoleResources((prev) => { const rr = prev[role]; const newMounts = rr.mounts.filter((_, idx) => idx !== i); - const pvcStorageMap = computePvcStorageMap(role, newMounts, jobName); - return { ...prev, [role]: { ...rr, mounts: newMounts, pvcStorageMap } }; + return { ...prev, [role]: { ...rr, mounts: newMounts } }; }); }; + const automaticDomain = automaticNetworkDomain(domains); const crd = generateJobCRD({ - name: jobName, + name: jobResourceName, + displayName: jobName.trim(), type, headerRole: effectiveHeader, roles, roleResources, runScript, - domain, + domain: automaticDomain, tensorBoardDir, sshPublicKey, }); @@ -677,18 +740,19 @@ export function CreateJobModal({ if (targetStep === 1) { const trimmedName = jobName.trim(); if (!trimmedName) return zh ? "请输入任务名称。" : "Enter a job name."; - if ( - trimmedName.length > 63 || - !/^[a-z0-9]([-a-z0-9]*[a-z0-9])?$/.test(trimmedName) - ) + if (trimmedName.length > 50) return zh - ? "任务名称需为 1-63 位小写字母、数字或连字符，且不能以连字符开头或结尾。" - : "Use 1-63 lowercase letters, numbers, or hyphens; do not start or end with a hyphen."; + ? "任务名称不能超过 50 个字符。" + : "Job name cannot exceed 50 characters."; if (roles.length === 0) return zh ? "至少添加一个角色。" : "Add at least one role."; const normalizedRoles = roles.map((role) => role.trim().toLowerCase()); if (normalizedRoles.some((role) => !role)) return zh ? "角色名称不能为空。" : "Role names cannot be empty."; + if (roles.some((role) => role.trim().length > 50)) + return zh + ? "角色名称不能超过 50 个字符。" + : "Role names cannot exceed 50 characters."; if (new Set(normalizedRoles).size !== normalizedRoles.length) return zh ? "角色名称不能重复。" : "Role names must be unique."; if (!effectiveHeader || !roles.includes(effectiveHeader)) @@ -816,19 +880,37 @@ export function CreateJobModal({ ? `/api/v1/rlinf.io/v1alpha1/jobs/${editJob!.name}` : "/api/v1/rlinf.io/v1alpha1/jobs"; const method = isEdit ? "PUT" : "POST"; - const requestBody = + let requestBody = isEdit && restartAfterSave ? { ...crd, metadata: { ...crd.metadata, annotations: { + ...crd.metadata.annotations, "rlark.io/restarted-at": new Date().toISOString(), }, }, spec: { ...crd.spec, stopped: false }, } : crd; + if (isEdit) { + const currentResp = await fetch(url); + if (!currentResp.ok) throw new Error(`HTTP ${currentResp.status}`); + const current = await currentResp.json(); + requestBody = { + ...requestBody, + metadata: { + ...current.metadata, + ...requestBody.metadata, + name: editJob!.name, + annotations: { + ...current.metadata?.annotations, + ...requestBody.metadata.annotations, + }, + }, + }; + } const resp = await fetch(url, { method, headers: { "Content-Type": "application/json" }, @@ -838,6 +920,7 @@ export function CreateJobModal({ const body = await resp.text(); throw new Error(`HTTP ${resp.status}: ${body}`); } + const savedJob = await resp.json(); onSuccess( isEdit ? restartAfterSave @@ -850,6 +933,7 @@ export function CreateJobModal({ : zh ? "任务提交成功" : "Job submitted successfully", + savedJob.metadata?.name ?? (isEdit ? editJob!.name : jobResourceName), ); } catch (e) { setError(e instanceof Error ? e.message : String(e)); @@ -930,8 +1014,8 @@ export function CreateJobModal({ {zh ? "任务名称" : "Job Name"} setJobName(e.target.value)} - disabled={isEdit} /> + {mount.type === "storage" && ( + + )} diff --git a/apps/rlark-ui/src/pages/Jobs.tsx b/apps/rlark-ui/src/pages/Jobs.tsx index 4452941..8fd7107 100644 --- a/apps/rlark-ui/src/pages/Jobs.tsx +++ b/apps/rlark-ui/src/pages/Jobs.tsx @@ -184,6 +184,8 @@ export function JobsPage({ const [phaseFilter, setPhaseFilter] = useState<"All" | Phase>("All"); const [realJobs, setRealJobs] = useState([]); const [loading, setLoading] = useState(true); + const [listRefreshing, setListRefreshing] = useState(false); + const [copiedJobId, setCopiedJobId] = useState(""); const [error, setError] = useState(""); const [actionNotice, setActionNotice] = useState(""); const [jobAction, setJobAction] = useState< @@ -233,19 +235,35 @@ export function JobsPage({ if (isInitial) setLoading(true); setError(""); try { - const [jobsResp, nodesResp] = await Promise.all([ - fetch("/api/v1/rlinf.io/v1alpha1/jobs"), - fetch("/api/v1/rlinf.io/v1alpha1/nodes"), - ]); + const jobsResp = await fetch("/api/v1/rlinf.io/v1alpha1/jobs"); if (!jobsResp.ok) throw new Error(`HTTP ${jobsResp.status}`); const data = await jobsResp.json(); const items: CRDJob[] = data.items ?? []; setRealJobs(items.map(crdToJob)); + + const nodeNames = new Set(); + for (const job of items) { + for (const task of job.status?.tasks ?? []) { + for (const nodeName of task.observedNodes ?? []) { + if (nodeName) nodeNames.add(nodeName); + } + } + } + // Build nodeName -> pullProgress / nodeName -> events maps. Failures // here are non-fatal: the hover tooltip simply won't appear. - if (nodesResp.ok) { - const nodesData = await nodesResp.json(); - const nodeItems: CRDNode[] = nodesData.items ?? []; + const nodeResponses = await Promise.all( + [...nodeNames].map(async (nodeName) => { + const response = await fetch( + `/api/v1/rlinf.io/v1alpha1/nodes/${encodeURIComponent(nodeName)}`, + ); + return response.ok ? response.json() : null; + }), + ); + { + const nodeItems: CRDNode[] = nodeResponses.filter( + (node): node is CRDNode => node !== null, + ); const progressMap: Record = {}; const eventsMap: Record = {}; const deviceModelMap: Record< @@ -281,6 +299,22 @@ export function JobsPage({ useAutoRefresh(fetchJobs, 10000); + const handleListRefresh = async () => { + if (listRefreshing) return; + setListRefreshing(true); + try { + await fetchJobs(false); + } finally { + setListRefreshing(false); + } + }; + + const handleCopyJobId = async (jobId: string) => { + if (!(await copyText(jobId))) return; + setCopiedJobId(jobId); + window.setTimeout(() => setCopiedJobId(""), 1600); + }; + const handleDelete = async (job: Job) => { setJobAction("delete"); setError(""); @@ -329,14 +363,12 @@ export function JobsPage({ (task) => task.status?.phase === "Stopped", ); if (current.phase === "Stopped" && workersStopped) { - return; + return current; } await new Promise((resolve) => window.setTimeout(resolve, 1000)); } throw new Error( - zh - ? "等待 Worker 停止超时，任务未删除。" - : "Timed out waiting for workers to stop; the job was not deleted.", + zh ? "等待 Worker 停止超时。" : "Timed out waiting for workers to stop.", ); }; @@ -350,15 +382,16 @@ export function JobsPage({ body: JSON.stringify({ spec: { stopped } }), }); if (!resp.ok) throw new Error(`HTTP ${resp.status}`); - if (stopped) await waitForJobWorkersStopped(job); + const stoppedJob = stopped ? await waitForJobWorkersStopped(job) : null; setRealJobs((prev) => prev.map((j) => j.id === job.id - ? { + ? (stoppedJob ?? { ...j, stopped, - phase: (stopped ? "Stopped" : "Pending") as Phase, - } + phase: "Pending" as Phase, + stoppedAt: "—", + }) : j, ), ); @@ -484,7 +517,7 @@ export function JobsPage({ : await handleRestart(job); if (succeeded) { setLifecycleConfirm(null); - if (selectedName) onSelect(undefined); + if (selectedName) onSelect(job.name); } }; @@ -573,7 +606,7 @@ export function JobsPage({ const job = restartTarget; setRestartTarget(null); void handleRestart(job).then((succeeded) => { - if (succeeded) onSelect(undefined); + if (succeeded) onSelect(job.name); }); }} onEditRestart={ @@ -665,7 +698,8 @@ export function JobsPage({ onChange={setQuery} count={filtered.length} copy={c} - onRefresh={() => fetchJobs()} + onRefresh={handleListRefresh} + refreshing={listRefreshing} filterValue={phaseFilter} onFilterChange={(value) => setPhaseFilter(value as "All" | Phase)} filterOptions={[ @@ -788,16 +822,29 @@ export function JobsPage({ return ( - +
+ + +
{c.jobType[job.type]} @@ -1505,6 +1552,12 @@ export function JobDetailPage({ >; }) { const zh = c.nav.overview === "总览"; + const [jobIdCopied, setJobIdCopied] = useState(false); + const handleCopyResourceId = async () => { + if (!(await copyText(job.id))) return; + setJobIdCopied(true); + window.setTimeout(() => setJobIdCopied(false), 1600); + }; const [activeTab, setActiveTab] = useState<"workers" | "logs" | "metrics">( "workers", ); @@ -1533,9 +1586,28 @@ export function JobDetailPage({ logs: string; }> >([]); + const [backendLogs, setBackendLogs] = useState< + Array<{ + timestamp: string; + line: string; + labels?: Record; + fields?: Record; + }> + >([]); const [logsLoading, setLogsLoading] = useState(false); const [logsError, setLogsError] = useState(null); - const [workerRoleFilter, setWorkerRoleFilter] = useState("All"); + const [logsHasMore, setLogsHasMore] = useState(false); + const [logsNextCursor, setLogsNextCursor] = useState(""); + const [logsLoadingMore, setLogsLoadingMore] = useState(false); + // 游标历史栈，用于上一页/下一页翻页。首页为空字符串，第一页查完后 push(nextCursor)。 + const [logsCursorHistory, setLogsCursorHistory] = useState([""]); + const [logsPageIndex, setLogsPageIndex] = useState(0); + // 角色配置区块的角色选择（默认第一个角色） + const [workerRoleFilter, setWorkerRoleFilter] = useState( + job.resources.length > 0 ? job.resources[0].role : "All", + ); + // Worker 列表的角色筛选（独立于角色配置区块，默认 All） + const [workerListRoleFilter, setWorkerListRoleFilter] = useState("All"); const [workerPage, setWorkerPage] = useState(1); const [workerSort, setWorkerSort] = useState<{ key: @@ -1545,6 +1617,8 @@ export function JobDetailPage({ | "node" | "kind" | "ip" + | "domainIP" + | "gpu" | "createdAt" | "phase"; direction: SortDirection; @@ -1558,6 +1632,9 @@ export function JobDetailPage({ const [logWorkerFilter, setLogWorkerFilter] = useState("All"); const [logQuery, setLogQuery] = useState(""); const [logRange, setLogRange] = useState("1h"); + const [logCustomRange, setLogCustomRange] = useState(false); + const [logCustomFrom, setLogCustomFrom] = useState(""); + const [logCustomTo, setLogCustomTo] = useState(""); const [logStreamEnabled, setLogStreamEnabled] = useState(false); // Aggregate Node CR pullProgress for the top StatusBadge hover. Uses Node CR @@ -1625,10 +1702,18 @@ export function JobDetailPage({ }); const workerTaskNamesKey = workerTaskNames.join(","); + const handleWorkerRefresh = async () => { + if (workerRefreshing) return; + setWorkerRefreshing(true); + await refreshTasks(); + setWorkerRefreshKey((key) => key + 1); + }; + useEffect(() => { if (!workerTaskNamesKey) { setPods([]); setDomainIPMap({}); + setWorkerRefreshing(false); return; } let cancelled = false; @@ -1737,69 +1822,180 @@ export function JobDetailPage({ [pendingPodNamesKey], ); - const fetchLogs = async (isInitial = true) => { + const getLogTimeRange = (): { from: string; to: string } => { + if (logCustomRange) { + if (logCustomFrom && logCustomTo) { + // Convert local datetime-local input (no timezone) to UTC ISO string + const fromDate = new Date(logCustomFrom); + const toDate = new Date(logCustomTo); + return { from: fromDate.toISOString(), to: toDate.toISOString() }; + } + // Fallback to 1h if custom range is incomplete + } + const to = new Date(); + const from = new Date(); + switch (logRange) { + case "15m": + from.setMinutes(from.getMinutes() - 15); + break; + case "1h": + from.setHours(from.getHours() - 1); + break; + case "6h": + from.setHours(from.getHours() - 6); + break; + case "24h": + from.setHours(from.getHours() - 24); + break; + case "7d": + from.setDate(from.getDate() - 7); + break; + case "30d": + from.setDate(from.getDate() - 30); + break; + default: + from.setHours(from.getHours() - 1); + } + // Convert to UTC ISO string (new Date() is already local time) + return { from: from.toISOString(), to: to.toISOString() }; + }; + + const fetchLogs = async (isInitial = true, cursor = "") => { if (activeTab !== "logs") return; - if (!isInitial && !logStreamEnabled) return; if (isInitial) setLogsLoading(true); + if (cursor) setLogsLoadingMore(true); setLogsError(null); try { + const params = new URLSearchParams(); + const { from, to } = getLogTimeRange(); + params.set("from", from); + params.set("to", to); + + // Always send the first task as the base filter (backend requires it) + let taskName = ""; + if (logRoleFilter !== "All") { + const resource = job.resources.find((r) => r.role === logRoleFilter); + taskName = resource + ? taskResourceName(job.name, resource.role) + : logRoleFilter; + } else if (logWorkerFilter !== "All") { + const matchedPod = pods.find((p) => p.podName === logWorkerFilter); + if (matchedPod && matchedPod.taskName) { + taskName = matchedPod.taskName; + } + } + + if (!taskName && job.resources.length > 0) { + // Default to the first resource's task if nothing selected + taskName = taskResourceName(job.name, job.resources[0].role); + } + + if (taskName) { + params.set("task", taskName); + } + + if (logWorkerFilter !== "All") { + params.set("pod", logWorkerFilter); + } + + if (logQuery.trim()) { + params.set("query", logQuery.trim()); + } + + if (cursor) { + params.set("cursor", cursor); + } + const resp = await fetch( - `/api/v1/rlinf.io/v1alpha1/jobs/${encodeURIComponent(job.name)}/logs`, + `/api/v1/rlinf.io/v1alpha1/jobs/${encodeURIComponent(job.name)}/logs?${params.toString()}`, ); if (!resp.ok) throw new Error(`HTTP ${resp.status}`); const data = await resp.json(); - setPodLogs(Array.isArray(data.pods) ? data.pods : []); + if (data.source === "backend" && Array.isArray(data.entries)) { + // 分页模式：替换当前页数据，而不是追加 + setBackendLogs(data.entries); + setPodLogs([]); + setLogsHasMore(Boolean(data.hasMore)); + setLogsNextCursor(data.nextCursor || ""); + } else { + setPodLogs(Array.isArray(data.pods) ? data.pods : []); + setBackendLogs([]); + setLogsHasMore(false); + setLogsNextCursor(""); + } } catch (e) { - setPodLogs([]); + if (!cursor) { + setPodLogs([]); + setBackendLogs([]); + } setLogsError(e instanceof Error ? e.message : String(e)); } finally { setLogsLoading(false); + setLogsLoadingMore(false); } }; - useAutoRefresh(fetchLogs, 5000, [activeTab, job.name, logStreamEnabled]); + // 查询条件变化时重置到第一页 + const resetLogsPagination = () => { + setLogsCursorHistory([""]); + setLogsPageIndex(0); + }; - const fallbackWorkers: WorkerItem[] = - job.taskStatuses.length > 0 - ? job.taskStatuses.map((ts, i) => { - const childTaskName = ts.name.toLowerCase().replace(/\s+/g, "-"); - const jobChildName = `${job.name}-${childTaskName}` - .toLowerCase() - .replace(/\s+/g, "-"); - return { - id: `${job.id}-${i}`, - name: ts.name, - jobId: job.id, - role: ts.name, - node: - taskNodes[jobChildName] ?? - taskNodes[ts.name] ?? - ts.observedNodes?.join(", ") ?? - "—", - cluster: taskClusters[jobChildName] ?? taskClusters[ts.name] ?? "—", - phase: (ts.phase || "Pending") as Phase, - cpu: job.resources.find((item) => item.role === ts.name)?.cpu ?? "", - memory: - job.resources.find((item) => item.role === ts.name)?.memory ?? "", - gpu: job.resources.find((item) => item.role === ts.name)?.gpu, - logs: ts.message - ? [ts.message] - : [ - `${ts.name}: worker state synced`, - `${ts.name}: waiting for runtime heartbeat`, - ], - statusMessage: ts.message || undefined, - pullProgress: - pullProgressMap[jobChildName] ?? - pullProgressMap[ts.name.toLowerCase()] ?? - [], - events: - taskEventsMap[jobChildName] ?? - taskEventsMap[ts.name.toLowerCase()] ?? - [], - }; - }) - : []; + const goToNextLogPage = () => { + if (!logsHasMore || !logsNextCursor || logsLoadingMore) return; + const nextHistory = [...logsCursorHistory]; + nextHistory[logsPageIndex + 1] = logsNextCursor; + setLogsCursorHistory(nextHistory); + setLogsPageIndex(logsPageIndex + 1); + fetchLogs(false, logsNextCursor); + }; + + const goToPrevLogPage = () => { + if (logsPageIndex === 0 || logsLoadingMore) return; + const prevIndex = logsPageIndex - 1; + const prevCursor = logsCursorHistory[prevIndex] || ""; + setLogsPageIndex(prevIndex); + fetchLogs(false, prevCursor); + }; + + // 查询条件变化时重置分页到第一页 + useEffect(() => { + resetLogsPagination(); + }, [ + logRange, + logCustomRange, + logCustomFrom, + logCustomTo, + logRoleFilter, + logWorkerFilter, + logQuery, + ]); + + useAutoRefresh(fetchLogs, 5000, [ + activeTab, + job.name, + logStreamEnabled, + logRange, + logCustomRange, + logCustomFrom, + logCustomTo, + logRoleFilter, + logWorkerFilter, + logQuery, + ]); + + const schedulingSummary = ( + message: string | undefined, + events: NodeEventEntry[], + ) => + message === "FailedScheduling" || + events.some((event) => event.reason === "FailedScheduling") + ? zh + ? "没有合适的节点可调度，资源可能被占用" + : "No suitable node is available; resources may be occupied" + : message; + + const fallbackWorkers: WorkerItem[] = []; const resourceForTask = (taskName: string) => job.resources.find( (resource) => taskResourceName(job.name, resource.role) === taskName, @@ -1815,7 +2011,9 @@ export function JobDetailPage({ // hover surfaces live image pull progress for this worker. const nodePullProgress = phase === "Pending" && pod.node - ? (nodePullProgressMap[pod.node] ?? []) + ? (nodePullProgressMap[pod.node] ?? []).filter( + (progress) => progress.image === resource?.image, + ) : []; // 同样从 Node.status.events 取节点 warning 事件；当 pod.node // 缺失时回退到 Task.status.events，让 Pending worker 行 tooltip @@ -1845,7 +2043,7 @@ export function JobDetailPage({ `${role}: worker state synced`, `${role}: waiting for runtime heartbeat`, ], - statusMessage: pod.message || undefined, + statusMessage: schedulingSummary(pod.message, workerEvents), pullProgress: nodePullProgress, events: workerEvents, }; @@ -1854,10 +2052,7 @@ export function JobDetailPage({ const runningWorkerCount = jobWorkers.filter( (worker) => worker.phase === "Running", ).length; - const displayPhase = effectiveJobPhase( - job, - jobWorkers.map((worker) => worker.phase), - ); + const displayPhase = effectiveJobPhase(job); const workerPodsByTask = new Map(); for (const worker of jobWorkers) { workerPodsByTask.set( @@ -1869,9 +2064,12 @@ export function JobDetailPage({ ), ); } - const workerRoles = [...new Set(jobWorkers.map((worker) => worker.role))]; + const workerRoles = [ + ...new Set(job.resources.map((resource) => resource.role)), + ]; const filteredWorkers = jobWorkers.filter( - (worker) => workerRoleFilter === "All" || worker.role === workerRoleFilter, + (worker) => + workerListRoleFilter === "All" || worker.role === workerListRoleFilter, ); const toggleWorkerSort = (key: typeof workerSort.key) => { setWorkerSort((current) => ({ @@ -1899,6 +2097,14 @@ export function JobDetailPage({ return getNodeKindLabel(worker); case "ip": return pod?.ip ?? ""; + case "domainIP": + return worker.id + ? (domainIPMap[ + `${worker.id.split("/")[0]}/${worker.id.split("/")[1]}/${worker.id.split("/")[2]}` + ] ?? "") + : ""; + case "gpu": + return worker.gpu ?? ""; case "createdAt": return formatWorkerCreatedAt(job.startedAt, index); case "phase": @@ -1953,33 +2159,74 @@ export function JobDetailPage({ workerTableDrag.current.active = false; setWorkerTableDragging(false); }; - const logEntries = podLogs.flatMap((pod) => { - const role = - resourceForTask(pod.taskName)?.role ?? - job.resources.find((resource) => resource.role === pod.taskName)?.role ?? - pod.taskName; - return pod.logs - .split("\n") - .filter(Boolean) - .map((message, index) => ({ - id: `${pod.podName}-${index}-${message}`, - worker: pod.podName, - role, - phase: pod.phase, - node: pod.node, - message, - })); - }); - const logRoles = [...new Set(logEntries.map((entry) => entry.role))]; - const logWorkers = [ - ...new Set( - logEntries - .filter( - (entry) => logRoleFilter === "All" || entry.role === logRoleFilter, - ) - .map((entry) => entry.worker), - ), - ]; + const logEntries = + backendLogs.length > 0 + ? backendLogs.map((entry, index) => { + const taskName = entry.labels?.task || entry.fields?.task || ""; + const podName = entry.labels?.pod || entry.fields?.pod || ""; + const node = entry.labels?.node || entry.fields?.node || ""; + const role = + resourceForTask(taskName)?.role ?? + job.resources.find((resource) => resource.role === taskName) + ?.role ?? + taskName; + return { + id: `${podName}-${entry.timestamp}-${index}`, + worker: podName, + role, + phase: "Running", // Backend logs don't have phase, assume running + node, + message: entry.line, + timestamp: entry.timestamp, + }; + }) + : podLogs.flatMap((pod) => { + const role = + resourceForTask(pod.taskName)?.role ?? + job.resources.find((resource) => resource.role === pod.taskName) + ?.role ?? + pod.taskName; + return pod.logs + .split("\n") + .filter(Boolean) + .map((message, index) => ({ + id: `${pod.podName}-${index}-${message}`, + worker: pod.podName, + role, + phase: pod.phase, + node: pod.node, + message, + timestamp: undefined, + })); + }); + // Derive available roles and workers from Job metadata (resources and pods) + // so the dropdowns remain populated and selectable even when log query returns 0 entries. + const logRoles = useMemo(() => { + const rolesFromResources = job.resources.map((r) => r.role); + const rolesFromPods = pods.map((p) => { + const resource = resourceForTask(p.taskName); + return resource?.role ?? p.taskName; + }); + return [...new Set([...rolesFromResources, ...rolesFromPods])].filter( + Boolean, + ); + }, [job.resources, pods]); + + const logWorkers = useMemo(() => { + return [ + ...new Set( + pods + .filter((p) => { + if (logRoleFilter === "All") return true; + const resource = resourceForTask(p.taskName); + const role = resource?.role ?? p.taskName; + return role === logRoleFilter; + }) + .map((p) => p.podName), + ), + ].filter(Boolean); + }, [pods, logRoleFilter]); + const filteredLogEntries = logEntries.filter( (entry) => (logRoleFilter === "All" || entry.role === logRoleFilter) && @@ -1989,7 +2236,7 @@ export function JobDetailPage({ .includes(logQuery.toLowerCase()), ); const tabs: Array<{ id: typeof activeTab; label: string }> = [ - { id: "workers", label: zh ? "Worker" : "Workers" }, + { id: "workers", label: zh ? "详情" : "Details" }, { id: "logs", label: c.common.logs }, { id: "metrics", label: zh ? "监控" : "Metrics" }, ]; @@ -2002,374 +2249,29 @@ export function JobDetailPage({ : undefined; return (
- -
- {tabs.map((tab) => ( - - ))} -
- {activeTab === "workers" && ( -
-
-
- - {zh ? "运行实例" : "Runtime instances"} - -

{zh ? "Worker 列表" : "Worker list"}

-
-
- - {filteredWorkers.length} {zh ? "个 Worker" : "workers"} - - -
-
- { - setWorkerRoleFilter(role); - setWorkerPage(1); - }} - /> -
- - - - {( - [ - ["name", zh ? "实例名称" : "Worker name"], - ["role", zh ? "角色" : "Role"], - ["cluster", zh ? "集群" : "Cluster"], - ["node", zh ? "节点" : "Node"], - ["kind", zh ? "节点类型" : "Node type"], - ["ip", zh ? "实例 IP" : "Worker IP"], - ["createdAt", zh ? "创建时间" : "Created"], - ["phase", zh ? "状态" : "Status"], - ] as const - ).map(([key, label]) => ( - - ))} - - - - {visibleWorkers.map(({ worker, index }) => ( - - ))} - -
- toggleWorkerSort(key)} - /> - -
-
-
- - {zh - ? `第 ${workerPage} / ${workerPageCount} 页` - : `Page ${workerPage} of ${workerPageCount}`} - -
- - -
-
-
- )} - {activeTab === "logs" && ( -
-
-
- {zh ? "任务日志" : "Job logs"} -

{zh ? "Worker 日志流" : "Worker log stream"}

-
-
- {logsLoading ? ( -
- - - -
- - {zh ? "正在连接 Worker 日志" : "Connecting to worker logs"} - - - {zh - ? "正在汇总各实例的最新输出…" - : "Collecting the latest output from each instance…"} - -
-