From 3431c101968c814864584f297a1d2c7833ae9fea Mon Sep 17 00:00:00 2001 From: Zexi Li Date: Wed, 1 Jul 2026 13:11:19 +0800 Subject: [PATCH] feat(llm): support local_path scheduling with prefer_hosts and deployment restart (#25090) --- cmd/climc/shell/llm/llm_deployment.go | 2 + docs/llm/llm-deployment-aiproxy.md | 4 +- pkg/apis/llm/llm_deployment.go | 28 ++- pkg/apis/llm/sku.go | 32 +-- .../llm_container/backend_parameters.go | 10 +- .../local_path_entrypoint_test.go | 46 ++++ pkg/llm/drivers/llm_container/sglang.go | 12 +- pkg/llm/drivers/llm_container/vllm.go | 12 +- pkg/llm/models/llm.go | 19 +- pkg/llm/models/llm_aiproxy_sync.go | 113 +++++++-- pkg/llm/models/llm_aiproxy_sync_test.go | 49 ++-- pkg/llm/models/llm_base.go | 28 +-- pkg/llm/models/llm_deployment.go | 226 ++++++++++++++++-- pkg/llm/models/llm_deployment_gpu.go | 49 ++-- .../llm_deployment_gpu_local_path_test.go | 66 +++++ pkg/llm/models/llm_deployment_restart_test.go | 31 +++ pkg/llm/models/llm_sku.go | 44 +++- pkg/llm/models/llm_sku_local_path.go | 88 +++++++ pkg/llm/models/llm_sku_local_path_test.go | 141 +++++++++++ pkg/llm/models/llm_status_resolver.go | 15 +- pkg/llm/models/llm_status_resolver_test.go | 102 ++++++++ pkg/llm/models/prefer_hosts.go | 141 +++++++++++ pkg/llm/models/prefer_hosts_test.go | 78 ++++++ pkg/llm/tasks/llm/llm_aiproxy_sync_task.go | 53 +++- .../tasks/llm/llm_deployment_create_task.go | 11 +- .../tasks/llm/llm_deployment_restart_task.go | 98 ++++++++ .../llm/llm_deployment_sync_status_task.go | 96 ++++++++ pkg/llm/tasks/llm/llm_sync_status_task.go | 15 +- pkg/mcclient/options/llm/llm_deployment.go | 24 ++ pkg/util/logclient/consts.go | 3 + 30 files changed, 1444 insertions(+), 192 deletions(-) create mode 100644 pkg/llm/drivers/llm_container/local_path_entrypoint_test.go create mode 100644 pkg/llm/models/llm_deployment_gpu_local_path_test.go create mode 100644 pkg/llm/models/llm_deployment_restart_test.go create mode 100644 pkg/llm/models/llm_sku_local_path.go create mode 100644 pkg/llm/models/llm_sku_local_path_test.go create mode 100644 pkg/llm/models/llm_status_resolver_test.go create mode 100644 pkg/llm/models/prefer_hosts.go create mode 100644 pkg/llm/models/prefer_hosts_test.go create mode 100644 pkg/llm/tasks/llm/llm_deployment_restart_task.go create mode 100644 pkg/llm/tasks/llm/llm_deployment_sync_status_task.go diff --git a/cmd/climc/shell/llm/llm_deployment.go b/cmd/climc/shell/llm/llm_deployment.go index f189991ff5..408987b542 100644 --- a/cmd/climc/shell/llm/llm_deployment.go +++ b/cmd/climc/shell/llm/llm_deployment.go @@ -15,4 +15,6 @@ func init() { cmd.Delete(new(options.LLMDeploymentDeleteOptions)) cmd.Perform("register-aiproxy", new(options.LLMDeploymentRegisterAiproxyOptions)) cmd.Perform("unregister-aiproxy", new(options.LLMDeploymentUnregisterAiproxyOptions)) + cmd.Perform("restart", new(options.LLMDeploymentRestartOptions)) + cmd.Perform("syncstatus", new(options.LLMDeploymentSyncstatusOptions)) } diff --git a/docs/llm/llm-deployment-aiproxy.md b/docs/llm/llm-deployment-aiproxy.md index 2f487c584b..a1b48807ca 100644 --- a/docs/llm/llm-deployment-aiproxy.md +++ b/docs/llm/llm-deployment-aiproxy.md @@ -27,9 +27,9 @@ climc llm-deployment-register-aiproxy my-qwen climc llm-deployment-unregister-aiproxy my-qwen ``` -`unregister-aiproxy` 会清理 aiproxy 侧资源，并将 `auto_register_aiproxy` 置为 false，同时清空 `aiproxy_bindings`。deployment `status` 恢复为副本健康态（`ready` / `partial` / `deploying` 等）。 +`unregister-aiproxy` 会清理 aiproxy 侧资源，并将 `auto_register_aiproxy` 置为 false，同时清空 `aiproxy_bindings`，`aiproxy_sync_status` 置为 `disabled`。deployment `status` 保持副本健康态（`ready` / `partial` / `deploying` 等），不再被网关同步覆盖。 -网关同步阶段通过 deployment `status` 表达：`aiproxy_pending`、`aiproxy_syncing`、`aiproxy_partial`、`aiproxy_sync_failed`；全部同步成功后恢复为 `ready` 或 `partial`。 +网关同步进度通过独立字段 `aiproxy_sync_status` 表达：`disabled`、`pending`、`syncing`、`synced`、`partial`、`failed`。 ## 查看对应关系 diff --git a/pkg/apis/llm/llm_deployment.go b/pkg/apis/llm/llm_deployment.go index 4457665242..63b8935147 100644 --- a/pkg/apis/llm/llm_deployment.go +++ b/pkg/apis/llm/llm_deployment.go @@ -85,12 +85,18 @@ const ( LLM_DEPLOYMENT_STATUS_DEPLOYING = "deploying" // Some replicas running but not all (e.g., one died, scale-up in progress). LLM_DEPLOYMENT_STATUS_PARTIAL = "partial" + // Replica reconcile or syncstatus in progress. + LLM_DEPLOYMENT_STATUS_SYNCING = "syncing" +) - // Aiproxy gateway sync phases (stored in deployment status, not a separate column). - LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING = "aiproxy_pending" - LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING = "aiproxy_syncing" - LLM_DEPLOYMENT_STATUS_AIPROXY_PARTIAL = "aiproxy_partial" - LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED = "aiproxy_sync_failed" +// AiproxySyncStatus values stored on SLLMDeployment.AiproxySyncStatus. +const ( + AIPROXY_SYNC_STATUS_DISABLED = "disabled" + AIPROXY_SYNC_STATUS_PENDING = "pending" + AIPROXY_SYNC_STATUS_SYNCING = "syncing" + AIPROXY_SYNC_STATUS_SYNCED = "synced" + AIPROXY_SYNC_STATUS_PARTIAL = "partial" + AIPROXY_SYNC_STATUS_FAILED = "failed" ) // LLMDeploymentCreateInput is the input for creating a new LLMDeployment deployment. @@ -119,6 +125,8 @@ type LLMDeploymentCreateInput struct { AutoStart bool `json:"auto_start"` // Prefer specific host PreferHost string `json:"prefer_host"` + // Prefer specific hosts for local_path scheduling (round-robin per replica). + PreferHosts []string `json:"prefer_hosts,omitempty"` // Host path mounts for instances. HostPaths *HostPaths `json:"host_paths,omitempty"` @@ -174,6 +182,12 @@ type LLMDeploymentUpdateInput struct { AiproxyModelPrefix *string `json:"aiproxy_model_prefix,omitempty"` } +type LLMDeploymentRestartInput struct { +} + +type LLMDeploymentSyncstatusInput struct { +} + // Model source types const ( LLM_MODEL_SOURCE_HUGGINGFACE = "huggingface" @@ -207,7 +221,8 @@ type LLMDeploymentDetails struct { ModelScopeModelId string `json:"model_scope_model_id"` ModelScopeFilePath string `json:"model_scope_file_path"` LocalPath string `json:"local_path"` - Categories string `json:"categories"` + PreferHosts []string `json:"prefer_hosts,omitempty"` + Categories []string `json:"categories,omitempty"` Backend string `json:"backend"` BackendVersion string `json:"backend_version"` Replicas int `json:"replicas"` @@ -229,6 +244,7 @@ type LLMDeploymentDetails struct { AutoRegisterAiproxy bool `json:"auto_register_aiproxy"` AiproxyModelPrefix string `json:"aiproxy_model_prefix"` AiproxyRoutingId string `json:"aiproxy_routing_id"` + AiproxySyncStatus string `json:"aiproxy_sync_status"` } // Per-replica aiproxy binding sync status (AiproxyInstanceBinding.sync_status). diff --git a/pkg/apis/llm/sku.go b/pkg/apis/llm/sku.go index 25432a1969..e5acd49d4a 100644 --- a/pkg/apis/llm/sku.go +++ b/pkg/apis/llm/sku.go @@ -127,17 +127,18 @@ type LLMSkuDetails struct { LLMSpec *LLMSpec `json:"llm_spec,omitempty"` // Model source - Source string `json:"source"` - HuggingfaceRepoId string `json:"huggingface_repo_id"` - HuggingfaceFilename string `json:"huggingface_filename"` - ModelScopeModelId string `json:"model_scope_model_id"` - ModelScopeFilePath string `json:"model_scope_file_path"` - LocalPath string `json:"local_path"` + Source string `json:"source"` + HuggingfaceRepoId string `json:"huggingface_repo_id"` + HuggingfaceFilename string `json:"huggingface_filename"` + ModelScopeModelId string `json:"model_scope_model_id"` + ModelScopeFilePath string `json:"model_scope_file_path"` + LocalPath string `json:"local_path"` + PreferHosts []string `json:"prefer_hosts,omitempty"` // Model categories - Categories string `json:"categories"` + Categories []string `json:"categories,omitempty"` // Inference backend version and parameters - BackendVersion string `json:"backend_version"` - BackendParameters string `json:"backend_parameters"` + BackendVersion string `json:"backend_version"` + BackendParameters []string `json:"backend_parameters,omitempty"` } type MountedAppResourceDetails struct { @@ -212,12 +213,13 @@ type LLMSkuCreateInput struct { LLMSpec *LLMSpec `json:"llm_spec,omitempty"` // Model source - Source string `json:"source"` - HuggingfaceRepoId string `json:"huggingface_repo_id"` - HuggingfaceFilename string `json:"huggingface_filename"` - ModelScopeModelId string `json:"model_scope_model_id"` - ModelScopeFilePath string `json:"model_scope_file_path"` - LocalPath string `json:"local_path"` + Source string `json:"source"` + HuggingfaceRepoId string `json:"huggingface_repo_id"` + HuggingfaceFilename string `json:"huggingface_filename"` + ModelScopeModelId string `json:"model_scope_model_id"` + ModelScopeFilePath string `json:"model_scope_file_path"` + LocalPath string `json:"local_path"` + PreferHosts []string `json:"prefer_hosts,omitempty"` // Model categories Categories []string `json:"categories"` // Inference backend version and parameters diff --git a/pkg/llm/drivers/llm_container/backend_parameters.go b/pkg/llm/drivers/llm_container/backend_parameters.go index bb8ead99ae..667640ec1c 100644 --- a/pkg/llm/drivers/llm_container/backend_parameters.go +++ b/pkg/llm/drivers/llm_container/backend_parameters.go @@ -1,7 +1,6 @@ package llm_container import ( - "encoding/json" "fmt" "strings" ) @@ -11,15 +10,10 @@ type runtimeArg struct { Value string } -func parseBackendParameterArgs(raw string, validate func(string) error) ([]runtimeArg, error) { - raw = strings.TrimSpace(raw) - if raw == "" { +func parseBackendParameterArgs(items []string, validate func(string) error) ([]runtimeArg, error) { + if len(items) == 0 { return nil, nil } - items := []string{} - if err := json.Unmarshal([]byte(raw), &items); err != nil { - items = []string{raw} - } args := make([]runtimeArg, 0, len(items)) errs := make([]string, 0) for _, item := range items { diff --git a/pkg/llm/drivers/llm_container/local_path_entrypoint_test.go b/pkg/llm/drivers/llm_container/local_path_entrypoint_test.go new file mode 100644 index 0000000000..e964076cd7 --- /dev/null +++ b/pkg/llm/drivers/llm_container/local_path_entrypoint_test.go @@ -0,0 +1,46 @@ +package llm_container + +import ( + "strings" + "testing" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/llm/models" +) + +func TestBuildVLLMEntrypointScriptMountedModelsFlag(t *testing.T) { + sleepScript := buildVLLMEntrypointScript(false, 1, nil, nil) + if !strings.Contains(sleepScript, "sleep infinity") { + t.Fatalf("expected idle script without mounted models, got %q", sleepScript) + } + serveScript := buildVLLMEntrypointScript(true, 1, nil, &api.LLMSpecVllm{PreferredModel: "Qwen3-8B"}) + if strings.Contains(serveScript, "sleep infinity") { + t.Fatalf("expected serve script with mounted models, got %q", serveScript) + } + if !strings.Contains(serveScript, api.LLM_VLLM_EXEC_PATH) { + t.Fatalf("expected vllm exec in serve script, got %q", serveScript) + } +} + +func TestLocalPathSkuEnablesServeEntrypoint(t *testing.T) { + hostPaths := api.HostPaths{ + { + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: api.ContainerHostPathRelations{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B"}, + }, + }, + } + sku := &models.SLLMSku{ + LLMType: string(api.LLM_CONTAINER_VLLM), + Source: api.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + } + sku.HostPaths = &hostPaths + postOverlaysLen := 0 + hasMountedModels := postOverlaysLen > 0 || models.SkuHasLocalHostPathModel(sku) + if !hasMountedModels { + t.Fatal("expected local_path sku to count as having mounted models") + } +} diff --git a/pkg/llm/drivers/llm_container/sglang.go b/pkg/llm/drivers/llm_container/sglang.go index 4539e2d4be..4d12c43453 100644 --- a/pkg/llm/drivers/llm_container/sglang.go +++ b/pkg/llm/drivers/llm_container/sglang.go @@ -222,11 +222,11 @@ func (s *sglang) GetContainerSpec(ctx context.Context, llm *models.SLLM, image * if eff := s.GetEffectiveSpec(llm, sku); eff != nil { effSpec = eff.(*api.LLMSpecSGLang) } - backendParameters := "" + var backendParameters []string if sku != nil { backendParameters = sku.BackendParameters } - hasMountedModels := len(postOverlays) > 0 + hasMountedModels := len(postOverlays) > 0 || models.SkuHasLocalHostPathModel(sku) startScript := buildSGLangEntrypointScript(hasMountedModels, tensorParallelSize, backendParameters, effSpec) envs := []*commonapi.ContainerKeyValue{ { @@ -637,7 +637,7 @@ func sglangCustomizedArgsToRuntime(args []*api.SGLangCustomizedArg) []runtimeArg return out } -func appendSGLangRuntimeFlags(flags []string, backendParameters string, effSpec *api.LLMSpecSGLang) []string { +func appendSGLangRuntimeFlags(flags []string, backendParameters []string, effSpec *api.LLMSpecSGLang) []string { backendArgs, err := parseBackendParameterArgs(backendParameters, validateSGLangArgKey) if err != nil { log.Errorf("parse sglang backend parameters: %v", err) @@ -654,7 +654,7 @@ func appendSGLangRuntimeFlags(flags []string, backendParameters string, effSpec return appendRuntimeFlags(flags, mergedArgs) } -func buildSGLangServeFlagsWithModelExpr(modelExpr string, servedModelNameExpr string, tensorParallelSize int, backendParameters string, effSpec *api.LLMSpecSGLang) []string { +func buildSGLangServeFlagsWithModelExpr(modelExpr string, servedModelNameExpr string, tensorParallelSize int, backendParameters []string, effSpec *api.LLMSpecSGLang) []string { flags := []string{ fmt.Sprintf("--model-path %s", modelExpr), fmt.Sprintf("--served-model-name %s", servedModelNameExpr), @@ -665,7 +665,7 @@ func buildSGLangServeFlagsWithModelExpr(modelExpr string, servedModelNameExpr st return appendSGLangRuntimeFlags(flags, backendParameters, effSpec) } -func buildSGLangServeFlags(modelPath string, tensorParallelSize int, backendParameters string, effSpec *api.LLMSpecSGLang) []string { +func buildSGLangServeFlags(modelPath string, tensorParallelSize int, backendParameters []string, effSpec *api.LLMSpecSGLang) []string { modelQuoted := shellQuoteSingle(modelPath) return buildSGLangServeFlagsWithModelExpr( modelQuoted, @@ -676,7 +676,7 @@ func buildSGLangServeFlags(modelPath string, tensorParallelSize int, backendPara ) } -func buildSGLangEntrypointScript(hasMountedModels bool, tensorParallelSize int, backendParameters string, effSpec *api.LLMSpecSGLang) string { +func buildSGLangEntrypointScript(hasMountedModels bool, tensorParallelSize int, backendParameters []string, effSpec *api.LLMSpecSGLang) string { modelsPath := shellQuoteSingle(api.LLM_SGLANG_MODELS_PATH) if !hasMountedModels { return fmt.Sprintf("mkdir -p %s && exec sleep infinity", modelsPath) diff --git a/pkg/llm/drivers/llm_container/vllm.go b/pkg/llm/drivers/llm_container/vllm.go index 6d7a5f741a..24b49af9af 100644 --- a/pkg/llm/drivers/llm_container/vllm.go +++ b/pkg/llm/drivers/llm_container/vllm.go @@ -164,7 +164,7 @@ func vllmCustomizedArgsToRuntime(args []*api.VllmCustomizedArg) []runtimeArg { return out } -func appendVLLMRuntimeFlags(flags []string, backendParameters string, effSpec *api.LLMSpecVllm) []string { +func appendVLLMRuntimeFlags(flags []string, backendParameters []string, effSpec *api.LLMSpecVllm) []string { backendArgs, err := parseBackendParameterArgs(backendParameters, validateVLLMArgKey) if err != nil { log.Errorf("parse vllm backend parameters: %v", err) @@ -181,7 +181,7 @@ func appendVLLMRuntimeFlags(flags []string, backendParameters string, effSpec *a return appendRuntimeFlags(flags, mergedArgs) } -func buildVLLMServeFlagsWithModelExpr(modelExpr string, servedModelNameExpr string, tensorParallelSize int, backendParameters string, effSpec *api.LLMSpecVllm) []string { +func buildVLLMServeFlagsWithModelExpr(modelExpr string, servedModelNameExpr string, tensorParallelSize int, backendParameters []string, effSpec *api.LLMSpecVllm) []string { flags := []string{ fmt.Sprintf("--model %s", modelExpr), fmt.Sprintf("--served-model-name %s", servedModelNameExpr), @@ -191,7 +191,7 @@ func buildVLLMServeFlagsWithModelExpr(modelExpr string, servedModelNameExpr stri return appendVLLMRuntimeFlags(flags, backendParameters, effSpec) } -func buildVLLMServeFlags(modelPath string, tensorParallelSize int, backendParameters string, effSpec *api.LLMSpecVllm) []string { +func buildVLLMServeFlags(modelPath string, tensorParallelSize int, backendParameters []string, effSpec *api.LLMSpecVllm) []string { modelQuoted := shellQuoteSingle(modelPath) return buildVLLMServeFlagsWithModelExpr( modelQuoted, @@ -202,7 +202,7 @@ func buildVLLMServeFlags(modelPath string, tensorParallelSize int, backendParame ) } -func buildVLLMEntrypointScript(hasMountedModels bool, tensorParallelSize int, backendParameters string, effSpec *api.LLMSpecVllm) string { +func buildVLLMEntrypointScript(hasMountedModels bool, tensorParallelSize int, backendParameters []string, effSpec *api.LLMSpecVllm) string { modelsPath := shellQuoteSingle(api.LLM_VLLM_MODELS_PATH) if !hasMountedModels { return fmt.Sprintf("mkdir -p %s && exec sleep infinity", modelsPath) @@ -430,11 +430,11 @@ func (v *vllm) GetContainerSpec(ctx context.Context, llm *models.SLLM, image *mo if eff := v.GetEffectiveSpec(llm, sku); eff != nil { effSpec = eff.(*api.LLMSpecVllm) } - backendParameters := "" + var backendParameters []string if sku != nil { backendParameters = sku.BackendParameters } - hasMountedModels := len(postOverlays) > 0 + hasMountedModels := len(postOverlays) > 0 || models.SkuHasLocalHostPathModel(sku) startScript := buildVLLMEntrypointScript(hasMountedModels, tensorParallelSize, backendParameters, effSpec) envs := []*commonapi.ContainerKeyValue{ { diff --git a/pkg/llm/models/llm.go b/pkg/llm/models/llm.go index 6b084cc988..b62fefac1a 100644 --- a/pkg/llm/models/llm.go +++ b/pkg/llm/models/llm.go @@ -138,12 +138,20 @@ func skuFromLLMSkuCreateInput(input *api.LLMSkuCreateInput) *SLLMSku { if input == nil { return nil } - return &SLLMSku{ - LLMType: input.LLMType, + sku := &SLLMSku{ + LLMType: input.LLMType, + Source: input.Source, + LocalPath: input.LocalPath, SLLMSkuBase: SLLMSkuBase{ - Devices: input.Devices, + Devices: input.Devices, + VramClaimMb: input.VramClaimMb, + HostPaths: input.HostPaths, }, } + if len(input.PreferHosts) > 0 { + sku.PreferHosts = append([]string(nil), input.PreferHosts...) + } + return sku } // ValidateRequireMountedModels errors if neither input nor existing llm nor sku supplies mounted_models. For create, pass nil/empty for llmCurMountedModels. @@ -153,6 +161,9 @@ func ValidateRequireMountedModels( llmCurMountedModels []string, sku *SLLMSku, ) error { + if sku != nil && SkuHasLocalHostPathModel(sku) { + return nil + } effectiveModels := llmCurMountedModels if inputMountedModels != nil { effectiveModels = inputMountedModels @@ -715,7 +726,7 @@ func (llm *SLLM) ServerCreate(ctx context.Context, userCred mcclient.TokenCreden if nil != err { return "", errors.Wrap(err, "GetPodCreateInput") } - log.Infoln("PodCreateInput Data: ", jsonutils.Marshal(data).String()) + log.Infof("PodCreateInput Data: %s", jsonutils.Marshal(data).String()) resp, err := compute.Servers.Create(s, jsonutils.Marshal(data)) if nil != err { diff --git a/pkg/llm/models/llm_aiproxy_sync.go b/pkg/llm/models/llm_aiproxy_sync.go index 06af294a24..4388a3f986 100644 --- a/pkg/llm/models/llm_aiproxy_sync.go +++ b/pkg/llm/models/llm_aiproxy_sync.go @@ -604,38 +604,100 @@ func computeAiproxyBindingSyncResult(bindings []api.AiproxyInstanceBinding, runn return aiproxyBindingSyncSynced } -func resolveDeploymentStatusAfterAiproxySync(dep *SLLMDeployment, result aiproxyBindingSyncResult) string { +func resolveAiproxySyncStatusAfterReconcile(result aiproxyBindingSyncResult) string { switch result { case aiproxyBindingSyncSynced: - if dep.Replicas > 0 && dep.ReadyReplicas >= dep.Replicas { - return api.STATUS_READY - } - if dep.ReadyReplicas > 0 { - return api.LLM_DEPLOYMENT_STATUS_PARTIAL - } - return api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING + return api.AIPROXY_SYNC_STATUS_SYNCED case aiproxyBindingSyncPartial: - return api.LLM_DEPLOYMENT_STATUS_AIPROXY_PARTIAL + return api.AIPROXY_SYNC_STATUS_PARTIAL case aiproxyBindingSyncFailed: - return api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED + return api.AIPROXY_SYNC_STATUS_FAILED default: - return api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING + return api.AIPROXY_SYNC_STATUS_PENDING } } -func deploymentStatusMessageAfterAiproxySync(dep *SLLMDeployment, result aiproxyBindingSyncResult) string { +func aiproxySyncStatusMessage(dep *SLLMDeployment, result aiproxyBindingSyncResult) string { switch result { case aiproxyBindingSyncSynced: return fmt.Sprintf("aiproxy synced, ready_replicas=%d/%d", dep.ReadyReplicas, dep.Replicas) case aiproxyBindingSyncPartial: - return fmt.Sprintf("aiproxy partially synced, ready_replicas=%d/%d", dep.ReadyReplicas, dep.Replicas) + msg := fmt.Sprintf("aiproxy partially synced, ready_replicas=%d/%d", dep.ReadyReplicas, dep.Replicas) + if reason := AiproxySyncFailureReason(dep); reason != "" { + return msg + ": " + reason + } + return msg case aiproxyBindingSyncFailed: + if reason := AiproxySyncFailureReason(dep); reason != "" { + return reason + } return "aiproxy sync failed" default: return "waiting for running replicas" } } +// AiproxySyncFailureReason summarizes per-replica binding errors for logs and UI. +func AiproxySyncFailureReason(dep *SLLMDeployment) string { + if dep == nil { + return "" + } + bindings := deploymentAiproxyBindings(dep) + if len(bindings) == 0 { + return "" + } + parts := make([]string, 0, len(bindings)) + for i := range bindings { + b := bindings[i] + if b.SyncStatus != api.AIPROXY_BINDING_SYNC_FAILED { + continue + } + errMsg := strings.TrimSpace(b.LastError) + if errMsg == "" { + errMsg = "unknown error" + } + if b.LlmId != "" { + parts = append(parts, fmt.Sprintf("llm %s: %s", b.LlmId, errMsg)) + } else { + parts = append(parts, errMsg) + } + } + return strings.Join(parts, "; ") +} + +func (dep *SLLMDeployment) SetAiproxySyncStatus(ctx context.Context, userCred mcclient.TokenCredential, status, msg string) error { + if dep == nil { + return errors.Wrap(httperrors.ErrInvalidStatus, "nil deployment") + } + if status == "" { + status = api.AIPROXY_SYNC_STATUS_DISABLED + } + if dep.AiproxySyncStatus == status { + return nil + } + if _, err := db.Update(dep, func() error { + dep.AiproxySyncStatus = status + return nil + }); err != nil { + return errors.Wrap(err, "update aiproxy_sync_status") + } + if msg != "" { + db.OpsLog.LogEvent(dep, "aiproxy_sync", msg, userCred) + } + return nil +} + +func inferAiproxySyncStatus(dep *SLLMDeployment) string { + if dep == nil || !dep.AutoRegisterAiproxy { + return api.AIPROXY_SYNC_STATUS_DISABLED + } + bindings := deploymentAiproxyBindings(dep) + if len(bindings) == 0 { + return api.AIPROXY_SYNC_STATUS_PENDING + } + return resolveAiproxySyncStatusAfterReconcile(computeAiproxyBindingSyncResult(bindings, dep.ReadyReplicas)) +} + // ReconcileDeploymentAiproxy syncs all running replicas and refreshes routing bindings. func ReconcileDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCredential, dep *SLLMDeployment) error { if dep == nil { @@ -645,7 +707,7 @@ func ReconcileDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCred return nil } - if err := dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING, "aiproxy sync in progress"); err != nil { + if err := dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_SYNCING, "aiproxy sync in progress"); err != nil { return errors.Wrap(err, "set aiproxy syncing status") } @@ -658,7 +720,7 @@ func ReconcileDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCred if err := persistDeploymentAiproxyBindings(dep, dep.AiproxyRoutingId, nil); err != nil { return err } - return dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING, "waiting for running replicas") + return dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_PENDING, "waiting for running replicas") } session := aiproxyAdminSession(ctx) @@ -667,19 +729,19 @@ func ReconcileDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCred items, bindings, err := buildRoutingModelItems(ctx, userCred, dep, llms) if err != nil { _ = persistDeploymentAiproxyBindings(dep, dep.AiproxyRoutingId, bindings) - _ = dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + _ = dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_FAILED, err.Error()) return err } primaryModelKey := primaryUpstreamModelKeyFromBindings(ctx, userCred, bindings) routingId, err := ensureAiRouting(session, routingName, dep, deploymentRoutingModelKey(dep, primaryModelKey)) if err != nil { _ = persistDeploymentAiproxyBindings(dep, dep.AiproxyRoutingId, bindings) - _ = dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + _ = dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_FAILED, err.Error()) return err } if err := applyRoutingModels(session, routingId, items); err != nil { _ = persistDeploymentAiproxyBindings(dep, routingId, bindings) - _ = dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) + _ = dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_FAILED, err.Error()) return err } @@ -687,8 +749,8 @@ func ReconcileDeploymentAiproxy(ctx context.Context, userCred mcclient.TokenCred if err := persistDeploymentAiproxyBindings(dep, routingId, bindings); err != nil { return err } - status := resolveDeploymentStatusAfterAiproxySync(dep, result) - return dep.SetStatus(ctx, userCred, status, deploymentStatusMessageAfterAiproxySync(dep, result)) + syncStatus := resolveAiproxySyncStatusAfterReconcile(result) + return dep.SetAiproxySyncStatus(ctx, userCred, syncStatus, aiproxySyncStatusMessage(dep, result)) } func deleteAiProviderById(session *mcclient.ClientSession, providerId string) error { @@ -812,7 +874,7 @@ func UnsyncLlmInstance(ctx context.Context, userCred mcclient.TokenCredential, d if err := persistDeploymentAiproxyBindings(dep, "", nil); err != nil { return err } - return dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING, "waiting for running replicas") + return dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_PENDING, "waiting for running replicas") } if len(remaining) == 0 { @@ -822,7 +884,7 @@ func UnsyncLlmInstance(ctx context.Context, userCred mcclient.TokenCredential, d if err := persistDeploymentAiproxyBindings(dep, "", nil); err != nil { return err } - return dep.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_PENDING, "waiting for running replicas") + return dep.SetAiproxySyncStatus(ctx, userCred, api.AIPROXY_SYNC_STATUS_PENDING, "waiting for running replicas") } items := make([]apapi.AiRoutingModelItem, 0, len(remaining)) @@ -870,8 +932,8 @@ func UnsyncLlmInstance(ctx context.Context, userCred mcclient.TokenCredential, d if err := persistDeploymentAiproxyBindings(dep, routingId, bindings); err != nil { return err } - status := resolveDeploymentStatusAfterAiproxySync(dep, result) - return dep.SetStatus(ctx, userCred, status, deploymentStatusMessageAfterAiproxySync(dep, result)) + syncStatus := resolveAiproxySyncStatusAfterReconcile(result) + return dep.SetAiproxySyncStatus(ctx, userCred, syncStatus, aiproxySyncStatusMessage(dep, result)) } func parseBindingForLlm(dep *SLLMDeployment, llmId string) (api.AiproxyInstanceBinding, error) { @@ -909,10 +971,11 @@ func clearDeploymentAiproxyRegistrationState(dep *SLLMDeployment) { dep.AutoRegisterAiproxy = false dep.AiproxyRoutingId = "" dep.AiproxyBindings = nil + dep.AiproxySyncStatus = api.AIPROXY_SYNC_STATUS_DISABLED } func (dep *SLLMDeployment) StartAiproxySyncTask(ctx context.Context, userCred mcclient.TokenCredential, llmId, parentTaskId string) error { - if dep.Status == api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING && parentTaskId == "" { + if dep.AiproxySyncStatus == api.AIPROXY_SYNC_STATUS_SYNCING && parentTaskId == "" { return nil } params := jsonutils.NewDict() diff --git a/pkg/llm/models/llm_aiproxy_sync_test.go b/pkg/llm/models/llm_aiproxy_sync_test.go index b95dfe66bc..6aabf57cc6 100644 --- a/pkg/llm/models/llm_aiproxy_sync_test.go +++ b/pkg/llm/models/llm_aiproxy_sync_test.go @@ -89,6 +89,7 @@ func TestClearDeploymentAiproxyRegistrationState(t *testing.T) { dep.AutoRegisterAiproxy = true dep.AiproxyRoutingId = "routing-1" dep.AiproxyBindings = &api.AiproxyBindings{{LlmId: "llm-1"}} + dep.AiproxySyncStatus = api.AIPROXY_SYNC_STATUS_SYNCED clearDeploymentAiproxyRegistrationState(dep) @@ -101,48 +102,64 @@ func TestClearDeploymentAiproxyRegistrationState(t *testing.T) { if dep.AiproxyBindings != nil { t.Fatal("AiproxyBindings should be nil") } + if dep.AiproxySyncStatus != api.AIPROXY_SYNC_STATUS_DISABLED { + t.Fatalf("AiproxySyncStatus should be disabled, got %q", dep.AiproxySyncStatus) + } } -func TestResolveDeploymentStatusAfterAiproxySync(t *testing.T) { +func TestResolveAiproxySyncStatusAfterReconcile(t *testing.T) { cases := []struct { name string - dep SLLMDeployment result aiproxyBindingSyncResult wantStat string }{ { - name: "fully synced all replicas", - dep: SLLMDeployment{Replicas: 2, ReadyReplicas: 2}, + name: "fully synced", result: aiproxyBindingSyncSynced, - wantStat: api.STATUS_READY, - }, - { - name: "synced partial replicas", - dep: SLLMDeployment{Replicas: 2, ReadyReplicas: 1}, - result: aiproxyBindingSyncSynced, - wantStat: api.LLM_DEPLOYMENT_STATUS_PARTIAL, + wantStat: api.AIPROXY_SYNC_STATUS_SYNCED, }, { name: "binding partial failure", - dep: SLLMDeployment{Replicas: 2, ReadyReplicas: 2}, result: aiproxyBindingSyncPartial, - wantStat: api.LLM_DEPLOYMENT_STATUS_AIPROXY_PARTIAL, + wantStat: api.AIPROXY_SYNC_STATUS_PARTIAL, }, { name: "all bindings failed", - dep: SLLMDeployment{Replicas: 1, ReadyReplicas: 1}, result: aiproxyBindingSyncFailed, - wantStat: api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, + wantStat: api.AIPROXY_SYNC_STATUS_FAILED, + }, + { + name: "pending", + result: aiproxyBindingSyncPending, + wantStat: api.AIPROXY_SYNC_STATUS_PENDING, }, } for _, c := range cases { - got := resolveDeploymentStatusAfterAiproxySync(&c.dep, c.result) + got := resolveAiproxySyncStatusAfterReconcile(c.result) if got != c.wantStat { t.Fatalf("%s: got %q want %q", c.name, got, c.wantStat) } } } +func TestAiproxySyncFailureReason(t *testing.T) { + dep := &SLLMDeployment{} + dep.AiproxyBindings = &api.AiproxyBindings{ + {LlmId: "llm-1", SyncStatus: api.AIPROXY_BINDING_SYNC_SYNCED}, + {LlmId: "llm-2", SyncStatus: api.AIPROXY_BINDING_SYNC_FAILED, LastError: "provider upsert failed"}, + } + got := AiproxySyncFailureReason(dep) + want := "llm llm-2: provider upsert failed" + if got != want { + t.Fatalf("AiproxySyncFailureReason() = %q, want %q", got, want) + } + + msg := aiproxySyncStatusMessage(dep, aiproxyBindingSyncFailed) + if msg != want { + t.Fatalf("aiproxySyncStatusMessage(failed) = %q, want %q", msg, want) + } +} + func TestUpstreamModelKeyForBackend(t *testing.T) { cases := []struct { llmType string diff --git a/pkg/llm/models/llm_base.go b/pkg/llm/models/llm_base.go index 57a3c0f463..c75edc13d5 100644 --- a/pkg/llm/models/llm_base.go +++ b/pkg/llm/models/llm_base.go @@ -127,29 +127,13 @@ func (man *SLLMBaseManager) ValidateCreateData(ctx context.Context, userCred mcc return input, errors.Wrap(err, "validate VirtualResourceCreateInput") } - /* - if len(input.PreferHost) > 0 { - s := auth.GetSession(ctx, userCred, "") - hostJson, err := compute.Hosts.Get(s, input.PreferHost, nil) - if err != nil { - return input, errors.Wrap(err, "get host") - } - hostDetails := computeapi.HostDetails{} - if err := hostJson.Unmarshal(&hostDetails); err != nil { - return input, errors.Wrap(err, "unmarshal hostDetails") - } - if hostDetails.Enabled == nil || !*hostDetails.Enabled { - return input, errors.Wrap(errors.ErrInvalidStatus, "not enabled") - } - if hostDetails.HostStatus != computeapi.HOST_ONLINE { - return input, errors.Wrap(errors.ErrInvalidStatus, "not online") - } - if hostDetails.HostType != computeapi.HOST_TYPE_CONTAINER { - return input, errors.Wrapf(httperrors.ErrNotAcceptable, "host_type %s not supported", hostDetails.HostType) - } - input.PreferHost = hostDetails.Id + if len(input.PreferHost) > 0 { + resolved, err := resolvePreferHost(ctx, userCred, input.PreferHost) + if err != nil { + return input, errors.Wrap(err, "resolve prefer_host") } - */ + input.PreferHost = resolved + } // 处理网络配置 var firstNet *computeapi.NetworkConfig diff --git a/pkg/llm/models/llm_deployment.go b/pkg/llm/models/llm_deployment.go index 8fe8824aa0..47b6af4f34 100644 --- a/pkg/llm/models/llm_deployment.go +++ b/pkg/llm/models/llm_deployment.go @@ -99,15 +99,18 @@ type SLLMDeployment struct { // Instance template — captured at create time so scale-up can re-create // new SLLM replicas with the same network / start / mount settings as the originals. // Scale request bodies don't carry these fields. - Nets *api.LLMDeploymentNets `charset:"utf8" length:"medium" nullable:"true" list:"user"` - AutoStart bool `nullable:"false" default:"false" list:"user"` - HostPaths *api.HostPaths `charset:"utf8" length:"medium" nullable:"true" list:"user"` + Nets *api.LLMDeploymentNets `charset:"utf8" length:"medium" nullable:"true" list:"user"` + AutoStart bool `nullable:"false" default:"false" list:"user"` + HostPaths *api.HostPaths `charset:"utf8" length:"medium" nullable:"true" list:"user"` + PreferHosts []string `charset:"utf8" length:"medium" nullable:"true" list:"user" create:"optional" update:"user"` // Aiproxy integration (llm sync writes aiproxy catalog via mcclient). AutoRegisterAiproxy bool `nullable:"false" default:"false" list:"user" create:"optional" update:"user"` AiproxyModelPrefix string `width:"128" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` AiproxyRoutingId string `width:"128" charset:"ascii" nullable:"true" list:"user"` AiproxyBindings *api.AiproxyBindings `charset:"utf8" length:"long" nullable:"true" list:"user"` + // Gateway registration/sync progress (independent from replica-health status). + AiproxySyncStatus string `width:"32" charset:"ascii" nullable:"false" default:"disabled" list:"user"` } func (man *SLLMDeploymentManager) ValidateCreateData( @@ -145,7 +148,11 @@ func (man *SLLMDeploymentManager) ValidateCreateData( if err := validateDeploymentGpuMemoryUtilization(input.GpuMemoryUtilization, input.AutoGpuMemoryUtilization, lSku.LLMType); err != nil { return input, err } - defaultDeploymentAutoGpuMemoryUtilization(input, lSku.LLMType) + defaultDeploymentAutoGpuMemoryUtilization(input, lSku, lSku.LLMType) + disableDeploymentAutoGpuMemoryUtilizationForLocalPath(input, lSku) + if err := validateLocalPathDeploymentPreferHosts(ctx, userCred, input, lSku); err != nil { + return input, err + } if err := ValidateDeploymentDevices(lSku.LLMType, lSku); err != nil { return input, err } @@ -174,11 +181,16 @@ func (man *SLLMDeploymentManager) ValidateCreateData( input.ModelSpec.LlmType = api.LLMContainerType(input.SkuSpec.LLMType) } } + skuInput := skuFromLLMSkuCreateInput(input.SkuSpec) if err := validateDeploymentGpuMemoryUtilization(input.GpuMemoryUtilization, input.AutoGpuMemoryUtilization, input.SkuSpec.LLMType); err != nil { return input, err } - defaultDeploymentAutoGpuMemoryUtilization(input, input.SkuSpec.LLMType) - if err := ValidateDeploymentDevices(input.SkuSpec.LLMType, skuFromLLMSkuCreateInput(input.SkuSpec)); err != nil { + defaultDeploymentAutoGpuMemoryUtilization(input, skuInput, input.SkuSpec.LLMType) + disableDeploymentAutoGpuMemoryUtilizationForLocalPath(input, skuInput) + if err := validateLocalPathDeploymentPreferHosts(ctx, userCred, input, skuInput); err != nil { + return input, err + } + if err := ValidateDeploymentDevices(input.SkuSpec.LLMType, skuInput); err != nil { return input, err } } @@ -327,6 +339,8 @@ func (man *SLLMDeploymentManager) FetchCustomizeColumns( res[i].AutoRegisterAiproxy = models[i].AutoRegisterAiproxy res[i].AiproxyModelPrefix = models[i].AiproxyModelPrefix res[i].AiproxyRoutingId = models[i].AiproxyRoutingId + res[i].AiproxySyncStatus = models[i].AiproxySyncStatus + res[i].PreferHosts = models[i].PreferHosts } // Batch fetch SKU data for source/backend/categories info @@ -416,13 +430,15 @@ type SyncReadyReplicasOptions struct { // to the deployment row, and transitions the deployment status based on replica // health: // -// ready_replicas == 0 && desired > 0 → deploying +// ready_replicas == 0 && desired > 0 → deploying (or create_fail / start_fail) // 0 < ready_replicas < desired → partial // ready_replicas == desired → ready -// replica create/start failure → create_fail (unless some replicas run) +// replica create_fail with 0 running → create_fail +// replica start_fail with 0 running → start_fail // -// Failure / lifecycle statuses (create_fail, deleting, importing_model, etc.) -// are not overridden once set. +// Failure / lifecycle statuses (deleting, importing_model, etc.) are not +// overridden once set. create_fail and start_fail may recover to ready/partial +// when replicas become healthy again (e.g. after syncstatus or auto-restart). // // Call after create/scale tasks finish and on every instance status change. func (model *SLLMDeployment) SyncReadyReplicas(ctx context.Context, userCred mcclient.TokenCredential, opts ...SyncReadyReplicasOptions) error { @@ -451,7 +467,7 @@ func (model *SLLMDeployment) SyncReadyReplicas(ctx context.Context, userCred mcc if desired == "" { return nil } - if !canUpdateReplicaHealthStatus(model.Status) { + if !canUpdateReplicaHealthStatus(model.Status, desired) { return nil } oldStatus := model.Status @@ -483,8 +499,10 @@ type deploymentReplicaStatusRow struct { } type deploymentReplicaStatusSummary struct { - Running int - HasFailure bool + Running int + HasCreateFailure bool + HasStartFailure bool + HasProbing bool } func summarizeDeploymentReplicaStatuses(rows []deploymentReplicaStatusRow) deploymentReplicaStatusSummary { @@ -493,10 +511,12 @@ func summarizeDeploymentReplicaStatuses(rows []deploymentReplicaStatusRow) deplo switch rows[i].Status { case api.LLM_STATUS_RUNNING: summary.Running++ - default: - if isDeploymentReplicaFailureStatus(rows[i].Status) { - summary.HasFailure = true - } + case api.LLM_STATUS_CREATE_FAIL: + summary.HasCreateFailure = true + case api.LLM_STATUS_START_FAIL: + summary.HasStartFailure = true + case api.LLM_STATUS_PROBING: + summary.HasProbing = true } } return summary @@ -511,8 +531,12 @@ func computeDeploymentReplicaStatus(summary deploymentReplicaStatusSummary, desi return api.STATUS_READY case summary.Running > 0: return api.LLM_DEPLOYMENT_STATUS_PARTIAL - case summary.HasFailure: + case summary.HasCreateFailure: return api.LLM_STATUS_CREATE_FAIL + case summary.HasStartFailure: + return api.LLM_STATUS_START_FAIL + case summary.HasProbing: + return api.LLM_DEPLOYMENT_STATUS_DEPLOYING default: return api.LLM_DEPLOYMENT_STATUS_DEPLOYING } @@ -529,17 +553,22 @@ func isDeploymentReplicaFailureStatus(status string) bool { // canUpdateReplicaHealthStatus reports whether replica-health-driven status can // override the current deployment value. Early lifecycle and terminal failure / -// delete states must not be clobbered. -func canUpdateReplicaHealthStatus(current string) bool { +// delete states must not be clobbered, except create_fail and start_fail may +// recover when replicas are running again. +func canUpdateReplicaHealthStatus(current, desired string) bool { + if current == api.LLM_STATUS_CREATE_FAIL { + return desired == api.STATUS_READY || desired == api.LLM_DEPLOYMENT_STATUS_PARTIAL + } + if current == api.LLM_STATUS_START_FAIL { + return desired == api.STATUS_READY || desired == api.LLM_DEPLOYMENT_STATUS_PARTIAL + } switch current { case api.LLM_DEPLOYMENT_STATUS_IMPORTING_MODEL, api.LLM_DEPLOYMENT_STATUS_IMPORT_MODEL_FAILED, api.LLM_DEPLOYMENT_STATUS_CREATING_SKU, api.LLM_DEPLOYMENT_STATUS_CREATE_SKU_FAILED, - api.LLM_STATUS_CREATE_FAIL, api.LLM_STATUS_DELETING, - api.LLM_STATUS_DELETE_FAILED, - api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNCING: + api.LLM_STATUS_DELETE_FAILED: return false } return true @@ -597,12 +626,20 @@ func (model *SLLMDeployment) PostCreate(ctx context.Context, userCred mcclient.T if input.HostPaths != nil && !input.HostPaths.IsZero() { model.HostPaths = input.HostPaths } + if len(input.PreferHosts) > 0 { + model.PreferHosts = append([]string(nil), input.PreferHosts...) + } if input.AutoRegisterAiproxy != nil { model.AutoRegisterAiproxy = *input.AutoRegisterAiproxy } if input.AiproxyModelPrefix != "" { model.AiproxyModelPrefix = strings.TrimSpace(input.AiproxyModelPrefix) } + if model.AutoRegisterAiproxy { + model.AiproxySyncStatus = api.AIPROXY_SYNC_STATUS_PENDING + } else { + model.AiproxySyncStatus = api.AIPROXY_SYNC_STATUS_DISABLED + } return nil }); err != nil { log.Errorf("SLLMDeployment.PostCreate persist instance template: %s", err) @@ -628,7 +665,7 @@ func (model *SLLMDeployment) StartCreateTask(ctx context.Context, userCred mccli } func (model *SLLMDeployment) StartSyncReplicasTask(ctx context.Context, userCred mcclient.TokenCredential, data jsonutils.JSONObject) error { - model.SetStatus(ctx, userCred, "syncing", "") + model.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_SYNCING, "") params, _ := data.(*jsonutils.JSONDict) if params == nil { params = jsonutils.NewDict() @@ -667,6 +704,7 @@ func (model *SLLMDeployment) PerformRegisterAiproxy( ) (jsonutils.JSONObject, error) { if _, err := db.Update(model, func() error { model.AutoRegisterAiproxy = true + model.AiproxySyncStatus = api.AIPROXY_SYNC_STATUS_PENDING return nil }); err != nil { return nil, errors.Wrap(err, "enable auto_register_aiproxy") @@ -691,6 +729,146 @@ func (model *SLLMDeployment) PerformUnregisterAiproxy( return nil, nil } +// canRestartDeploymentStatus reports whether a deployment may be restarted. +func canRestartDeploymentStatus(status string) bool { + switch status { + case api.STATUS_READY, + api.LLM_DEPLOYMENT_STATUS_PARTIAL, + api.LLM_STATUS_RUNNING: + return true + } + return false +} + +func (model *SLLMDeployment) ValidateRestartInput( + ctx context.Context, + userCred mcclient.TokenCredential, + input *api.LLMDeploymentRestartInput, +) error { + if !canRestartDeploymentStatus(model.Status) { + return httperrors.NewInvalidStatusError("invalid deployment status %s", model.Status) + } + var rows []struct { + Id string `json:"id"` + } + err := GetLLMManager().Query("id").Equals("llm_deployment_id", model.Id).All(&rows) + if err != nil { + return errors.Wrap(err, "fetch deployment instances") + } + if len(rows) == 0 { + return httperrors.NewInvalidStatusError("no instances under deployment") + } + restartable := 0 + for i := range rows { + llmObj, err := GetLLMManager().FetchById(rows[i].Id) + if err != nil { + continue + } + llm := llmObj.(*SLLM) + if _, err := llm.ValidateRestartInput(ctx, userCred, &api.LLMRestartInput{}); err == nil { + restartable++ + } + } + if restartable == 0 { + return httperrors.NewInvalidStatusError("no restartable instances") + } + return nil +} + +func (model *SLLMDeployment) PerformRestart( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + input *api.LLMDeploymentRestartInput, +) (jsonutils.JSONObject, error) { + if err := model.ValidateRestartInput(ctx, userCred, input); err != nil { + return nil, err + } + if err := model.StartRestartTask(ctx, userCred); err != nil { + return nil, errors.Wrap(err, "StartRestartTask") + } + return nil, nil +} + +func (model *SLLMDeployment) StartRestartTask(ctx context.Context, userCred mcclient.TokenCredential) error { + task, err := taskman.TaskManager.NewTask(ctx, "LLMDeploymentRestartTask", model, userCred, nil, "", "", nil) + if err != nil { + return errors.Wrap(err, "NewTask LLMDeploymentRestartTask") + } + return task.ScheduleRun(nil) +} + +// canSyncDeploymentStatus reports whether a deployment may sync replica statuses. +func canSyncDeploymentStatus(status string) bool { + switch status { + case api.LLM_STATUS_DELETING, + api.LLM_STATUS_START_DELETE, + api.LLM_STATUS_DELETED, + api.LLM_STATUS_DELETE_FAILED, + "creating", + api.LLM_DEPLOYMENT_STATUS_IMPORTING_MODEL, + api.LLM_DEPLOYMENT_STATUS_CREATING_SKU, + api.LLM_DEPLOYMENT_STATUS_SYNCING: + return false + } + return true +} + +func (model *SLLMDeployment) ValidateSyncstatusInput( + ctx context.Context, + userCred mcclient.TokenCredential, + input *api.LLMDeploymentSyncstatusInput, +) error { + if !canSyncDeploymentStatus(model.Status) { + return httperrors.NewInvalidStatusError("invalid deployment status %s", model.Status) + } + var rows []struct { + Id string `json:"id"` + CmpId string `json:"cmp_id"` + } + err := GetLLMManager().Query("id", "cmp_id").Equals("llm_deployment_id", model.Id).All(&rows) + if err != nil { + return errors.Wrap(err, "fetch deployment instances") + } + if len(rows) == 0 { + return httperrors.NewInvalidStatusError("no instances under deployment") + } + syncable := 0 + for i := range rows { + if rows[i].CmpId != "" { + syncable++ + } + } + if syncable == 0 { + return httperrors.NewInvalidStatusError("no syncable instances") + } + return nil +} + +func (model *SLLMDeployment) PerformSyncstatus( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + input *api.LLMDeploymentSyncstatusInput, +) (jsonutils.JSONObject, error) { + if err := model.ValidateSyncstatusInput(ctx, userCred, input); err != nil { + return nil, err + } + if err := model.StartSyncStatusTask(ctx, userCred); err != nil { + return nil, errors.Wrap(err, "StartSyncStatusTask") + } + return nil, nil +} + +func (model *SLLMDeployment) StartSyncStatusTask(ctx context.Context, userCred mcclient.TokenCredential) error { + model.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_SYNCING, "") + task, err := taskman.TaskManager.NewTask(ctx, "LLMDeploymentSyncStatusTask", model, userCred, nil, "", "", nil) + if err != nil { + return errors.Wrap(err, "NewTask LLMDeploymentSyncStatusTask") + } + return task.ScheduleRun(nil) +} + func (model *SLLMDeployment) CustomizeDelete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error { if err := DeleteDeploymentAiproxyResources(ctx, model.Id); err != nil { log.Warningf("CustomizeDelete: delete aiproxy resources for %s: %v", model.Name, err) diff --git a/pkg/llm/models/llm_deployment_gpu.go b/pkg/llm/models/llm_deployment_gpu.go index 065f3c22ef..c585f312ad 100644 --- a/pkg/llm/models/llm_deployment_gpu.go +++ b/pkg/llm/models/llm_deployment_gpu.go @@ -2,7 +2,6 @@ package models import ( "context" - "encoding/json" "math" "strconv" "strings" @@ -44,17 +43,38 @@ func deploymentAutoGpuMemoryUtilizationEnabled(auto *bool, llmType string) bool return ok } -func defaultDeploymentAutoGpuMemoryUtilization(input *api.LLMDeploymentCreateInput, llmType string) { +func defaultDeploymentAutoGpuMemoryUtilization(input *api.LLMDeploymentCreateInput, sku *SLLMSku, llmType string) { if input == nil || input.GpuMemoryUtilization != nil || input.AutoGpuMemoryUtilization != nil { return } if !deploymentAutoGpuMemoryUtilizationEnabled(nil, llmType) { return } + if !skuCanAutoGpuMemoryUtilization(sku) { + return + } enabled := true input.AutoGpuMemoryUtilization = &enabled } +// disableDeploymentAutoGpuMemoryUtilizationForLocalPath forces auto GPU memory utilization off for host-mounted SKUs. +func disableDeploymentAutoGpuMemoryUtilizationForLocalPath(input *api.LLMDeploymentCreateInput, sku *SLLMSku) { + if input == nil || sku == nil || !SkuHasLocalHostPathModel(sku) { + return + } + disabled := false + input.AutoGpuMemoryUtilization = &disabled +} + +// skuCanAutoGpuMemoryUtilization reports whether auto GPU memory utilization can be derived for sku. +// Host-mounted local_path SKUs always opt out; use gpu_memory_utilization manually when needed. +func skuCanAutoGpuMemoryUtilization(sku *SLLMSku) bool { + if sku == nil { + return true + } + return !SkuHasLocalHostPathModel(sku) +} + func validateDeploymentGpuMemoryUtilization(util *float64, auto *bool, llmType string) error { needsRuntimeArg := util != nil || boolPtrValue(auto) if util != nil { @@ -154,8 +174,7 @@ func runtimeHasExplicitArg(sku *SLLMSku, keys []string) bool { return false } -func backendParametersContainRuntimeArg(raw string, keys []string) bool { - items := backendParameterItems(raw) +func backendParametersContainRuntimeArg(items []string, keys []string) bool { for i := range items { argKey, _, ok := splitBackendParameterFlag(items[i]) if ok && runtimeArgKeyIn(argKey, keys) { @@ -165,8 +184,7 @@ func backendParametersContainRuntimeArg(raw string, keys []string) bool { return false } -func tokenLimitFromBackendParameters(raw string, keys []string) (int64, bool) { - items := backendParameterItems(raw) +func tokenLimitFromBackendParameters(items []string, keys []string) (int64, bool) { var tokenLimit int64 found := false for i := range items { @@ -188,18 +206,6 @@ func tokenLimitFromBackendParameters(raw string, keys []string) (int64, bool) { return tokenLimit, found } -func backendParameterItems(raw string) []string { - raw = strings.TrimSpace(raw) - if raw == "" { - return nil - } - items := []string{} - if err := json.Unmarshal([]byte(raw), &items); err != nil { - return []string{raw} - } - return items -} - func splitBackendParameterFlag(item string) (string, string, bool) { item = strings.TrimSpace(item) if item == "" || strings.HasPrefix(item, "-") && !strings.HasPrefix(item, "--") { @@ -477,6 +483,9 @@ func BuildDeploymentResolvedGpuMemoryLLMSpec(ctx context.Context, userCred mccli if deploy.GpuMemoryUtilization != nil { return buildDeploymentGpuMemoryLLMSpec(deploy, sku) } + if !skuCanAutoGpuMemoryUtilization(sku) { + return nil, nil + } if !deploymentAutoGpuMemoryUtilizationEnabled(deploy.AutoGpuMemoryUtilization, sku.LLMType) { return nil, nil } @@ -500,6 +509,10 @@ func BuildDeploymentResolvedGpuMemoryLLMSpec(ctx context.Context, userCred mccli } func maxMountedModelVramRequirementMB(sku *SLLMSku) (int64, error) { + if sku != nil && SkuHasLocalHostPathModel(sku) { + return 0, httperrors.NewInputParameterError( + "auto_gpu_memory_utilization is not supported for local_path SKU: set gpu_memory_utilization manually or omit both") + } modelIds := sku.GetMountedModels() if len(modelIds) == 0 { return 0, httperrors.NewInputParameterError("auto_gpu_memory_utilization requires mounted models: configure mounted_models on the LLM SKU") diff --git a/pkg/llm/models/llm_deployment_gpu_local_path_test.go b/pkg/llm/models/llm_deployment_gpu_local_path_test.go new file mode 100644 index 0000000000..35992c0b9b --- /dev/null +++ b/pkg/llm/models/llm_deployment_gpu_local_path_test.go @@ -0,0 +1,66 @@ +package models + +import ( + "testing" + + api "yunion.io/x/onecloud/pkg/apis/llm" +) + +func localPathTestSku() *SLLMSku { + return &SLLMSku{ + LLMType: string(api.LLM_CONTAINER_VLLM), + Source: api.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + SLLMSkuBase: SLLMSkuBase{ + HostPaths: &api.HostPaths{{ + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: map[string]*api.ContainerHostPathRelation{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B"}, + }, + }}, + }, + } +} + +func TestSkuCanAutoGpuMemoryUtilizationLocalPath(t *testing.T) { + localSku := localPathTestSku() + if skuCanAutoGpuMemoryUtilization(localSku) { + t.Fatal("local_path SKU should not allow auto GPU util") + } + localSku.VramClaimMb = 16384 + if skuCanAutoGpuMemoryUtilization(localSku) { + t.Fatal("local_path SKU should not allow auto GPU util even with vram_claim_mb") + } + if !skuCanAutoGpuMemoryUtilization(nil) { + t.Fatal("nil sku should keep legacy default true") + } +} + +func TestDefaultDeploymentAutoGpuMemoryUtilizationSkipsLocalPath(t *testing.T) { + input := &api.LLMDeploymentCreateInput{} + sku := localPathTestSku() + defaultDeploymentAutoGpuMemoryUtilization(input, sku, string(api.LLM_CONTAINER_VLLM)) + if input.AutoGpuMemoryUtilization != nil { + t.Fatalf("expected auto_gpu_memory_utilization unset for local_path SKU, got %v", *input.AutoGpuMemoryUtilization) + } +} + +func TestDisableDeploymentAutoGpuMemoryUtilizationForLocalPath(t *testing.T) { + input := &api.LLMDeploymentCreateInput{ + AutoGpuMemoryUtilization: func() *bool { v := true; return &v }(), + } + disableDeploymentAutoGpuMemoryUtilizationForLocalPath(input, localPathTestSku()) + if input.AutoGpuMemoryUtilization == nil || *input.AutoGpuMemoryUtilization { + t.Fatalf("expected auto_gpu_memory_utilization forced false, got %v", input.AutoGpuMemoryUtilization) + } +} + +func TestMaxMountedModelVramRequirementMBLocalPathRejected(t *testing.T) { + sku := localPathTestSku() + sku.VramClaimMb = 20480 + _, err := maxMountedModelVramRequirementMB(sku) + if err == nil { + t.Fatal("expected error for local_path SKU auto GPU util") + } +} diff --git a/pkg/llm/models/llm_deployment_restart_test.go b/pkg/llm/models/llm_deployment_restart_test.go new file mode 100644 index 0000000000..6b631aa3bd --- /dev/null +++ b/pkg/llm/models/llm_deployment_restart_test.go @@ -0,0 +1,31 @@ +package models + +import ( + "testing" + + api "yunion.io/x/onecloud/pkg/apis/llm" +) + +func TestCanRestartDeploymentStatus(t *testing.T) { + cases := []struct { + status string + want bool + }{ + {api.STATUS_READY, true}, + {api.LLM_DEPLOYMENT_STATUS_PARTIAL, true}, + {api.LLM_STATUS_RUNNING, true}, + {api.LLM_DEPLOYMENT_STATUS_DEPLOYING, false}, + {api.LLM_STATUS_DELETING, false}, + {api.LLM_DEPLOYMENT_STATUS_IMPORTING_MODEL, false}, + {api.LLM_DEPLOYMENT_STATUS_CREATING_SKU, false}, + {api.LLM_STATUS_CREATE_FAIL, false}, + {api.LLM_STATUS_DELETE_FAILED, false}, + {"unknown", false}, + } + for _, tc := range cases { + got := canRestartDeploymentStatus(tc.status) + if got != tc.want { + t.Errorf("canRestartDeploymentStatus(%q) = %v, want %v", tc.status, got, tc.want) + } + } +} diff --git a/pkg/llm/models/llm_sku.go b/pkg/llm/models/llm_sku.go index d7bc54a60e..58cb3c13e3 100644 --- a/pkg/llm/models/llm_sku.go +++ b/pkg/llm/models/llm_sku.go @@ -2,6 +2,7 @@ package models import ( "context" + "fmt" "strings" "yunion.io/x/jsonutils" @@ -58,18 +59,16 @@ type SLLMSku struct { LLMSpec *api.LLMSpec `json:"llm_spec" length:"long" list:"user" create:"optional" update:"user"` // Model source - Source string `width:"32" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user"` - HuggingfaceRepoId string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` - HuggingfaceFilename string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` - ModelScopeModelId string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` - ModelScopeFilePath string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` - LocalPath string `width:"512" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` - // Model categories, JSON array: ["llm"], ["embedding"], ["image"] - Categories string `charset:"utf8" length:"medium" nullable:"true" list:"user" create:"optional" update:"user"` - // Inference backend version - BackendVersion string `width:"64" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user"` - // Backend CLI parameters, JSON array - BackendParameters string `charset:"utf8" length:"long" nullable:"true" list:"user" create:"optional" update:"user"` + Source string `width:"32" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user"` + HuggingfaceRepoId string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + HuggingfaceFilename string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + ModelScopeModelId string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + ModelScopeFilePath string `width:"256" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + LocalPath string `width:"512" charset:"utf8" nullable:"true" list:"user" create:"optional" update:"user"` + PreferHosts []string `charset:"utf8" length:"medium" nullable:"true" list:"user" create:"optional" update:"user"` + Categories []string `charset:"utf8" length:"medium" nullable:"true" list:"user" create:"optional" update:"user"` + BackendVersion string `width:"64" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user"` + BackendParameters []string `charset:"utf8" length:"long" nullable:"true" list:"user" create:"optional" update:"user"` } func (man *SLLMSkuManager) ListItemFilter( @@ -97,7 +96,7 @@ func (man *SLLMSkuManager) ListItemFilter( q = q.Equals("source", input.Source) } if len(input.Categories) > 0 { - q = q.Contains("categories", input.Categories) + q = q.Contains("categories", fmt.Sprintf("%q", input.Categories)) } return q, nil } @@ -147,6 +146,7 @@ func (manager *SLLMSkuManager) FetchCustomizeColumns( res[i].ModelScopeModelId = sku.ModelScopeModelId res[i].ModelScopeFilePath = sku.ModelScopeFilePath res[i].LocalPath = sku.LocalPath + res[i].PreferHosts = GetSkuPreferHosts(&sku) res[i].Categories = sku.Categories res[i].BackendVersion = sku.BackendVersion res[i].BackendParameters = sku.BackendParameters @@ -239,6 +239,16 @@ func (man *SLLMSkuManager) ValidateCreateData(ctx context.Context, userCred mccl if err != nil { return input, errors.Wrap(err, "validate create input") } + if isLocalPathSkuCreate(input) { + if err := ValidateLocalPathSkuCreate(input); err != nil { + return input, err + } + resolved, err := resolvePreferHosts(ctx, userCred, input.PreferHosts) + if err != nil { + return input, err + } + input.PreferHosts = resolved + } importInput, err := resolveLLMSkuImport(input) if err != nil { return input, err @@ -251,6 +261,14 @@ func (man *SLLMSkuManager) ValidateCreateData(ctx context.Context, userCred mccl return input, nil } +func (sku *SLLMSku) CustomizeCreate(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data jsonutils.JSONObject) error { + err := sku.SSharableVirtualResourceBase.CustomizeCreate(ctx, userCred, ownerId, query, data) + if err != nil { + return errors.Wrap(err, "SSharableVirtualResourceBase.CustomizeCreate") + } + return nil +} + func (sku *SLLMSku) PostCreate(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data jsonutils.JSONObject) { sku.SSharableVirtualResourceBase.PostCreate(ctx, userCred, ownerId, query, data) input := api.LLMSkuCreateInput{} diff --git a/pkg/llm/models/llm_sku_local_path.go b/pkg/llm/models/llm_sku_local_path.go new file mode 100644 index 0000000000..5b4c6869e6 --- /dev/null +++ b/pkg/llm/models/llm_sku_local_path.go @@ -0,0 +1,88 @@ +package models + +import ( + "fmt" + "strings" + + "yunion.io/x/pkg/errors" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/httperrors" +) + +func isLocalPathSkuCreate(input *api.LLMSkuCreateInput) bool { + if input == nil { + return false + } + return strings.EqualFold(strings.TrimSpace(input.Source), api.LLM_MODEL_SOURCE_LOCAL_PATH) +} + +// ValidateLocalPathSkuCreate validates SKU create requests that mount an on-host model directory. +func ValidateLocalPathSkuCreate(input *api.LLMSkuCreateInput) error { + if input == nil { + return errors.Wrap(httperrors.ErrInputParameter, "empty sku input") + } + llmType := strings.TrimSpace(input.LLMType) + if llmType != string(api.LLM_CONTAINER_VLLM) && llmType != string(api.LLM_CONTAINER_SGLANG) { + return errors.Wrapf(httperrors.ErrInputParameter, "local_path import supports vllm and sglang only, got %q", llmType) + } + localPath := strings.TrimSpace(input.LocalPath) + if localPath == "" { + return errors.Wrap(httperrors.ErrMissingParameter, "local_path is required for local_path source") + } + if !strings.HasPrefix(localPath, "/") { + return errors.Wrap(httperrors.ErrInputParameter, "local_path must be an absolute path") + } + if input.ModelSpec != nil { + return errors.Wrap(httperrors.ErrInputParameter, "model_spec is not allowed for local_path import") + } + if input.HostPaths == nil || input.HostPaths.IsZero() { + return errors.Wrap(httperrors.ErrMissingParameter, "host_paths is required for local_path source") + } + if !hostPathsHasContainerMount(*input.HostPaths, 0) { + return errors.Wrap(httperrors.ErrInputParameter, "host_paths must include a mount for container index 0") + } + if len(normalizePreferHostInputs(input.PreferHosts)) == 0 { + return errors.Wrap(httperrors.ErrMissingParameter, "prefer_hosts is required for local_path source") + } + input.Source = api.LLM_MODEL_SOURCE_LOCAL_PATH + input.LocalPath = localPath + return nil +} + +func hostPathsHasContainerMount(paths api.HostPaths, containerIndex int) bool { + key := fmt.Sprintf("%d", containerIndex) + for _, hp := range paths { + if hp.IsZero() { + continue + } + if hp.Containers == nil { + continue + } + rel, ok := hp.Containers[key] + if !ok || rel == nil { + continue + } + if strings.TrimSpace(rel.MountPath) != "" { + return true + } + } + return false +} + +// SkuHasLocalHostPathModel reports whether sku carries a host-mounted local model (no InstantModel import). +func SkuHasLocalHostPathModel(sku *SLLMSku) bool { + if sku == nil { + return false + } + if strings.TrimSpace(sku.Source) != api.LLM_MODEL_SOURCE_LOCAL_PATH { + return false + } + if strings.TrimSpace(sku.LocalPath) == "" { + return false + } + if sku.HostPaths == nil || sku.HostPaths.IsZero() { + return false + } + return hostPathsHasContainerMount(*sku.HostPaths, 0) +} diff --git a/pkg/llm/models/llm_sku_local_path_test.go b/pkg/llm/models/llm_sku_local_path_test.go new file mode 100644 index 0000000000..e0238c03cd --- /dev/null +++ b/pkg/llm/models/llm_sku_local_path_test.go @@ -0,0 +1,141 @@ +package models + +import ( + "testing" + + "yunion.io/x/onecloud/pkg/apis/llm" +) + +func TestValidateLocalPathSkuCreate(t *testing.T) { + hostPaths := llm.HostPaths{ + { + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: llm.ContainerHostPathRelations{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B", ReadOnly: true}, + }, + }, + } + input := &llm.LLMSkuCreateInput{ + LLMSKuBaseCreateInput: llm.LLMSKuBaseCreateInput{ + HostPaths: &hostPaths, + }, + LLMType: string(llm.LLM_CONTAINER_VLLM), + Source: llm.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + PreferHosts: []string{"host-1"}, + } + if err := ValidateLocalPathSkuCreate(input); err != nil { + t.Fatalf("expected valid local_path sku, got %v", err) + } + if input.Source != llm.LLM_MODEL_SOURCE_LOCAL_PATH { + t.Fatalf("expected source local_path, got %q", input.Source) + } +} + +func TestValidateLocalPathSkuCreateRejectsModelSpec(t *testing.T) { + hostPaths := llm.HostPaths{ + { + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: llm.ContainerHostPathRelations{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B"}, + }, + }, + } + input := &llm.LLMSkuCreateInput{ + LLMSKuBaseCreateInput: llm.LLMSKuBaseCreateInput{ + HostPaths: &hostPaths, + }, + LLMType: string(llm.LLM_CONTAINER_VLLM), + Source: llm.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + PreferHosts: []string{"host-1"}, + ModelSpec: &llm.InstantModelImportInput{ModelName: "x", ModelTag: "main"}, + } + if err := ValidateLocalPathSkuCreate(input); err == nil { + t.Fatal("expected error when model_spec is set") + } +} + +func TestValidateLocalPathSkuCreateRequiresPreferHosts(t *testing.T) { + hostPaths := llm.HostPaths{ + { + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: llm.ContainerHostPathRelations{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B"}, + }, + }, + } + input := &llm.LLMSkuCreateInput{ + LLMSKuBaseCreateInput: llm.LLMSKuBaseCreateInput{ + HostPaths: &hostPaths, + }, + LLMType: string(llm.LLM_CONTAINER_VLLM), + Source: llm.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + } + if err := ValidateLocalPathSkuCreate(input); err == nil { + t.Fatal("expected error when prefer_hosts is missing") + } +} + +func TestValidateLocalPathSkuCreateRequiresContainerMount(t *testing.T) { + hostPaths := llm.HostPaths{ + {Type: "directory", Path: "/data/models/Qwen3-8B"}, + } + input := &llm.LLMSkuCreateInput{ + LLMSKuBaseCreateInput: llm.LLMSKuBaseCreateInput{ + HostPaths: &hostPaths, + }, + LLMType: string(llm.LLM_CONTAINER_SGLANG), + Source: llm.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + PreferHosts: []string{"host-1"}, + } + if err := ValidateLocalPathSkuCreate(input); err == nil { + t.Fatal("expected error when container 0 mount is missing") + } +} + +func TestSkuHasLocalHostPathModel(t *testing.T) { + hostPaths := llm.HostPaths{ + { + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: llm.ContainerHostPathRelations{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B"}, + }, + }, + } + sku := &SLLMSku{ + Source: llm.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + } + sku.HostPaths = &hostPaths + if !SkuHasLocalHostPathModel(sku) { + t.Fatal("expected local host path model sku") + } +} + +func TestValidateRequireMountedModelsSkipsLocalPathSku(t *testing.T) { + hostPaths := llm.HostPaths{ + { + Type: "directory", + Path: "/data/models/Qwen3-8B", + Containers: llm.ContainerHostPathRelations{ + "0": {MountPath: "/data/models/huggingface/Qwen3-8B"}, + }, + }, + } + sku := &SLLMSku{ + LLMType: string(llm.LLM_CONTAINER_VLLM), + Source: llm.LLM_MODEL_SOURCE_LOCAL_PATH, + LocalPath: "/data/models/Qwen3-8B", + } + sku.HostPaths = &hostPaths + if err := ValidateRequireMountedModels(string(llm.LLM_CONTAINER_VLLM), nil, nil, sku); err != nil { + t.Fatalf("expected mounted_models not required for local_path sku, got %v", err) + } +} diff --git a/pkg/llm/models/llm_status_resolver.go b/pkg/llm/models/llm_status_resolver.go index 60ebea50b7..01539c6251 100644 --- a/pkg/llm/models/llm_status_resolver.go +++ b/pkg/llm/models/llm_status_resolver.go @@ -33,6 +33,12 @@ func resolveLLMStatusFromPod(currentStatus string, serverStatus string, primaryC reason := fmt.Sprintf("pod status=%s primary_container_status=%s", serverStatus, primaryContainerStatus) switch { + case serverStatus == computeapi.VM_READY: + targetStatus = api.LLM_STATUS_READY + case isPrimaryContainerRunning(primaryContainerStatus): + targetStatus = api.LLM_STATUS_RUNNING + case serverStatus == computeapi.VM_RUNNING && primaryContainerStatus == computeapi.CONTAINER_STATUS_PROBING: + targetStatus = api.LLM_STATUS_PROBING case isLLMPodCrashLoopStatus(serverStatus, primaryContainerStatus) || isLLMStartupProbeFailedStatus(primaryContainerStatus): if currentStatus == commonapi.STATUS_CREATING { targetStatus = api.LLM_STATUS_CREATE_FAIL @@ -41,12 +47,6 @@ func resolveLLMStatusFromPod(currentStatus string, serverStatus string, primaryC } case serverStatus == computeapi.POD_STATUS_CONTAINER_EXITED || primaryContainerStatus == computeapi.CONTAINER_STATUS_EXITED: targetStatus = api.LLM_STATUS_START_FAIL - case serverStatus == computeapi.VM_RUNNING && primaryContainerStatus == computeapi.CONTAINER_STATUS_PROBING: - targetStatus = api.LLM_STATUS_PROBING - case serverStatus == computeapi.VM_RUNNING && isPrimaryContainerRunning(primaryContainerStatus): - targetStatus = api.LLM_STATUS_RUNNING - case serverStatus == computeapi.VM_READY: - targetStatus = api.LLM_STATUS_READY default: return llmStatusResolution{} } @@ -65,6 +65,9 @@ func resolveLLMStatusFromPod(currentStatus string, serverStatus string, primaryC } func isLLMPodCrashLoopStatus(serverStatus string, primaryContainerStatus string) bool { + if isPrimaryContainerRunning(primaryContainerStatus) { + return false + } return serverStatus == computeapi.POD_STATUS_CRASH_LOOP_BACK_OFF || primaryContainerStatus == computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF } diff --git a/pkg/llm/models/llm_status_resolver_test.go b/pkg/llm/models/llm_status_resolver_test.go new file mode 100644 index 0000000000..d7ce595f53 --- /dev/null +++ b/pkg/llm/models/llm_status_resolver_test.go @@ -0,0 +1,102 @@ +package models + +import ( + "testing" + + computeapi "yunion.io/x/onecloud/pkg/apis/compute" + api "yunion.io/x/onecloud/pkg/apis/llm" +) + +func TestResolveLLMStatusFromPod(t *testing.T) { + cases := []struct { + name string + currentStatus string + serverStatus string + primaryContainerStatus string + wantStatus string + wantUpdate bool + }{ + { + name: "server ready container exited after start_fail sync", + currentStatus: api.LLM_STATUS_START_FAIL, + serverStatus: computeapi.VM_READY, + primaryContainerStatus: computeapi.CONTAINER_STATUS_EXITED, + wantStatus: api.LLM_STATUS_READY, + wantUpdate: true, + }, + { + name: "server ready stale probe_failed after start_fail sync", + currentStatus: api.LLM_STATUS_START_FAIL, + serverStatus: computeapi.VM_READY, + primaryContainerStatus: computeapi.CONTAINER_STATUS_PROBE_FAILED, + wantStatus: api.LLM_STATUS_READY, + wantUpdate: true, + }, + { + name: "already ready no update", + currentStatus: api.LLM_STATUS_READY, + serverStatus: computeapi.VM_READY, + primaryContainerStatus: computeapi.CONTAINER_STATUS_EXITED, + wantUpdate: false, + }, + { + name: "running with probe failed is start_fail", + currentStatus: api.LLM_STATUS_RUNNING, + serverStatus: computeapi.VM_RUNNING, + primaryContainerStatus: computeapi.CONTAINER_STATUS_PROBE_FAILED, + wantStatus: api.LLM_STATUS_START_FAIL, + wantUpdate: true, + }, + { + name: "running container running", + currentStatus: api.LLM_STATUS_PROBING, + serverStatus: computeapi.VM_RUNNING, + primaryContainerStatus: computeapi.CONTAINER_STATUS_RUNNING, + wantStatus: api.LLM_STATUS_RUNNING, + wantUpdate: true, + }, + { + name: "running container exited is start_fail", + currentStatus: api.LLM_STATUS_RUNNING, + serverStatus: computeapi.VM_RUNNING, + primaryContainerStatus: computeapi.CONTAINER_STATUS_EXITED, + wantStatus: api.LLM_STATUS_START_FAIL, + wantUpdate: true, + }, + { + name: "start_fail recovers when server crash_loop but container running", + currentStatus: api.LLM_STATUS_START_FAIL, + serverStatus: computeapi.POD_STATUS_CRASH_LOOP_BACK_OFF, + primaryContainerStatus: computeapi.CONTAINER_STATUS_RUNNING, + wantStatus: api.LLM_STATUS_RUNNING, + wantUpdate: true, + }, + { + name: "start_fail recovers when server and container running", + currentStatus: api.LLM_STATUS_START_FAIL, + serverStatus: computeapi.VM_RUNNING, + primaryContainerStatus: computeapi.CONTAINER_STATUS_RUNNING, + wantStatus: api.LLM_STATUS_RUNNING, + wantUpdate: true, + }, + { + name: "running degrades when container crash_loop", + currentStatus: api.LLM_STATUS_RUNNING, + serverStatus: computeapi.VM_RUNNING, + primaryContainerStatus: computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF, + wantStatus: api.LLM_STATUS_START_FAIL, + wantUpdate: true, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := resolveLLMStatusFromPod(tc.currentStatus, tc.serverStatus, tc.primaryContainerStatus) + if got.Update != tc.wantUpdate { + t.Fatalf("Update = %v, want %v", got.Update, tc.wantUpdate) + } + if tc.wantUpdate && got.Status != tc.wantStatus { + t.Fatalf("Status = %q, want %q", got.Status, tc.wantStatus) + } + }) + } +} diff --git a/pkg/llm/models/prefer_hosts.go b/pkg/llm/models/prefer_hosts.go new file mode 100644 index 0000000000..68ec76570f --- /dev/null +++ b/pkg/llm/models/prefer_hosts.go @@ -0,0 +1,141 @@ +package models + +import ( + "context" + "strings" + + "yunion.io/x/pkg/errors" + "yunion.io/x/pkg/util/sets" + + computeapi "yunion.io/x/onecloud/pkg/apis/compute" + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/mcclient/auth" + "yunion.io/x/onecloud/pkg/mcclient/modules/compute" +) + +// GetSkuPreferHosts returns host ids stored on sku. +func GetSkuPreferHosts(sku *SLLMSku) []string { + if sku == nil || len(sku.PreferHosts) == 0 { + return nil + } + return append([]string(nil), sku.PreferHosts...) +} + +// GetDeploymentPreferHosts returns deployment prefer_hosts, falling back to sku. +func GetDeploymentPreferHosts(dep *SLLMDeployment, sku *SLLMSku) []string { + if dep != nil && len(dep.PreferHosts) > 0 { + return append([]string(nil), dep.PreferHosts...) + } + return GetSkuPreferHosts(sku) +} + +// SelectPreferHostForInstanceIndex picks a host from prefer_hosts using round-robin. +func SelectPreferHostForInstanceIndex(hosts []string, index int) string { + if len(hosts) == 0 { + return "" + } + if index < 0 { + index = 0 + } + return hosts[index%len(hosts)] +} + +func validatePreferHostsSubset(selected, allowed []string) error { + if len(selected) == 0 { + return httperrors.NewMissingParameterError("prefer_hosts") + } + if len(allowed) == 0 { + return nil + } + allowedSet := sets.NewString(allowed...) + for _, h := range selected { + if !allowedSet.Has(h) { + return httperrors.NewInputParameterError("prefer_hosts %q is not declared on the LLM SKU", h) + } + } + return nil +} + +func normalizePreferHostInputs(hosts []string) []string { + seen := sets.NewString() + out := make([]string, 0, len(hosts)) + for _, h := range hosts { + h = strings.TrimSpace(h) + if h == "" || seen.Has(h) { + continue + } + seen.Insert(h) + out = append(out, h) + } + return out +} + +func resolvePreferHosts(ctx context.Context, userCred mcclient.TokenCredential, hosts []string) ([]string, error) { + hosts = normalizePreferHostInputs(hosts) + if len(hosts) == 0 { + return nil, httperrors.NewMissingParameterError("prefer_hosts") + } + s := auth.GetSession(ctx, userCred, "") + resolved := make([]string, 0, len(hosts)) + for _, hostRef := range hosts { + hostJson, err := compute.Hosts.Get(s, hostRef, nil) + if err != nil { + return nil, errors.Wrapf(err, "get host %s", hostRef) + } + hostDetails := computeapi.HostDetails{} + if err := hostJson.Unmarshal(&hostDetails); err != nil { + return nil, errors.Wrap(err, "unmarshal hostDetails") + } + if hostDetails.Enabled == nil || !*hostDetails.Enabled { + return nil, errors.Wrapf(httperrors.ErrInvalidStatus, "host %s is not enabled", hostRef) + } + if hostDetails.HostStatus != computeapi.HOST_ONLINE { + return nil, errors.Wrapf(httperrors.ErrInvalidStatus, "host %s is not online", hostRef) + } + if hostDetails.HostType != computeapi.HOST_TYPE_CONTAINER { + return nil, errors.Wrapf(httperrors.ErrNotAcceptable, "host %s type %s is not supported", hostRef, hostDetails.HostType) + } + resolved = append(resolved, hostDetails.Id) + } + return normalizePreferHostInputs(resolved), nil +} + +func resolvePreferHost(ctx context.Context, userCred mcclient.TokenCredential, hostRef string) (string, error) { + hosts, err := resolvePreferHosts(ctx, userCred, []string{hostRef}) + if err != nil { + return "", err + } + if len(hosts) == 0 { + return "", httperrors.NewMissingParameterError("prefer_host") + } + return hosts[0], nil +} + +func validateLocalPathDeploymentPreferHosts( + ctx context.Context, + userCred mcclient.TokenCredential, + input *api.LLMDeploymentCreateInput, + sku *SLLMSku, +) error { + if input == nil || sku == nil || !SkuHasLocalHostPathModel(sku) { + return nil + } + skuHosts := GetSkuPreferHosts(sku) + if len(input.PreferHosts) == 0 { + input.PreferHosts = append([]string(nil), skuHosts...) + } + if len(input.PreferHosts) == 0 { + return httperrors.NewMissingParameterError("prefer_hosts is required for local_path SKU") + } + resolved, err := resolvePreferHosts(ctx, userCred, input.PreferHosts) + if err != nil { + return err + } + if err := validatePreferHostsSubset(resolved, skuHosts); err != nil { + return err + } + input.PreferHosts = resolved + return nil +} diff --git a/pkg/llm/models/prefer_hosts_test.go b/pkg/llm/models/prefer_hosts_test.go new file mode 100644 index 0000000000..ac97e55e64 --- /dev/null +++ b/pkg/llm/models/prefer_hosts_test.go @@ -0,0 +1,78 @@ +package models + +import ( + "testing" +) + +func TestNormalizePreferHostInputs(t *testing.T) { + got := normalizePreferHostInputs([]string{" host-1 ", "host-2", "host-1", "", "host-3"}) + want := []string{"host-1", "host-2", "host-3"} + if len(got) != len(want) { + t.Fatalf("got %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("got %v, want %v", got, want) + } + } +} + +func TestGetSkuPreferHosts(t *testing.T) { + sku := &SLLMSku{PreferHosts: []string{"host-a", "host-b"}} + got := GetSkuPreferHosts(sku) + if len(got) != 2 || got[0] != "host-a" || got[1] != "host-b" { + t.Fatalf("unexpected hosts: %v", got) + } +} + +func TestGetDeploymentPreferHostsPrefersDeployment(t *testing.T) { + sku := &SLLMSku{PreferHosts: []string{"sku-host"}} + dep := &SLLMDeployment{PreferHosts: []string{"dep-host"}} + got := GetDeploymentPreferHosts(dep, sku) + if len(got) != 1 || got[0] != "dep-host" { + t.Fatalf("expected deployment hosts, got %v", got) + } +} + +func TestGetDeploymentPreferHostsFallsBackToSku(t *testing.T) { + sku := &SLLMSku{PreferHosts: []string{"sku-host"}} + dep := &SLLMDeployment{} + got := GetDeploymentPreferHosts(dep, sku) + if len(got) != 1 || got[0] != "sku-host" { + t.Fatalf("expected sku hosts, got %v", got) + } +} + +func TestSelectPreferHostForInstanceIndexRoundRobin(t *testing.T) { + hosts := []string{"h1", "h2", "h3"} + cases := []struct { + index int + want string + }{ + {0, "h1"}, + {1, "h2"}, + {2, "h3"}, + {3, "h1"}, + {5, "h3"}, + } + for _, c := range cases { + if got := SelectPreferHostForInstanceIndex(hosts, c.index); got != c.want { + t.Fatalf("index %d: got %q, want %q", c.index, got, c.want) + } + } + if got := SelectPreferHostForInstanceIndex(nil, 0); got != "" { + t.Fatalf("expected empty for nil hosts, got %q", got) + } +} + +func TestValidatePreferHostsSubset(t *testing.T) { + if err := validatePreferHostsSubset([]string{"h1"}, []string{"h1", "h2"}); err != nil { + t.Fatalf("expected valid subset, got %v", err) + } + if err := validatePreferHostsSubset([]string{"h3"}, []string{"h1", "h2"}); err == nil { + t.Fatal("expected error for host not on sku") + } + if err := validatePreferHostsSubset(nil, []string{"h1"}); err == nil { + t.Fatal("expected error for empty selected hosts") + } +} diff --git a/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go b/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go index a88c6a78cb..37285aac35 100644 --- a/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go +++ b/pkg/llm/tasks/llm/llm_aiproxy_sync_task.go @@ -10,6 +10,7 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" "yunion.io/x/onecloud/pkg/llm/models" + "yunion.io/x/onecloud/pkg/util/logclient" ) type LLMAiproxySyncTask struct { @@ -20,11 +21,45 @@ func init() { taskman.RegisterTask(LLMAiproxySyncTask{}) } +func (task *LLMAiproxySyncTask) aiproxyLogAction() string { + unregister, _ := task.GetParams().Bool("unregister") + if unregister { + return logclient.ACT_UNREGISTER_AIPROXY + } + return logclient.ACT_REGISTER_AIPROXY +} + +func (task *LLMAiproxySyncTask) logAiproxySync(ctx context.Context, dep *models.SLLMDeployment, err error) { + action := task.aiproxyLogAction() + if err != nil { + db.OpsLog.LogEvent(dep, "aiproxy_sync", err.Error(), task.UserCred) + logclient.AddActionLogWithStartable(task, dep, action, err, task.UserCred, false) + return + } + depObj, fetchErr := models.GetLLMDeploymentManager().FetchById(dep.Id) + if fetchErr == nil { + dep = depObj.(*models.SLLMDeployment) + } + if dep.AiproxySyncStatus == api.AIPROXY_SYNC_STATUS_FAILED { + msg := models.AiproxySyncFailureReason(dep) + if msg == "" { + msg = "aiproxy sync failed" + } + db.OpsLog.LogEvent(dep, "aiproxy_sync", msg, task.UserCred) + logclient.AddActionLogWithStartable(task, dep, action, msg, task.UserCred, false) + return + } + logclient.AddActionLogWithStartable(task, dep, action, nil, task.UserCred, true) +} + func (task *LLMAiproxySyncTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { dep := obj.(*models.SLLMDeployment) if !dep.AutoRegisterAiproxy { - task.SetStageComplete(ctx, nil) - return + unregister, _ := task.GetParams().Bool("unregister") + if !unregister { + task.SetStageComplete(ctx, nil) + return + } } llmId, _ := task.GetParams().GetString("llm_id") @@ -51,19 +86,11 @@ func (task *LLMAiproxySyncTask) OnInit(ctx context.Context, obj db.IStandaloneMo if err != nil { log.Errorf("LLMAiproxySyncTask deployment=%s: %v", dep.Name, err) - if !isDeploymentHealthyAfterAiproxySync(dep.Status) { - _ = dep.SetStatus(ctx, task.UserCred, api.LLM_DEPLOYMENT_STATUS_AIPROXY_SYNC_FAILED, err.Error()) - } + _ = dep.SetAiproxySyncStatus(ctx, task.UserCred, api.AIPROXY_SYNC_STATUS_FAILED, err.Error()) + task.logAiproxySync(ctx, dep, err) task.SetStageFailed(ctx, jsonutils.NewString(err.Error())) return } + task.logAiproxySync(ctx, dep, nil) task.SetStageComplete(ctx, nil) } - -func isDeploymentHealthyAfterAiproxySync(status string) bool { - switch status { - case api.STATUS_READY, api.LLM_DEPLOYMENT_STATUS_PARTIAL: - return true - } - return false -} diff --git a/pkg/llm/tasks/llm/llm_deployment_create_task.go b/pkg/llm/tasks/llm/llm_deployment_create_task.go index 0e5a3e0a32..65af3f551e 100644 --- a/pkg/llm/tasks/llm/llm_deployment_create_task.go +++ b/pkg/llm/tasks/llm/llm_deployment_create_task.go @@ -373,6 +373,7 @@ func scaleUp(ctx context.Context, userCred mcclient.TokenCredential, deployment if err != nil { return fmt.Errorf("resolve GPU memory utilization: %w", err) } + preferHosts := models.GetDeploymentPreferHosts(deployment, llmSku) var lastErr error created := 0 @@ -380,7 +381,8 @@ func scaleUp(ctx context.Context, userCred mcclient.TokenCredential, deployment instanceName := fmt.Sprintf("%s-%d", deployment.Name, nextIndex+i) // Build typed LLMCreateInput, then marshal to JSON for handler.Create - llmInput := buildDeploymentLLMCreateInput(deployment, nets, imageId, llmSpec) + preferHost := models.SelectPreferHostForInstanceIndex(preferHosts, nextIndex+i) + llmInput := buildDeploymentLLMCreateInput(deployment, nets, imageId, llmSpec, preferHost) llmParams := jsonutils.Marshal(llmInput).(*jsonutils.JSONDict) // Name lives on the embedded VirtualResourceCreateInput → set explicitly @@ -405,11 +407,12 @@ func scaleUp(ctx context.Context, userCred mcclient.TokenCredential, deployment return nil } -func buildDeploymentLLMCreateInput(deployment *models.SLLMDeployment, nets []*computeapi.NetworkConfig, imageId string, llmSpec *api.LLMSpec) api.LLMCreateInput { +func buildDeploymentLLMCreateInput(deployment *models.SLLMDeployment, nets []*computeapi.NetworkConfig, imageId string, llmSpec *api.LLMSpec, preferHost string) api.LLMCreateInput { return api.LLMCreateInput{ LLMBaseCreateInput: api.LLMBaseCreateInput{ - AutoStart: deployment.AutoStart, - Nets: nets, + AutoStart: deployment.AutoStart, + Nets: nets, + PreferHost: preferHost, }, LLMSkuId: deployment.LLMSkuId, LLMImageId: imageId, diff --git a/pkg/llm/tasks/llm/llm_deployment_restart_task.go b/pkg/llm/tasks/llm/llm_deployment_restart_task.go new file mode 100644 index 0000000000..31dbf153ed --- /dev/null +++ b/pkg/llm/tasks/llm/llm_deployment_restart_task.go @@ -0,0 +1,98 @@ +package llm + +import ( + "context" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + + api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" + "yunion.io/x/onecloud/pkg/llm/models" + "yunion.io/x/onecloud/pkg/util/logclient" +) + +// LLMDeploymentRestartTask restarts all restartable SLLM instances under a +// deployment in parallel, then reconciles replica health on the deployment row. +// +// Flow: +// 1. OnInit: enumerate instances, start each restartable instance's +// LLMRestartTask with this task as parent. +// 2. Framework invokes OnInstancesRestarted after all child tasks complete. +// 3. OnInstancesRestarted calls SyncReadyReplicas and completes. +type LLMDeploymentRestartTask struct { + taskman.STask +} + +func init() { + taskman.RegisterTask(LLMDeploymentRestartTask{}) +} + +func (task *LLMDeploymentRestartTask) taskFailed(ctx context.Context, model *models.SLLMDeployment, err error) { + db.OpsLog.LogEvent(model, "restart", err, task.UserCred) + logclient.AddActionLogWithStartable(task, model, logclient.ACT_VM_RESTART, err, task.UserCred, false) + task.SetStageFailed(ctx, jsonutils.NewString(err.Error())) +} + +func (task *LLMDeploymentRestartTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + model := obj.(*models.SLLMDeployment) + + instances, err := fetchModelInstances(model.Id) + if err != nil { + task.taskFailed(ctx, model, err) + return + } + if len(instances) == 0 { + task.taskFailed(ctx, model, errors.Error("no instances under deployment")) + return + } + + task.SetStage("OnInstancesRestarted", nil) + + startedCount := 0 + for _, inst := range instances { + llmObj, err := models.GetLLMManager().FetchById(inst.Id) + if err != nil { + log.Errorf("LLMDeploymentRestartTask: fetch instance %s: %s", inst.Id, err) + continue + } + llm := llmObj.(*models.SLLM) + taskInput, err := llm.ValidateRestartInput(ctx, task.UserCred, &api.LLMRestartInput{}) + if err != nil { + log.Warningf("LLMDeploymentRestartTask: skip instance %s: %s", inst.Id, err) + continue + } + if _, err := llm.StartRestartTask(ctx, task.UserCred, taskInput, task.GetTaskId()); err != nil { + log.Errorf("LLMDeploymentRestartTask: start restart for %s: %s", inst.Id, err) + continue + } + startedCount++ + } + + if startedCount == 0 { + task.taskFailed(ctx, model, errors.Error("no instance restart tasks could be started")) + } +} + +func (task *LLMDeploymentRestartTask) OnInstancesRestarted(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + model := obj.(*models.SLLMDeployment) + if err := model.SyncReadyReplicas(ctx, task.UserCred); err != nil { + log.Warningf("LLMDeploymentRestartTask: SyncReadyReplicas for %s: %s", model.Name, err) + } + db.OpsLog.LogEvent(model, "restart", nil, task.UserCred) + logclient.AddActionLogWithStartable(task, model, logclient.ACT_VM_RESTART, nil, task.UserCred, true) + task.SetStageComplete(ctx, nil) +} + +func (task *LLMDeploymentRestartTask) OnInstancesRestartedFailed(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + model := obj.(*models.SLLMDeployment) + log.Warningf("LLMDeploymentRestartTask: some instances failed to restart: %s", body) + if err := model.SyncReadyReplicas(ctx, task.UserCred); err != nil { + log.Warningf("LLMDeploymentRestartTask: SyncReadyReplicas for %s: %s", model.Name, err) + } + db.OpsLog.LogEvent(model, "restart", body, task.UserCred) + logclient.AddActionLogWithStartable(task, model, logclient.ACT_VM_RESTART, body, task.UserCred, false) + task.SetStageComplete(ctx, nil) +} diff --git a/pkg/llm/tasks/llm/llm_deployment_sync_status_task.go b/pkg/llm/tasks/llm/llm_deployment_sync_status_task.go new file mode 100644 index 0000000000..93c26163e7 --- /dev/null +++ b/pkg/llm/tasks/llm/llm_deployment_sync_status_task.go @@ -0,0 +1,96 @@ +package llm + +import ( + "context" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" + "yunion.io/x/onecloud/pkg/llm/models" + "yunion.io/x/onecloud/pkg/util/logclient" +) + +// LLMDeploymentSyncStatusTask syncs all SLLM instances under a deployment in +// parallel, then reconciles replica health on the deployment row. +// +// Flow: +// 1. OnInit: enumerate instances, start each syncable instance's +// LLMSyncStatusTask with this task as parent. +// 2. Framework invokes OnInstancesSyncStatusComplete after all child tasks complete. +// 3. OnInstancesSyncStatusComplete calls SyncReadyReplicas and completes. +type LLMDeploymentSyncStatusTask struct { + taskman.STask +} + +func init() { + taskman.RegisterTask(LLMDeploymentSyncStatusTask{}) +} + +func (task *LLMDeploymentSyncStatusTask) taskFailed(ctx context.Context, model *models.SLLMDeployment, err error) { + db.OpsLog.LogEvent(model, db.ACT_SYNC_STATUS, err, task.UserCred) + logclient.AddActionLogWithStartable(task, model, logclient.ACT_SYNC_STATUS, err, task.UserCred, false) + task.SetStageFailed(ctx, jsonutils.NewString(err.Error())) +} + +func (task *LLMDeploymentSyncStatusTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + model := obj.(*models.SLLMDeployment) + + instances, err := fetchModelInstances(model.Id) + if err != nil { + task.taskFailed(ctx, model, err) + return + } + if len(instances) == 0 { + task.taskFailed(ctx, model, errors.Error("no instances under deployment")) + return + } + + task.SetStage("OnInstancesSyncStatusComplete", nil) + + startedCount := 0 + for _, inst := range instances { + llmObj, err := models.GetLLMManager().FetchById(inst.Id) + if err != nil { + log.Errorf("LLMDeploymentSyncStatusTask: fetch instance %s: %s", inst.Id, err) + continue + } + llm := llmObj.(*models.SLLM) + if llm.CmpId == "" { + log.Warningf("LLMDeploymentSyncStatusTask: skip instance %s: no cmp_id", inst.Id) + continue + } + if err := llm.StartSyncStatusTask(ctx, task.UserCred, task.GetTaskId()); err != nil { + log.Errorf("LLMDeploymentSyncStatusTask: start syncstatus for %s: %s", inst.Id, err) + continue + } + startedCount++ + } + + if startedCount == 0 { + task.taskFailed(ctx, model, errors.Error("no instance syncstatus tasks could be started")) + } +} + +func (task *LLMDeploymentSyncStatusTask) OnInstancesSyncStatusComplete(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + model := obj.(*models.SLLMDeployment) + if err := model.SyncReadyReplicas(ctx, task.UserCred); err != nil { + log.Warningf("LLMDeploymentSyncStatusTask: SyncReadyReplicas for %s: %s", model.Name, err) + } + db.OpsLog.LogEvent(model, db.ACT_SYNC_STATUS, nil, task.UserCred) + logclient.AddActionLogWithStartable(task, model, logclient.ACT_SYNC_STATUS, nil, task.UserCred, true) + task.SetStageComplete(ctx, nil) +} + +func (task *LLMDeploymentSyncStatusTask) OnInstancesSyncStatusCompleteFailed(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { + model := obj.(*models.SLLMDeployment) + log.Warningf("LLMDeploymentSyncStatusTask: some instances failed to sync status: %s", body) + if err := model.SyncReadyReplicas(ctx, task.UserCred); err != nil { + log.Warningf("LLMDeploymentSyncStatusTask: SyncReadyReplicas for %s: %s", model.Name, err) + } + db.OpsLog.LogEvent(model, db.ACT_SYNC_STATUS, body, task.UserCred) + logclient.AddActionLogWithStartable(task, model, logclient.ACT_SYNC_STATUS, body, task.UserCred, false) + task.SetStageComplete(ctx, nil) +} diff --git a/pkg/llm/tasks/llm/llm_sync_status_task.go b/pkg/llm/tasks/llm/llm_sync_status_task.go index 5f11f43843..c00b4fda9a 100644 --- a/pkg/llm/tasks/llm/llm_sync_status_task.go +++ b/pkg/llm/tasks/llm/llm_sync_status_task.go @@ -32,8 +32,15 @@ func (task *LLMSyncStatusTask) setLLMStatus(ctx context.Context, llm *models.SLL } } +// applyResolvedLLMStatus persists the reconciled LLM status after sync completes. +// Unlike setLLMStatus, this always writes even when invoked as a child task (e.g. +// LLMDeploymentSyncStatusTask) so derived statuses such as ready are not dropped. +func (task *LLMSyncStatusTask) applyResolvedLLMStatus(ctx context.Context, llm *models.SLLM, status string, reason string) { + llm.SetStatus(ctx, task.UserCred, status, reason) +} + func (task *LLMSyncStatusTask) taskFailed(ctx context.Context, llm *models.SLLM, err string) { - task.setLLMStatus(ctx, llm, computeapi.VM_SYNC_FAIL, err) + task.applyResolvedLLMStatus(ctx, llm, computeapi.VM_SYNC_FAIL, err) db.OpsLog.LogEvent(llm, db.ACT_SYNC_STATUS, err, task.UserCred) logclient.AddActionLogWithStartable(task, llm, logclient.ACT_SYNC_STATUS, err, task.UserCred, false) // llm.NotifyRequest(ctx, task.GetUserCred(), notify.ActionStart, nil, false) @@ -107,14 +114,14 @@ func (task *LLMSyncStatusTask) OnInit(ctx context.Context, obj db.IStandaloneMod if err != nil { return nil, errors.Wrap(err, "WaitServerStatus") } - task.setLLMStatus(ctx, llm, srv.Status, "stop server") + task.applyResolvedLLMStatus(ctx, llm, srv.Status, "stop server") } else { resolved, err := models.ResolveLLMStatusFromServerDetails(ctx, llm, srv) if err != nil { return nil, errors.Wrap(err, "ResolveLLMStatusFromServerDetails") } if resolved.Update { - task.setLLMStatus(ctx, llm, resolved.Status, resolved.Reason) + task.applyResolvedLLMStatus(ctx, llm, resolved.Status, resolved.Reason) } } @@ -122,7 +129,7 @@ func (task *LLMSyncStatusTask) OnInit(ctx context.Context, obj db.IStandaloneMod if volume != nil { disk, err := volume.GetDisk(ctx) if err != nil { - task.setLLMStatus(ctx, llm, computeapi.VM_DISK_RESET_FAIL, errors.Wrap(err, "GetDisk").Error()) + task.applyResolvedLLMStatus(ctx, llm, computeapi.VM_DISK_RESET_FAIL, errors.Wrap(err, "GetDisk").Error()) return nil, errors.Wrap(err, "GetDisk") } if disk.Status != computeapi.DISK_READY { diff --git a/pkg/mcclient/options/llm/llm_deployment.go b/pkg/mcclient/options/llm/llm_deployment.go index 4cb6c7f940..59a52492ab 100644 --- a/pkg/mcclient/options/llm/llm_deployment.go +++ b/pkg/mcclient/options/llm/llm_deployment.go @@ -290,6 +290,30 @@ func (o *LLMDeploymentUnregisterAiproxyOptions) Params() (jsonutils.JSONObject, return jsonutils.NewDict(), nil } +type LLMDeploymentRestartOptions struct { + options.BaseIdOptions +} + +func (o *LLMDeploymentRestartOptions) GetId() string { + return o.ID +} + +func (o *LLMDeploymentRestartOptions) Params() (jsonutils.JSONObject, error) { + return jsonutils.NewDict(), nil +} + +type LLMDeploymentSyncstatusOptions struct { + options.BaseIdOptions +} + +func (o *LLMDeploymentSyncstatusOptions) GetId() string { + return o.ID +} + +func (o *LLMDeploymentSyncstatusOptions) Params() (jsonutils.JSONObject, error) { + return jsonutils.NewDict(), nil +} + func applyGpuUtilizationAlias(params *jsonutils.JSONDict, gpuMemoryUtilization, gpuUtilization *float64) error { if gpuMemoryUtilization != nil && gpuUtilization != nil { return fmt.Errorf("--gpu-memory-utilization and --gpu-utilization are aliases; specify only one") diff --git a/pkg/util/logclient/consts.go b/pkg/util/logclient/consts.go index 3d082e4721..3d0dfbfaa5 100644 --- a/pkg/util/logclient/consts.go +++ b/pkg/util/logclient/consts.go @@ -318,4 +318,7 @@ const ( ACT_CLONE = "clone" ACT_REBUILD = "rebuild" + + ACT_REGISTER_AIPROXY = "register_aiproxy" + ACT_UNREGISTER_AIPROXY = "unregister_aiproxy" )