From da6ea51b5caf1d5dffab78fbfa909af0360d874b Mon Sep 17 00:00:00 2001 From: Zexi Li Date: Mon, 7 Sep 2026 21:14:00 +0800 Subject: [PATCH] feat(llm): support force restart for LLM and deployments (#25583) Allow force restart when instances are failed, protect restarting status during sync, refresh GPU memory from deployment, and tolerate crash-loop while waiting for service ready. --- pkg/apis/llm/llm.go | 2 + pkg/apis/llm/llm_deployment.go | 3 + pkg/llm/models/llm.go | 20 ++- pkg/llm/models/llm_deployment.go | 57 +++++-- pkg/llm/models/llm_deployment_gpu.go | 147 ++++++++++++++++++ .../llm_deployment_gpu_explicit_util_test.go | 42 +++++ pkg/llm/models/llm_deployment_restart_test.go | 69 ++++++-- pkg/llm/models/llm_service_ready.go | 45 +++++- pkg/llm/models/llm_service_ready_test.go | 123 +++++++++++++++ .../tasks/llm/llm_deployment_restart_task.go | 10 +- pkg/llm/tasks/llm/llm_restart_task.go | 32 +++- .../llm/llm_start_save_model_image_task.go | 2 +- pkg/llm/tasks/llm/llm_stop_task.go | 11 +- pkg/mcclient/options/llm/llm_deployment.go | 7 +- 14 files changed, 531 insertions(+), 39 deletions(-) create mode 100644 pkg/llm/models/llm_service_ready_test.go diff --git a/pkg/apis/llm/llm.go b/pkg/apis/llm/llm.go index 4a525c2f72..6a2fc01e3b 100644 --- a/pkg/apis/llm/llm.go +++ b/pkg/apis/llm/llm.go @@ -205,6 +205,7 @@ type LLMSyncStatusInput struct { } type LLMRestartInput struct { + Force bool `json:"force"` } type LLMRestartTaskInput struct { @@ -218,6 +219,7 @@ type LLMRestartTaskInput struct { RebindVolumeId string OnlyStop bool + Force bool `json:"force"` } type LLMChangeNetworkInput struct { diff --git a/pkg/apis/llm/llm_deployment.go b/pkg/apis/llm/llm_deployment.go index 63b8935147..ab8f22164b 100644 --- a/pkg/apis/llm/llm_deployment.go +++ b/pkg/apis/llm/llm_deployment.go @@ -87,6 +87,8 @@ const ( LLM_DEPLOYMENT_STATUS_PARTIAL = "partial" // Replica reconcile or syncstatus in progress. LLM_DEPLOYMENT_STATUS_SYNCING = "syncing" + // Restart of all replicas in progress. + LLM_DEPLOYMENT_STATUS_RESTARTING = "restarting" ) // AiproxySyncStatus values stored on SLLMDeployment.AiproxySyncStatus. @@ -183,6 +185,7 @@ type LLMDeploymentUpdateInput struct { } type LLMDeploymentRestartInput struct { + Force bool `json:"force"` } type LLMDeploymentSyncstatusInput struct { diff --git a/pkg/llm/models/llm.go b/pkg/llm/models/llm.go index 3e5032bdd0..d7f38441bb 100644 --- a/pkg/llm/models/llm.go +++ b/pkg/llm/models/llm.go @@ -785,7 +785,7 @@ func (llm *SLLM) PerformStop(ctx context.Context, userCred mcclient.TokenCredent return nil, errors.Wrapf(errors.ErrInvalidStatus, "llm id: %s status: %s", llm.Id, llm.Status) } llm.SetStatus(ctx, userCred, computeapi.VM_START_STOP, "perform stop") - err := llm.StartLLMStopTask(ctx, userCred, "") + err := llm.StartLLMStopTask(ctx, userCred, "", false) if err != nil { return nil, errors.Wrap(err, "StartStopTask") } @@ -793,6 +793,9 @@ func (llm *SLLM) PerformStop(ctx context.Context, userCred mcclient.TokenCredent } func (llm *SLLM) ValidateRestartInput(ctx context.Context, userCred mcclient.TokenCredential, input *api.LLMRestartInput) (*api.LLMRestartTaskInput, error) { + if input == nil { + input = &api.LLMRestartInput{} + } if len(llm.CmpId) == 0 { return nil, errors.Wrap(errors.ErrInvalidStatus, "empty cmp_id") } @@ -802,8 +805,10 @@ func (llm *SLLM) ValidateRestartInput(ctx context.Context, userCred mcclient.Tok return nil, errors.Wrap(err, "GetServer") } - if (llm.Status != api.LLM_STATUS_READY && llm.Status != api.LLM_STATUS_RUNNING) || (srv.Status != computeapi.VM_READY && !utils.IsInArray(srv.Status, computeapi.VM_RUNNING_STATUS)) { - return nil, errors.Wrapf(errors.ErrInvalidStatus, "invalid llm status %s", llm.Status) + if !input.Force { + if (llm.Status != api.LLM_STATUS_READY && llm.Status != api.LLM_STATUS_RUNNING) || (srv.Status != computeapi.VM_READY && !utils.IsInArray(srv.Status, computeapi.VM_RUNNING_STATUS)) { + return nil, errors.Wrapf(errors.ErrInvalidStatus, "invalid llm status %s", llm.Status) + } } sku, err := llm.GetLLMSku(llm.LLMSkuId) @@ -813,6 +818,7 @@ func (llm *SLLM) ValidateRestartInput(ctx context.Context, userCred mcclient.Tok return &api.LLMRestartTaskInput{ ImageId: sku.GetLLMImageId(), + Force: input.Force, }, nil } @@ -889,8 +895,12 @@ func (llm *SLLM) NotifyRequest(ctx context.Context, userCred mcclient.TokenCrede }) } -func (llm *SLLM) StartLLMStopTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error { - task, err := taskman.TaskManager.NewTask(ctx, "LLMStopTask", llm, userCred, nil, parentTaskId, "", nil) +func (llm *SLLM) StartLLMStopTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string, force bool) error { + params := jsonutils.NewDict() + if force { + params.Set("force", jsonutils.JSONTrue) + } + task, err := taskman.TaskManager.NewTask(ctx, "LLMStopTask", llm, userCred, params, parentTaskId, "", nil) if err != nil { return errors.Wrap(err, "NewTask") } diff --git a/pkg/llm/models/llm_deployment.go b/pkg/llm/models/llm_deployment.go index 182dd7a450..a49b72c690 100644 --- a/pkg/llm/models/llm_deployment.go +++ b/pkg/llm/models/llm_deployment.go @@ -434,6 +434,9 @@ func (model *SLLMDeployment) RealDelete(ctx context.Context, userCred mcclient.T type SyncReadyReplicasOptions struct { // SkipAiproxySync avoids scheduling LLMAiproxySyncTask (e.g. create task uses a child sync instead). SkipAiproxySync bool + // ForceHealthStatus applies replica-health status even when the current + // value would otherwise be protected (e.g. restarting after the restart task). + ForceHealthStatus bool } // SyncReadyReplicas recomputes ReadyReplicas from SLLM instances, persists it @@ -453,8 +456,10 @@ type SyncReadyReplicasOptions struct { // 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 { skipAiproxySync := false + forceHealthStatus := false if len(opts) > 0 { skipAiproxySync = opts[0].SkipAiproxySync + forceHealthStatus = opts[0].ForceHealthStatus } var rows []deploymentReplicaStatusRow err := GetLLMManager().Query("status"). @@ -477,7 +482,7 @@ func (model *SLLMDeployment) SyncReadyReplicas(ctx context.Context, userCred mcc if desired == "" { return nil } - if !canUpdateReplicaHealthStatus(model.Status, desired) { + if !forceHealthStatus && !canUpdateReplicaHealthStatus(model.Status, desired) { return nil } oldStatus := model.Status @@ -562,10 +567,14 @@ 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, except create_fail and start_fail may -// recover when replicas are running again. +// override the current deployment value. Early lifecycle, restarting, and +// terminal failure / delete states must not be clobbered, except create_fail +// and start_fail may recover when replicas are running again. Restarting is +// cleared only by SyncReadyReplicas with ForceHealthStatus after the restart task. func canUpdateReplicaHealthStatus(current, desired string) bool { + if current == api.LLM_DEPLOYMENT_STATUS_RESTARTING { + return false + } if current == api.LLM_STATUS_CREATE_FAIL { return desired == api.STATUS_READY || desired == api.LLM_DEPLOYMENT_STATUS_PARTIAL } @@ -740,7 +749,23 @@ func (model *SLLMDeployment) PerformUnregisterAiproxy( } // canRestartDeploymentStatus reports whether a deployment may be restarted. -func canRestartDeploymentStatus(status string) bool { +// force relaxes the allowlist so failed/deploying deployments can still apply SKU updates. +func canRestartDeploymentStatus(status string, force bool) 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, + api.LLM_DEPLOYMENT_STATUS_RESTARTING: + return false + } + if force { + return true + } switch status { case api.STATUS_READY, api.LLM_DEPLOYMENT_STATUS_PARTIAL, @@ -755,7 +780,10 @@ func (model *SLLMDeployment) ValidateRestartInput( userCred mcclient.TokenCredential, input *api.LLMDeploymentRestartInput, ) error { - if !canRestartDeploymentStatus(model.Status) { + if input == nil { + input = &api.LLMDeploymentRestartInput{} + } + if !canRestartDeploymentStatus(model.Status, input.Force) { return httperrors.NewInvalidStatusError("invalid deployment status %s", model.Status) } var rows []struct { @@ -769,13 +797,14 @@ func (model *SLLMDeployment) ValidateRestartInput( return httperrors.NewInvalidStatusError("no instances under deployment") } restartable := 0 + instInput := &api.LLMRestartInput{Force: input.Force} 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 { + if _, err := llm.ValidateRestartInput(ctx, userCred, instInput); err == nil { restartable++ } } @@ -791,17 +820,25 @@ func (model *SLLMDeployment) PerformRestart( query jsonutils.JSONObject, input *api.LLMDeploymentRestartInput, ) (jsonutils.JSONObject, error) { + if input == nil { + input = &api.LLMDeploymentRestartInput{} + } if err := model.ValidateRestartInput(ctx, userCred, input); err != nil { return nil, err } - if err := model.StartRestartTask(ctx, userCred); err != nil { + if err := model.StartRestartTask(ctx, userCred, input.Force); 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) +func (model *SLLMDeployment) StartRestartTask(ctx context.Context, userCred mcclient.TokenCredential, force bool) error { + model.SetStatus(ctx, userCred, api.LLM_DEPLOYMENT_STATUS_RESTARTING, "") + params := jsonutils.NewDict() + if force { + params.Set("force", jsonutils.JSONTrue) + } + task, err := taskman.TaskManager.NewTask(ctx, "LLMDeploymentRestartTask", model, userCred, params, "", "", nil) if err != nil { return errors.Wrap(err, "NewTask LLMDeploymentRestartTask") } diff --git a/pkg/llm/models/llm_deployment_gpu.go b/pkg/llm/models/llm_deployment_gpu.go index 5215a526de..fce5a1e180 100644 --- a/pkg/llm/models/llm_deployment_gpu.go +++ b/pkg/llm/models/llm_deployment_gpu.go @@ -13,6 +13,7 @@ import ( computeapi "yunion.io/x/onecloud/pkg/apis/compute" api "yunion.io/x/onecloud/pkg/apis/llm" + "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/llm/options" "yunion.io/x/onecloud/pkg/llm/utils/vram" @@ -616,6 +617,152 @@ func BuildDeploymentResolvedGpuMemoryLLMSpec(ctx context.Context, userCred mccli return buildAutoGpuMemoryUtilizationLLMSpec(sku, utilization) } +func cloneLLMSpec(spec *api.LLMSpec) *api.LLMSpec { + if spec == nil { + return nil + } + copied := &api.LLMSpec{} + if err := jsonutils.Marshal(spec).Unmarshal(copied); err != nil { + return spec + } + return copied +} + +func stripLLMSpecCustomizedArg(spec *api.LLMSpec, llmType string, key string) { + if spec == nil || key == "" { + return + } + switch api.LLMContainerType(llmType) { + case api.LLM_CONTAINER_VLLM: + if spec.Vllm == nil { + return + } + spec.Vllm.CustomizedArgs = filterVllmCustomizedArgs(spec.Vllm.CustomizedArgs, key) + case api.LLM_CONTAINER_SGLANG: + if spec.SGLang == nil { + return + } + spec.SGLang.CustomizedArgs = filterSGLangCustomizedArgs(spec.SGLang.CustomizedArgs, key) + } +} + +func filterVllmCustomizedArgs(args []*api.VllmCustomizedArg, key string) []*api.VllmCustomizedArg { + if len(args) == 0 { + return args + } + out := make([]*api.VllmCustomizedArg, 0, len(args)) + for _, arg := range args { + if arg == nil || runtimeArgKeyIn(arg.Key, []string{key}) { + continue + } + out = append(out, arg) + } + return out +} + +func filterSGLangCustomizedArgs(args []*api.SGLangCustomizedArg, key string) []*api.SGLangCustomizedArg { + if len(args) == 0 { + return args + } + out := make([]*api.SGLangCustomizedArg, 0, len(args)) + for _, arg := range args { + if arg == nil || runtimeArgKeyIn(arg.Key, []string{key}) { + continue + } + out = append(out, arg) + } + return out +} + +func mergeLLMSpecCustomizedArgs(dst *api.LLMSpec, src *api.LLMSpec, llmType string) { + if dst == nil || src == nil { + return + } + switch api.LLMContainerType(llmType) { + case api.LLM_CONTAINER_VLLM: + if src.Vllm == nil { + return + } + if dst.Vllm == nil { + dst.Vllm = &api.LLMSpecVllm{} + } + for _, arg := range src.Vllm.CustomizedArgs { + if arg == nil { + continue + } + dst.Vllm.CustomizedArgs = filterVllmCustomizedArgs(dst.Vllm.CustomizedArgs, arg.Key) + next := *arg + dst.Vllm.CustomizedArgs = append(dst.Vllm.CustomizedArgs, &next) + } + case api.LLM_CONTAINER_SGLANG: + if src.SGLang == nil { + return + } + if dst.SGLang == nil { + dst.SGLang = &api.LLMSpecSGLang{} + } + for _, arg := range src.SGLang.CustomizedArgs { + if arg == nil { + continue + } + dst.SGLang.CustomizedArgs = filterSGLangCustomizedArgs(dst.SGLang.CustomizedArgs, arg.Key) + next := *arg + dst.SGLang.CustomizedArgs = append(dst.SGLang.CustomizedArgs, &next) + } + } +} + +func applyResolvedGpuMemoryLLMSpec(current *api.LLMSpec, resolved *api.LLMSpec, llmType string) *api.LLMSpec { + key, ok := gpuMemoryUtilizationRuntimeArgKey(llmType) + if !ok { + return current + } + out := cloneLLMSpec(current) + if out == nil { + out = &api.LLMSpec{} + } + stripLLMSpecCustomizedArg(out, llmType, key) + if resolved != nil { + mergeLLMSpecCustomizedArgs(out, resolved, llmType) + } + if out.IsZero() { + return nil + } + return out +} + +// RefreshLLMGpuMemorySpecFromDeployment rewrites instance LLMSpec GPU util args from the +// current deployment+SKU rules so restart applies the latest SKU backend parameters. +func RefreshLLMGpuMemorySpecFromDeployment(ctx context.Context, userCred mcclient.TokenCredential, llm *SLLM, sku *SLLMSku) error { + if llm == nil || llm.LLMDeploymentId == "" { + return nil + } + depObj, err := GetLLMDeploymentManager().FetchById(llm.LLMDeploymentId) + if err != nil { + return errors.Wrap(err, "fetch deployment") + } + deploy := depObj.(*SLLMDeployment) + if sku == nil { + sku, err = llm.GetLLMSku(llm.LLMSkuId) + if err != nil { + return errors.Wrap(err, "GetLLMSku") + } + } + resolved, err := BuildDeploymentResolvedGpuMemoryLLMSpec(ctx, userCred, deploy, sku) + if err != nil { + return err + } + next := applyResolvedGpuMemoryLLMSpec(llm.LLMSpec, resolved, sku.LLMType) + _, err = db.Update(llm, func() error { + llm.LLMSpec = next + return nil + }) + if err != nil { + return errors.Wrap(err, "update llm_spec gpu memory args") + } + return nil +} + func maxMountedModelVramRequirementMB(sku *SLLMSku) (int64, error) { if sku != nil && SkuHasLocalHostPathModel(sku) { return 0, httperrors.NewInputParameterError( diff --git a/pkg/llm/models/llm_deployment_gpu_explicit_util_test.go b/pkg/llm/models/llm_deployment_gpu_explicit_util_test.go index ffed69f885..0de3c7b3e8 100644 --- a/pkg/llm/models/llm_deployment_gpu_explicit_util_test.go +++ b/pkg/llm/models/llm_deployment_gpu_explicit_util_test.go @@ -61,3 +61,45 @@ func TestBuildDeploymentResolvedGpuMemoryLLMSpecSkipsExplicitSkuArg(t *testing.T t.Fatalf("expected nil spec when SKU already sets gpu-memory-utilization, got %#v", spec) } } + +func TestApplyResolvedGpuMemoryLLMSpecStripsStaleAutoArg(t *testing.T) { + current := &api.LLMSpec{ + Vllm: &api.LLMSpecVllm{ + PreferredModel: "qwen", + CustomizedArgs: []*api.VllmCustomizedArg{ + {Key: "gpu-memory-utilization", Value: "0.98"}, + {Key: "max-model-len", Value: "4096"}, + }, + }, + } + got := applyResolvedGpuMemoryLLMSpec(current, nil, string(api.LLM_CONTAINER_VLLM)) + if got == nil || got.Vllm == nil { + t.Fatal("expected remaining spec after stripping gpu util") + } + if got.Vllm.PreferredModel != "qwen" { + t.Fatalf("preferred model: %q", got.Vllm.PreferredModel) + } + if len(got.Vllm.CustomizedArgs) != 1 || got.Vllm.CustomizedArgs[0].Key != "max-model-len" { + t.Fatalf("customized args after strip: %#v", got.Vllm.CustomizedArgs) + } +} + +func TestApplyResolvedGpuMemoryLLMSpecWritesResolvedArg(t *testing.T) { + current := &api.LLMSpec{ + Vllm: &api.LLMSpecVllm{ + CustomizedArgs: []*api.VllmCustomizedArg{{Key: "gpu-memory-utilization", Value: "0.98"}}, + }, + } + resolved := &api.LLMSpec{ + Vllm: &api.LLMSpecVllm{ + CustomizedArgs: []*api.VllmCustomizedArg{{Key: "gpu-memory-utilization", Value: "0.9"}}, + }, + } + got := applyResolvedGpuMemoryLLMSpec(current, resolved, string(api.LLM_CONTAINER_VLLM)) + if got == nil || got.Vllm == nil || len(got.Vllm.CustomizedArgs) != 1 { + t.Fatalf("unexpected spec: %#v", got) + } + if got.Vllm.CustomizedArgs[0].Value != "0.9" { + t.Fatalf("expected resolved 0.9, got %q", got.Vllm.CustomizedArgs[0].Value) + } +} diff --git a/pkg/llm/models/llm_deployment_restart_test.go b/pkg/llm/models/llm_deployment_restart_test.go index 6b631aa3bd..c54c9575ee 100644 --- a/pkg/llm/models/llm_deployment_restart_test.go +++ b/pkg/llm/models/llm_deployment_restart_test.go @@ -9,23 +9,68 @@ import ( func TestCanRestartDeploymentStatus(t *testing.T) { cases := []struct { status string + force bool 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}, + {api.STATUS_READY, false, true}, + {api.LLM_DEPLOYMENT_STATUS_PARTIAL, false, true}, + {api.LLM_STATUS_RUNNING, false, true}, + {api.LLM_DEPLOYMENT_STATUS_DEPLOYING, false, false}, + {api.LLM_STATUS_DELETING, false, false}, + {api.LLM_DEPLOYMENT_STATUS_IMPORTING_MODEL, false, false}, + {api.LLM_DEPLOYMENT_STATUS_CREATING_SKU, false, false}, + {api.LLM_STATUS_CREATE_FAIL, false, false}, + {api.LLM_STATUS_DELETE_FAILED, false, false}, + {"unknown", false, false}, + {api.LLM_STATUS_START_FAIL, true, true}, + {api.LLM_STATUS_CREATE_FAIL, true, true}, + {api.LLM_DEPLOYMENT_STATUS_DEPLOYING, true, true}, + {api.LLM_STATUS_RESTART_FAILED, true, true}, + {api.LLM_STATUS_UNKNOWN, true, true}, + {api.STATUS_READY, true, true}, + {api.LLM_STATUS_DELETING, true, false}, + {api.LLM_STATUS_START_DELETE, true, false}, + {api.LLM_STATUS_DELETED, true, false}, + {api.LLM_STATUS_DELETE_FAILED, true, false}, + {"creating", true, false}, + {api.LLM_DEPLOYMENT_STATUS_IMPORTING_MODEL, true, false}, + {api.LLM_DEPLOYMENT_STATUS_CREATING_SKU, true, false}, + {api.LLM_DEPLOYMENT_STATUS_SYNCING, true, false}, + {api.LLM_DEPLOYMENT_STATUS_RESTARTING, false, false}, + {api.LLM_DEPLOYMENT_STATUS_RESTARTING, true, false}, } for _, tc := range cases { - got := canRestartDeploymentStatus(tc.status) + got := canRestartDeploymentStatus(tc.status, tc.force) if got != tc.want { - t.Errorf("canRestartDeploymentStatus(%q) = %v, want %v", tc.status, got, tc.want) + t.Errorf("canRestartDeploymentStatus(%q, force=%v) = %v, want %v", tc.status, tc.force, got, tc.want) + } + } +} + +func TestCanUpdateReplicaHealthStatus(t *testing.T) { + cases := []struct { + current string + desired string + want bool + }{ + {api.STATUS_READY, api.LLM_DEPLOYMENT_STATUS_PARTIAL, true}, + {api.LLM_DEPLOYMENT_STATUS_PARTIAL, api.STATUS_READY, true}, + {api.LLM_DEPLOYMENT_STATUS_DEPLOYING, api.STATUS_READY, true}, + {api.LLM_DEPLOYMENT_STATUS_RESTARTING, api.STATUS_READY, false}, + {api.LLM_DEPLOYMENT_STATUS_RESTARTING, api.LLM_DEPLOYMENT_STATUS_PARTIAL, false}, + {api.LLM_DEPLOYMENT_STATUS_RESTARTING, api.LLM_DEPLOYMENT_STATUS_DEPLOYING, false}, + {api.LLM_DEPLOYMENT_STATUS_RESTARTING, api.LLM_STATUS_START_FAIL, false}, + {api.LLM_STATUS_CREATE_FAIL, api.STATUS_READY, true}, + {api.LLM_STATUS_CREATE_FAIL, api.LLM_DEPLOYMENT_STATUS_DEPLOYING, false}, + {api.LLM_STATUS_START_FAIL, api.LLM_DEPLOYMENT_STATUS_PARTIAL, true}, + {api.LLM_STATUS_START_FAIL, api.LLM_DEPLOYMENT_STATUS_DEPLOYING, false}, + {api.LLM_DEPLOYMENT_STATUS_IMPORTING_MODEL, api.STATUS_READY, false}, + {api.LLM_STATUS_DELETING, api.STATUS_READY, false}, + } + for _, tc := range cases { + got := canUpdateReplicaHealthStatus(tc.current, tc.desired) + if got != tc.want { + t.Errorf("canUpdateReplicaHealthStatus(%q, %q) = %v, want %v", tc.current, tc.desired, got, tc.want) } } } diff --git a/pkg/llm/models/llm_service_ready.go b/pkg/llm/models/llm_service_ready.go index 816f7c2d20..bf0117b376 100644 --- a/pkg/llm/models/llm_service_ready.go +++ b/pkg/llm/models/llm_service_ready.go @@ -12,7 +12,9 @@ import ( "yunion.io/x/onecloud/pkg/mcclient" ) -const LLMServiceReadyTimeoutSeconds = 3600 +const LLMServiceReadyTimeoutSeconds = 1800 + +const llmCrashLoopFailThreshold = 3 var errLLMServiceProbing = errors.Error("llm service probing") @@ -44,13 +46,31 @@ func IsLLMServiceProbingError(err error) bool { return errors.Cause(err) == errLLMServiceProbing } +func isLLMCrashLoopStatus(status string) bool { + return status == computeapi.POD_STATUS_CRASH_LOOP_BACK_OFF || + status == computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF +} + func isLLMServiceFailedContainerStatus(status string) bool { return status == computeapi.CONTAINER_STATUS_PROBE_FAILED || status == computeapi.CONTAINER_STATUS_NET_FAILED || - status == computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF || status == computeapi.CONTAINER_STATUS_EXITED } +// crashLoopEpisodeAfter counts crash_loop_back_off episodes, not poll samples. +// The count increments only when entering crash_loop from a non-crash status. +// Staying in crash_loop does not increment. Leaving crash_loop keeps the count. +func crashLoopEpisodeAfter(episodes int, inCrashLoop bool, status string) (int, bool, bool) { + if !isLLMCrashLoopStatus(status) { + return episodes, false, false + } + if inCrashLoop { + return episodes, true, false + } + next := episodes + 1 + return next, true, next >= llmCrashLoopFailThreshold +} + func (llm *SLLM) WaitServiceReady(ctx context.Context, userCred mcclient.TokenCredential, timeoutSecs int) (*computeapi.SContainer, error) { return llm.WaitServiceReadyWithProbingCallback(ctx, userCred, timeoutSecs, nil) } @@ -64,7 +84,15 @@ func (llm *SLLM) WaitServiceReadyWithProbingCallback(ctx context.Context, userCr if err != nil { return nil, errors.Wrap(err, "WaitServerStatus") } - if server.Status != computeapi.VM_RUNNING { + crashLoopEpisodes := 0 + inCrashLoop := false + if isLLMCrashLoopStatus(server.Status) { + var failed bool + crashLoopEpisodes, inCrashLoop, failed = crashLoopEpisodeAfter(crashLoopEpisodes, inCrashLoop, server.Status) + if failed { + return nil, errors.Wrapf(errors.ErrInvalidStatus, "server status %s (crash_loop episodes %d)", server.Status, crashLoopEpisodes) + } + } else if server.Status != computeapi.VM_RUNNING { return nil, errors.Wrapf(errors.ErrInvalidStatus, "server status %s", server.Status) } @@ -84,6 +112,7 @@ func (llm *SLLM) WaitServiceReadyWithProbingCallback(ctx context.Context, userCr return ctr, nil } if ctr.Status == computeapi.CONTAINER_STATUS_PROBING { + crashLoopEpisodes, inCrashLoop, _ = crashLoopEpisodeAfter(crashLoopEpisodes, inCrashLoop, ctr.Status) if onProbing != nil && !probingNotified { if err := onProbing(); err != nil { return nil, errors.Wrap(err, "on probing") @@ -93,6 +122,16 @@ func (llm *SLLM) WaitServiceReadyWithProbingCallback(ctx context.Context, userCr time.Sleep(time.Second) continue } + if isLLMCrashLoopStatus(ctr.Status) { + var failed bool + crashLoopEpisodes, inCrashLoop, failed = crashLoopEpisodeAfter(crashLoopEpisodes, inCrashLoop, ctr.Status) + if failed { + return nil, errors.Wrapf(errors.ErrInvalidStatus, "container status %s (crash_loop episodes %d)", ctr.Status, crashLoopEpisodes) + } + time.Sleep(time.Second) + continue + } + crashLoopEpisodes, inCrashLoop, _ = crashLoopEpisodeAfter(crashLoopEpisodes, inCrashLoop, ctr.Status) if isLLMServiceFailedContainerStatus(ctr.Status) { return nil, errors.Wrapf(errors.ErrInvalidStatus, "container status %s", ctr.Status) } diff --git a/pkg/llm/models/llm_service_ready_test.go b/pkg/llm/models/llm_service_ready_test.go new file mode 100644 index 0000000000..360c7ff93b --- /dev/null +++ b/pkg/llm/models/llm_service_ready_test.go @@ -0,0 +1,123 @@ +package models + +import ( + "testing" + + computeapi "yunion.io/x/onecloud/pkg/apis/compute" +) + +func TestCrashLoopEpisodeAfter(t *testing.T) { + cases := []struct { + name string + episodes int + inCrashLoop bool + status string + wantEpisodes int + wantInCrash bool + wantFailed bool + }{ + { + name: "enter container crash_loop", + episodes: 0, + inCrashLoop: false, + status: computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF, + wantEpisodes: 1, + wantInCrash: true, + wantFailed: false, + }, + { + name: "stay in container crash_loop does not increment", + episodes: 1, + inCrashLoop: true, + status: computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF, + wantEpisodes: 1, + wantInCrash: true, + wantFailed: false, + }, + { + name: "pod crash_loop while already in episode does not increment", + episodes: 1, + inCrashLoop: true, + status: computeapi.POD_STATUS_CRASH_LOOP_BACK_OFF, + wantEpisodes: 1, + wantInCrash: true, + wantFailed: false, + }, + { + name: "running leaves crash_loop and keeps count", + episodes: 1, + inCrashLoop: true, + status: computeapi.CONTAINER_STATUS_RUNNING, + wantEpisodes: 1, + wantInCrash: false, + wantFailed: false, + }, + { + name: "probing leaves crash_loop and keeps count", + episodes: 2, + inCrashLoop: true, + status: computeapi.CONTAINER_STATUS_PROBING, + wantEpisodes: 2, + wantInCrash: false, + wantFailed: false, + }, + { + name: "third episode fails", + episodes: 2, + inCrashLoop: false, + status: computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF, + wantEpisodes: 3, + wantInCrash: true, + wantFailed: true, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + gotEpisodes, gotInCrash, gotFailed := crashLoopEpisodeAfter(tc.episodes, tc.inCrashLoop, tc.status) + if gotEpisodes != tc.wantEpisodes || gotInCrash != tc.wantInCrash || gotFailed != tc.wantFailed { + t.Fatalf("crashLoopEpisodeAfter(%d, %v, %q) = (%d, %v, %v), want (%d, %v, %v)", + tc.episodes, tc.inCrashLoop, tc.status, + gotEpisodes, gotInCrash, gotFailed, + tc.wantEpisodes, tc.wantInCrash, tc.wantFailed) + } + }) + } +} + +func TestCrashLoopEpisodeAfterSingleCrashThenRunning(t *testing.T) { + episodes, inCrash, failed := crashLoopEpisodeAfter(0, false, computeapi.POD_STATUS_CRASH_LOOP_BACK_OFF) + if failed || episodes != 1 || !inCrash { + t.Fatalf("server crash_loop: episodes=%d inCrash=%v failed=%v", episodes, inCrash, failed) + } + episodes, inCrash, failed = crashLoopEpisodeAfter(episodes, inCrash, computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF) + if failed || episodes != 1 || !inCrash { + t.Fatalf("same crash still in loop: episodes=%d inCrash=%v failed=%v", episodes, inCrash, failed) + } + episodes, inCrash, failed = crashLoopEpisodeAfter(episodes, inCrash, computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF) + if failed || episodes != 1 { + t.Fatalf("poll again should not fail: episodes=%d failed=%v", episodes, failed) + } + episodes, inCrash, failed = crashLoopEpisodeAfter(episodes, inCrash, computeapi.CONTAINER_STATUS_RUNNING) + if failed || episodes != 1 || inCrash { + t.Fatalf("recover to running: episodes=%d inCrash=%v failed=%v", episodes, inCrash, failed) + } +} + +func TestCrashLoopEpisodeAfterThreeCyclesFail(t *testing.T) { + episodes, inCrash, failed := 0, false, false + for cycle := 1; cycle <= llmCrashLoopFailThreshold; cycle++ { + episodes, inCrash, failed = crashLoopEpisodeAfter(episodes, inCrash, computeapi.CONTAINER_STATUS_CRASH_LOOP_BACK_OFF) + if cycle < llmCrashLoopFailThreshold && failed { + t.Fatalf("cycle %d should not fail: episodes=%d", cycle, episodes) + } + if cycle < llmCrashLoopFailThreshold { + episodes, inCrash, failed = crashLoopEpisodeAfter(episodes, inCrash, computeapi.CONTAINER_STATUS_PROBING) + if failed || inCrash { + t.Fatalf("recover after cycle %d: episodes=%d inCrash=%v failed=%v", cycle, episodes, inCrash, failed) + } + } + } + if !failed || episodes != llmCrashLoopFailThreshold { + t.Fatalf("third crash cycle: episodes=%d failed=%v", episodes, failed) + } +} diff --git a/pkg/llm/tasks/llm/llm_deployment_restart_task.go b/pkg/llm/tasks/llm/llm_deployment_restart_task.go index 31dbf153ed..e6ed23f470 100644 --- a/pkg/llm/tasks/llm/llm_deployment_restart_task.go +++ b/pkg/llm/tasks/llm/llm_deployment_restart_task.go @@ -31,6 +31,9 @@ func init() { } func (task *LLMDeploymentRestartTask) taskFailed(ctx context.Context, model *models.SLLMDeployment, err error) { + if syncErr := model.SyncReadyReplicas(ctx, task.UserCred, models.SyncReadyReplicasOptions{ForceHealthStatus: true}); syncErr != nil { + log.Warningf("LLMDeploymentRestartTask: SyncReadyReplicas for %s: %s", model.Name, syncErr) + } 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())) @@ -59,7 +62,8 @@ func (task *LLMDeploymentRestartTask) OnInit(ctx context.Context, obj db.IStanda continue } llm := llmObj.(*models.SLLM) - taskInput, err := llm.ValidateRestartInput(ctx, task.UserCred, &api.LLMRestartInput{}) + force := jsonutils.QueryBoolean(task.GetParams(), "force", false) + taskInput, err := llm.ValidateRestartInput(ctx, task.UserCred, &api.LLMRestartInput{Force: force}) if err != nil { log.Warningf("LLMDeploymentRestartTask: skip instance %s: %s", inst.Id, err) continue @@ -78,7 +82,7 @@ func (task *LLMDeploymentRestartTask) OnInit(ctx context.Context, obj db.IStanda 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 { + if err := model.SyncReadyReplicas(ctx, task.UserCred, models.SyncReadyReplicasOptions{ForceHealthStatus: true}); err != nil { log.Warningf("LLMDeploymentRestartTask: SyncReadyReplicas for %s: %s", model.Name, err) } db.OpsLog.LogEvent(model, "restart", nil, task.UserCred) @@ -89,7 +93,7 @@ func (task *LLMDeploymentRestartTask) OnInstancesRestarted(ctx context.Context, 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 { + if err := model.SyncReadyReplicas(ctx, task.UserCred, models.SyncReadyReplicasOptions{ForceHealthStatus: true}); err != nil { log.Warningf("LLMDeploymentRestartTask: SyncReadyReplicas for %s: %s", model.Name, err) } db.OpsLog.LogEvent(model, "restart", body, task.UserCred) diff --git a/pkg/llm/tasks/llm/llm_restart_task.go b/pkg/llm/tasks/llm/llm_restart_task.go index 6adb59e6d0..f449ed32d7 100644 --- a/pkg/llm/tasks/llm/llm_restart_task.go +++ b/pkg/llm/tasks/llm/llm_restart_task.go @@ -90,6 +90,16 @@ func (task *LLMRestartTask) OnSyncLLMInitStatusCompleteFailed(ctx context.Contex task.taskFailed(ctx, llm, err.String()) } +func (task *LLMRestartTask) restartForce() bool { + params := task.GetParams() + if params == nil { + return false + } + input := api.LLMRestartTaskInput{} + params.Unmarshal(&input) + return input.Force +} + func (task *LLMRestartTask) OnSyncLLMInitStatusComplete(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { llm := obj.(*models.SLLM) @@ -99,15 +109,27 @@ func (task *LLMRestartTask) OnSyncLLMInitStatusComplete(ctx context.Context, obj return } + if srv.Status == computeapi.VM_READY { + task.OnServerStopComplete(ctx, llm, nil) + return + } + + force := task.restartForce() + if force { + task.SetStage("OnServerStopComplete", nil) + if err := llm.StartLLMStopTask(ctx, task.UserCred, task.GetTaskId(), true); err != nil { + task.taskFailed(ctx, llm, errors.Wrap(err, "StartLLMStopTask").Error()) + } + return + } + switch srv.Status { case computeapi.VM_RUNNING: task.SetStage("OnServerStopComplete", nil) - if err := llm.StartLLMStopTask(ctx, task.UserCred, task.GetTaskId()); err != nil { + if err := llm.StartLLMStopTask(ctx, task.UserCred, task.GetTaskId(), false); err != nil { task.taskFailed(ctx, llm, errors.Wrap(err, "StartLLMStopTask").Error()) return } - case computeapi.VM_READY: - task.OnServerStopComplete(ctx, llm, nil) default: if strings.Contains(srv.Status, "fail") { task.taskFailed(ctx, llm, errors.Wrap(errors.ErrInvalidStatus, srv.Status).Error()) @@ -149,6 +171,10 @@ func (task *LLMRestartTask) OnServerStopComplete(ctx context.Context, obj db.ISt task.taskFailed(ctx, llm, errors.Wrap(err, "GetLLMSku").Error()) return } + if err := models.RefreshLLMGpuMemorySpecFromDeployment(ctx, task.UserCred, llm, sku); err != nil { + task.taskFailed(ctx, llm, errors.Wrap(err, "RefreshLLMGpuMemorySpecFromDeployment").Error()) + return + } if len(server.Nics) > 0 { // check bandwidth diff --git a/pkg/llm/tasks/llm/llm_start_save_model_image_task.go b/pkg/llm/tasks/llm/llm_start_save_model_image_task.go index ce9cdef387..37b51ab49c 100644 --- a/pkg/llm/tasks/llm/llm_start_save_model_image_task.go +++ b/pkg/llm/tasks/llm/llm_start_save_model_image_task.go @@ -34,7 +34,7 @@ func (task *LLMStartSaveModelImageTask) OnInit(ctx context.Context, obj db.IStan // first stop the desktop task.SetStage("OnStopLLMComplete", nil) - err := llm.StartLLMStopTask(ctx, task.UserCred, task.GetTaskId()) + err := llm.StartLLMStopTask(ctx, task.UserCred, task.GetTaskId(), false) if err != nil { task.taskFailed(ctx, llm, err.Error()) return diff --git a/pkg/llm/tasks/llm/llm_stop_task.go b/pkg/llm/tasks/llm/llm_stop_task.go index 5b147affe8..444e48fc75 100644 --- a/pkg/llm/tasks/llm/llm_stop_task.go +++ b/pkg/llm/tasks/llm/llm_stop_task.go @@ -56,8 +56,17 @@ func (task *LLMStopTask) OnInit(ctx context.Context, obj db.IStandaloneModel, bo task.SetStage("OnStopComplete", nil) s := auth.GetSession(ctx, task.UserCred, "") + force := jsonutils.QueryBoolean(task.GetParams(), "force", false) err = s.WithTaskCallback(task.GetId(), func() error { - _, err = compute.Servers.PerformAction(s, llm.CmpId, "stop", nil) + var params jsonutils.JSONObject + if force { + timeout := 10 + params = jsonutils.Marshal(computeapi.ServerStopInput{ + IsForce: true, + TimeoutSecs: &timeout, + }) + } + _, err = compute.Servers.PerformAction(s, llm.CmpId, "stop", params) return err }) if err != nil { diff --git a/pkg/mcclient/options/llm/llm_deployment.go b/pkg/mcclient/options/llm/llm_deployment.go index 32c505748e..53ea5d305f 100644 --- a/pkg/mcclient/options/llm/llm_deployment.go +++ b/pkg/mcclient/options/llm/llm_deployment.go @@ -292,6 +292,7 @@ func (o *LLMDeploymentUnregisterAiproxyOptions) Params() (jsonutils.JSONObject, type LLMDeploymentRestartOptions struct { options.BaseIdOptions + Force bool `help:"force restart even if instances are in failed status" json:"force"` } func (o *LLMDeploymentRestartOptions) GetId() string { @@ -299,7 +300,11 @@ func (o *LLMDeploymentRestartOptions) GetId() string { } func (o *LLMDeploymentRestartOptions) Params() (jsonutils.JSONObject, error) { - return jsonutils.NewDict(), nil + params := jsonutils.NewDict() + if o.Force { + params.Set("force", jsonutils.JSONTrue) + } + return params, nil } type LLMDeploymentSyncstatusOptions struct {