From bf4f1755c8a9ac6d329867eeb3ee1dcda19061ae Mon Sep 17 00:00:00 2001 From: cwz_eikoh Date: Fri, 7 Nov 2025 10:19:37 +0800 Subject: [PATCH] fix(llm): don't rely on polling (#23709) --- pkg/llm/models/dify.go | 4 +- pkg/llm/models/dify_containers.go | 29 ++++++ pkg/llm/models/llm.go | 4 +- pkg/llm/models/llm_base.go | 6 +- pkg/llm/tasks/dify/dify_create_task.go | 53 +++++----- pkg/llm/tasks/dify/dify_delete_task.go | 22 ++-- pkg/llm/tasks/dify/dify_start_task.go | 35 +++---- pkg/llm/tasks/dify/dify_stop_task.go | 61 ++++++----- pkg/llm/tasks/llm/llm_create_task.go | 54 +++++----- pkg/llm/tasks/llm/llm_delete_task.go | 17 ++- pkg/llm/tasks/llm/llm_start_task.go | 35 +++---- pkg/llm/tasks/llm/llm_stop_task.go | 33 ++---- scripts/download_ollama_models.sh | 137 +++++++++++++++++++++++++ 13 files changed, 326 insertions(+), 164 deletions(-) create mode 100644 scripts/download_ollama_models.sh diff --git a/pkg/llm/models/dify.go b/pkg/llm/models/dify.go index 29d3dabc8c..6743595672 100644 --- a/pkg/llm/models/dify.go +++ b/pkg/llm/models/dify.go @@ -17,7 +17,6 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" "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" ) @@ -172,7 +171,7 @@ func (dify *SDify) StartDeleteTask(ctx context.Context, userCred mcclient.TokenC return task.ScheduleRun(nil) } -func (dify *SDify) ServerCreate(ctx context.Context, userCred mcclient.TokenCredential, input *api.DifyCreateInput) (string, error) { +func (dify *SDify) ServerCreate(ctx context.Context, userCred mcclient.TokenCredential, s *mcclient.ClientSession, input *api.DifyCreateInput) (string, error) { model, err := dify.GetDifyModel(dify.DifyModelId) if nil != err { return "", errors.Wrap(err, "GetDifyModel") @@ -184,7 +183,6 @@ func (dify *SDify) ServerCreate(ctx context.Context, userCred mcclient.TokenCred } log.Infoln("PodCreateInput Data: ", jsonutils.Marshal(data).String()) - s := auth.GetSession(ctx, userCred, "") resp, err := compute.Servers.Create(s, jsonutils.Marshal(data)) if nil != err { return "", errors.Wrap(err, "Servers.Create") diff --git a/pkg/llm/models/dify_containers.go b/pkg/llm/models/dify_containers.go index 6340480f38..717dd9e4ba 100644 --- a/pkg/llm/models/dify_containers.go +++ b/pkg/llm/models/dify_containers.go @@ -173,6 +173,12 @@ func (m *DifyContainersManager) _getApiContainer(name, key, image string) *compu } ctr.Image = image + // set depends on + ctr.DependsOn = []string{ + name + "-" + api.DIFY_REDIS_KEY, + name + "-" + api.DIFY_POSTGRES_KEY, + } + // set container environments envs := &DifyContainerEnv{ "MODE": api.DIFY_API_MODE, @@ -201,6 +207,12 @@ func (m *DifyContainersManager) _getWorkerContainer(name, key, image string) *co } ctr.Image = image + // set depends on + ctr.DependsOn = []string{ + name + "-" + api.DIFY_REDIS_KEY, + name + "-" + api.DIFY_POSTGRES_KEY, + } + // set container environments envs := &DifyContainerEnv{ "MODE": api.DIFY_WORKER_MODE, @@ -227,6 +239,12 @@ func (m *DifyContainersManager) _getWorkerBeatContainer(name, key, image string) } ctr.Image = image + // set depends on + ctr.DependsOn = []string{ + name + "-" + api.DIFY_REDIS_KEY, + name + "-" + api.DIFY_POSTGRES_KEY, + } + // set container environments envs := &DifyContainerEnv{ "MODE": api.DIFY_WORKER_BEAT_MODE, @@ -243,6 +261,11 @@ func (m *DifyContainersManager) _getPluginContainer(name, key, image string) *co } ctr.Image = image + // set depends on + ctr.DependsOn = []string{ + name + "-" + api.DIFY_POSTGRES_KEY, + } + // set container environments envs := &DifyContainerEnv{ "DB_DATABASE": api.DIFY_PLUGIN_DB_DATABASE, @@ -350,6 +373,12 @@ func (m *DifyContainersManager) _getNginxContainer(name, key, image string) *com } ctr.Image = image + // set depends on + ctr.DependsOn = []string{ + name + "-" + api.DIFY_API_KEY, + name + "-" + api.DIFY_WEB_KEY, + } + // set container environments envs := &DifyContainerEnv{ "NGINX_SERVER_NAME": api.DIFY_NGINX_SERVER_NAME, diff --git a/pkg/llm/models/llm.go b/pkg/llm/models/llm.go index d2307bdf01..14cc7220e6 100644 --- a/pkg/llm/models/llm.go +++ b/pkg/llm/models/llm.go @@ -18,7 +18,6 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" "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" ) @@ -232,7 +231,7 @@ func (llm *SLLM) StartDeleteTask(ctx context.Context, userCred mcclient.TokenCre return task.ScheduleRun(nil) } -func (llm *SLLM) ServerCreate(ctx context.Context, userCred mcclient.TokenCredential, input *api.LLMCreateInput) (string, error) { +func (llm *SLLM) ServerCreate(ctx context.Context, userCred mcclient.TokenCredential, s *mcclient.ClientSession, input *api.LLMCreateInput) (string, error) { model, err := llm.GetLLMModel(llm.LLMModelId) if nil != err { return "", errors.Wrap(err, "GetLLMModel") @@ -248,7 +247,6 @@ func (llm *SLLM) ServerCreate(ctx context.Context, userCred mcclient.TokenCreden } log.Infoln("PodCreateInput Data: ", jsonutils.Marshal(data).String()) - s := auth.GetSession(ctx, userCred, "") resp, err := compute.Servers.Create(s, jsonutils.Marshal(data)) if nil != err { return "", errors.Wrap(err, "Servers.Create") diff --git a/pkg/llm/models/llm_base.go b/pkg/llm/models/llm_base.go index 23b76d4b8e..d333ae43cb 100644 --- a/pkg/llm/models/llm_base.go +++ b/pkg/llm/models/llm_base.go @@ -261,11 +261,10 @@ func (llm *SLLMBase) RealDelete(ctx context.Context, userCred mcclient.TokenCred return llm.SVirtualResourceBase.Delete(ctx, userCred) } -func (llm *SLLMBase) ServerDelete(ctx context.Context, userCred mcclient.TokenCredential) error { +func (llm *SLLMBase) ServerDelete(ctx context.Context, userCred mcclient.TokenCredential, s *mcclient.ClientSession) error { if len(llm.SvrId) == 0 { return nil } - s := auth.GetSession(ctx, userCred, "") server, err := llm.GetServer(ctx) if err != nil { if errors.Cause(err) == errors.ErrNotFound { @@ -276,7 +275,8 @@ func (llm *SLLMBase) ServerDelete(ctx context.Context, userCred mcclient.TokenCr } if server.DisableDelete != nil && *server.DisableDelete { // update to allow delete - _, err = compute.Servers.Update(s, llm.SvrId, jsonutils.Marshal(map[string]interface{}{"disable_delete": false})) + s2 := auth.GetSession(ctx, userCred, "") + _, err = compute.Servers.Update(s2, llm.SvrId, jsonutils.Marshal(map[string]interface{}{"disable_delete": false})) if err != nil { return errors.Wrap(err, "update server to delete") } diff --git a/pkg/llm/tasks/dify/dify_create_task.go b/pkg/llm/tasks/dify/dify_create_task.go index b5028213e5..a16da4845f 100644 --- a/pkg/llm/tasks/dify/dify_create_task.go +++ b/pkg/llm/tasks/dify/dify_create_task.go @@ -11,6 +11,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/mcclient/auth" "yunion.io/x/onecloud/pkg/util/logclient" ) @@ -44,31 +45,36 @@ func (task *DifyCreateTask) OnInit(ctx context.Context, obj db.IStandaloneModel, } serverCreateInput.Name = dify.Name - serverId, err := dify.ServerCreate(ctx, task.UserCred, &serverCreateInput) - if err != nil { - task.taskFailed(ctx, dify, err) - return - } - db.Update(dify, func() error { + task.SetStage("OnDifyRefreshStatusComplete", nil) + s := auth.GetSession(ctx, task.GetUserCred(), "") + s.WithTaskCallback(task.GetId(), func() error { + serverId, err := dify.ServerCreate(ctx, task.UserCred, s, &serverCreateInput) + if err != nil { + task.taskFailed(ctx, dify, err) + return err + } + + db.Update(dify, func() error { + dify.SvrId = serverId + return nil + }) dify.SvrId = serverId return nil }) - dify.SvrId = serverId - task.SetStage("OnDifyRefreshStatusComplete", nil) - var expectStatus []string - if serverCreateInput.AutoStart { - expectStatus = []string{computeapi.VM_RUNNING} - } else { - expectStatus = []string{computeapi.VM_READY} - } - taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { - server, err := dify.WaitServerStatus(ctx, task.UserCred, expectStatus, 7200) - if err != nil { - return nil, errors.Wrap(err, "WaitServerStatus") - } - return jsonutils.Marshal(server), nil - }) + // var expectStatus []string + // if serverCreateInput.AutoStart { + // expectStatus = []string{computeapi.VM_RUNNING} + // } else { + // expectStatus = []string{computeapi.VM_READY} + // } + // taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { + // server, err := dify.WaitServerStatus(ctx, task.UserCred, expectStatus, 7200) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // return jsonutils.Marshal(server), nil + // }) } func (task *DifyCreateTask) OnDifyRefreshStatusCompleteFailed(ctx context.Context, dify *models.SDify, err jsonutils.JSONObject) { @@ -76,10 +82,9 @@ func (task *DifyCreateTask) OnDifyRefreshStatusCompleteFailed(ctx context.Contex } func (task *DifyCreateTask) OnDifyRefreshStatusComplete(ctx context.Context, dify *models.SDify, body jsonutils.JSONObject) { - server := computeapi.ServerDetails{} - err := body.Unmarshal(&server) + server, err := dify.GetServer(ctx) if err != nil { - task.taskFailed(ctx, dify, errors.Wrap(err, "Unmarshal")) + task.taskFailed(ctx, dify, errors.Wrap(err, "GetServer")) return } diff --git a/pkg/llm/tasks/dify/dify_delete_task.go b/pkg/llm/tasks/dify/dify_delete_task.go index a82bd26042..4ecc8519e3 100644 --- a/pkg/llm/tasks/dify/dify_delete_task.go +++ b/pkg/llm/tasks/dify/dify_delete_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/mcclient/auth" "yunion.io/x/onecloud/pkg/util/logclient" ) @@ -37,20 +38,23 @@ func (task *DifyDeleteTask) OnInit(ctx context.Context, obj db.IStandaloneModel, return } - err := dify.ServerDelete(ctx, task.UserCred) + task.SetStage("OnDifyRefreshStatusComplete", nil) + s := auth.GetSession(ctx, task.GetUserCred(), "") + err := s.WithTaskCallback(task.GetId(), func() error { + return dify.ServerDelete(ctx, task.UserCred, s) + }) if err != nil { task.taskFailed(ctx, dify, err) return } - task.SetStage("OnDifyRefreshStatusComplete", nil) - taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { - err = dify.WaitDelete(ctx, task.UserCred, 1800) - if err != nil { - return nil, errors.Wrap(err, "llm.WaitDelete") - } + // taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { + // err = dify.WaitDelete(ctx, task.UserCred, 1800) + // if err != nil { + // return nil, errors.Wrap(err, "llm.WaitDelete") + // } - return nil, nil - }) + // return nil, nil + // }) } func (task *DifyDeleteTask) OnDifyRefreshStatusCompleteFailed(ctx context.Context, dify *models.SDify, err jsonutils.JSONObject) { diff --git a/pkg/llm/tasks/dify/dify_start_task.go b/pkg/llm/tasks/dify/dify_start_task.go index 2ebc9def4e..d00e497272 100644 --- a/pkg/llm/tasks/dify/dify_start_task.go +++ b/pkg/llm/tasks/dify/dify_start_task.go @@ -4,15 +4,12 @@ import ( "context" "yunion.io/x/jsonutils" - "yunion.io/x/pkg/errors" - 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/cloudcommon/db/taskman" "yunion.io/x/onecloud/pkg/llm/models" "yunion.io/x/onecloud/pkg/llm/options" - "yunion.io/x/onecloud/pkg/llm/tasks/worker" "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/mcclient/modules/compute" "yunion.io/x/onecloud/pkg/util/logclient" @@ -45,26 +42,28 @@ func (t *DifyStartTask) OnInit(ctx context.Context, obj db.IStandaloneModel, bod } func (t *DifyStartTask) requestStart(ctx context.Context, dify *models.SDify) { + t.SetStage("OnStarted", nil) s := auth.GetSession(ctx, t.GetUserCred(), options.Options.Region) - _, err := compute.Servers.PerformAction(s, dify.SvrId, "start", nil) + err := s.WithTaskCallback(t.GetId(), func() error { + _, err := compute.Servers.PerformAction(s, dify.SvrId, "start", nil) + return err + }) if err != nil { t.taskFailed(ctx, dify, err.Error()) return } - - t.SetStage("OnStarted", nil) - worker.StartTaskRun(t, func() (jsonutils.JSONObject, error) { - _, err := dify.WaitServerStatus(ctx, t.GetUserCred(), []string{computeapi.VM_RUNNING}, 900) - if err != nil { - return nil, errors.Wrap(err, "WaitServerStatus") - } - // time.Sleep(time.Second) - // _, err = d.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_RUNNING}, 900) - // if err != nil { - // return nil, errors.Wrap(err, "WaitServerStatus") - // } - return nil, nil - }) + // worker.StartTaskRun(t, func() (jsonutils.JSONObject, error) { + // _, err := dify.WaitServerStatus(ctx, t.GetUserCred(), []string{computeapi.VM_RUNNING}, 900) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // time.Sleep(time.Second) + // _, err = d.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_RUNNING}, 900) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // return nil, nil + // }) // if err := llm.RunModel(ctx, t.GetUserCred()); nil != err { // t.OnStartedFailed(ctx, llm, jsonutils.NewString(err.Error())) // return diff --git a/pkg/llm/tasks/dify/dify_stop_task.go b/pkg/llm/tasks/dify/dify_stop_task.go index 957d8b58c6..8751b065eb 100644 --- a/pkg/llm/tasks/dify/dify_stop_task.go +++ b/pkg/llm/tasks/dify/dify_stop_task.go @@ -55,34 +55,43 @@ func (task *DifyStopTask) OnInit(ctx context.Context, obj db.IStandaloneModel, b } task.SetStage("OnStopComplete", nil) - taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { - s := auth.GetSession(ctx, task.UserCred, "") + s := auth.GetSession(ctx, task.UserCred, "") + s.WithTaskCallback(task.GetId(), func() error { _, err = compute.Servers.PerformAction(s, dify.SvrId, "stop", nil) - if err != nil { - task.taskFailed(ctx, dify, err.Error()) - return nil, errors.Wrap(err, "server perform stop") - } - _, err := dify.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_READY}, 600) - if err != nil { - if errors.Cause(err) == errors.ErrTimeout { - params := computeapi.ServerStopInput{ - IsForce: true, - TimeoutSecs: 10, - } - _, err = compute.Servers.PerformAction(s, dify.SvrId, "stop", jsonutils.Marshal(params)) - if err != nil { - return nil, errors.Wrap(err, "server perform stop by force") - } - _, err := dify.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_READY}, 600) - if err != nil { - return nil, errors.Wrap(err, "WaitServerStatus 2") - } - } else { - return nil, errors.Wrap(err, "WaitServerStatus") - } - } - return nil, nil + return err }) + if err != nil { + task.taskFailed(ctx, dify, err.Error()) + return + } + // taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { + // s := auth.GetSession(ctx, task.UserCred, "") + // _, err = compute.Servers.PerformAction(s, dify.SvrId, "stop", nil) + // if err != nil { + // task.taskFailed(ctx, dify, err.Error()) + // return nil, errors.Wrap(err, "server perform stop") + // } + // _, err := dify.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_READY}, 600) + // if err != nil { + // if errors.Cause(err) == errors.ErrTimeout { + // params := computeapi.ServerStopInput{ + // IsForce: true, + // TimeoutSecs: 10, + // } + // _, err = compute.Servers.PerformAction(s, dify.SvrId, "stop", jsonutils.Marshal(params)) + // if err != nil { + // return nil, errors.Wrap(err, "server perform stop by force") + // } + // _, err := dify.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_READY}, 600) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus 2") + // } + // } else { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // } + // return nil, nil + // }) } func (task *DifyStopTask) OnStopComplete(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { diff --git a/pkg/llm/tasks/llm/llm_create_task.go b/pkg/llm/tasks/llm/llm_create_task.go index 884c7f29e9..b69d66a5b2 100644 --- a/pkg/llm/tasks/llm/llm_create_task.go +++ b/pkg/llm/tasks/llm/llm_create_task.go @@ -11,6 +11,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/mcclient/auth" "yunion.io/x/onecloud/pkg/util/logclient" ) @@ -44,31 +45,39 @@ func (task *LLMCreateTask) OnInit(ctx context.Context, obj db.IStandaloneModel, } serverCreateInput.Name = llm.Name - serverId, err := llm.ServerCreate(ctx, task.UserCred, &serverCreateInput) - if err != nil { - task.taskFailed(ctx, llm, err) - return - } - db.Update(llm, func() error { + task.SetStage("OnLLMRefreshStatusComplete", nil) + s := auth.GetSession(ctx, task.GetUserCred(), "") + err = s.WithTaskCallback(task.GetId(), func() error { + serverId, err := llm.ServerCreate(ctx, task.UserCred, s, &serverCreateInput) + if err != nil { + task.taskFailed(ctx, llm, err) + return err + } + + db.Update(llm, func() error { + llm.SvrId = serverId + return nil + }) llm.SvrId = serverId return nil }) - llm.SvrId = serverId - task.SetStage("OnLLMRefreshStatusComplete", nil) - var expectStatus []string - if serverCreateInput.AutoStart { - expectStatus = []string{computeapi.VM_RUNNING} - } else { - expectStatus = []string{computeapi.VM_READY} + if err != nil { + task.OnLLMRefreshStatusCompleteFailed(ctx, llm, jsonutils.Marshal(err)) } - taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { - server, err := llm.WaitServerStatus(ctx, task.UserCred, expectStatus, 7200) - if err != nil { - return nil, errors.Wrap(err, "WaitServerStatus") - } - return jsonutils.Marshal(server), nil - }) + // var expectStatus []string + // if serverCreateInput.AutoStart { + // expectStatus = []string{computeapi.VM_RUNNING} + // } else { + // expectStatus = []string{computeapi.VM_READY} + // } + // taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { + // server, err := llm.WaitServerStatus(ctx, task.UserCred, expectStatus, 7200) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // return jsonutils.Marshal(server), nil + // }) } func (task *LLMCreateTask) OnLLMRefreshStatusCompleteFailed(ctx context.Context, llm *models.SLLM, err jsonutils.JSONObject) { @@ -76,10 +85,9 @@ func (task *LLMCreateTask) OnLLMRefreshStatusCompleteFailed(ctx context.Context, } func (task *LLMCreateTask) OnLLMRefreshStatusComplete(ctx context.Context, llm *models.SLLM, body jsonutils.JSONObject) { - server := computeapi.ServerDetails{} - err := body.Unmarshal(&server) + server, err := llm.GetServer(ctx) if err != nil { - task.taskFailed(ctx, llm, errors.Wrap(err, "Unmarshal")) + task.taskFailed(ctx, llm, errors.Wrap(err, "Get Server")) return } diff --git a/pkg/llm/tasks/llm/llm_delete_task.go b/pkg/llm/tasks/llm/llm_delete_task.go index 76ec0fc371..84fd736a0a 100644 --- a/pkg/llm/tasks/llm/llm_delete_task.go +++ b/pkg/llm/tasks/llm/llm_delete_task.go @@ -11,6 +11,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/mcclient/auth" "yunion.io/x/onecloud/pkg/util/logclient" ) @@ -38,20 +39,14 @@ func (task *LLMDeleteTask) OnInit(ctx context.Context, obj db.IStandaloneModel, return } - err := llm.ServerDelete(ctx, task.UserCred) + task.SetStage("OnLLMRefreshStatusComplete", nil) + s := auth.GetSession(ctx, task.GetUserCred(), "") + err := s.WithTaskCallback(task.GetId(), func() error { + return llm.ServerDelete(ctx, task.UserCred, s) + }) if err != nil { task.taskFailed(ctx, llm, err) - return } - task.SetStage("OnLLMRefreshStatusComplete", nil) - taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { - err = llm.WaitDelete(ctx, task.UserCred, 1800) - if err != nil { - return nil, errors.Wrap(err, "llm.WaitDelete") - } - - return nil, nil - }) } func (task *LLMDeleteTask) OnLLMRefreshStatusCompleteFailed(ctx context.Context, llm *models.SLLM, err jsonutils.JSONObject) { diff --git a/pkg/llm/tasks/llm/llm_start_task.go b/pkg/llm/tasks/llm/llm_start_task.go index ebc052671d..a530159162 100644 --- a/pkg/llm/tasks/llm/llm_start_task.go +++ b/pkg/llm/tasks/llm/llm_start_task.go @@ -4,15 +4,12 @@ import ( "context" "yunion.io/x/jsonutils" - "yunion.io/x/pkg/errors" - 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/cloudcommon/db/taskman" "yunion.io/x/onecloud/pkg/llm/models" "yunion.io/x/onecloud/pkg/llm/options" - "yunion.io/x/onecloud/pkg/llm/tasks/worker" "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/mcclient/modules/compute" "yunion.io/x/onecloud/pkg/util/logclient" @@ -45,26 +42,28 @@ func (t *LLMStartTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body } func (t *LLMStartTask) requestStart(ctx context.Context, llm *models.SLLM) { + t.SetStage("OnStarted", nil) s := auth.GetSession(ctx, t.GetUserCred(), options.Options.Region) - _, err := compute.Servers.PerformAction(s, llm.SvrId, "start", nil) + err := s.WithTaskCallback(t.GetId(), func() error { + _, err := compute.Servers.PerformAction(s, llm.SvrId, "start", nil) + return err + }) if err != nil { t.taskFailed(ctx, llm, err.Error()) return } - - t.SetStage("OnStarted", nil) - worker.StartTaskRun(t, func() (jsonutils.JSONObject, error) { - _, err := llm.WaitServerStatus(ctx, t.GetUserCred(), []string{computeapi.VM_RUNNING}, 900) - if err != nil { - return nil, errors.Wrap(err, "WaitServerStatus") - } - // time.Sleep(time.Second) - // _, err = d.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_RUNNING}, 900) - // if err != nil { - // return nil, errors.Wrap(err, "WaitServerStatus") - // } - return nil, nil - }) + // worker.StartTaskRun(t, func() (jsonutils.JSONObject, error) { + // _, err := llm.WaitServerStatus(ctx, t.GetUserCred(), []string{computeapi.VM_RUNNING}, 900) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // time.Sleep(time.Second) + // _, err = d.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_RUNNING}, 900) + // if err != nil { + // return nil, errors.Wrap(err, "WaitServerStatus") + // } + // return nil, nil + // }) // if err := llm.RunModel(ctx, t.GetUserCred()); nil != err { // t.OnStartedFailed(ctx, llm, jsonutils.NewString(err.Error())) // return diff --git a/pkg/llm/tasks/llm/llm_stop_task.go b/pkg/llm/tasks/llm/llm_stop_task.go index 9bad225a68..a15f8cc4fe 100644 --- a/pkg/llm/tasks/llm/llm_stop_task.go +++ b/pkg/llm/tasks/llm/llm_stop_task.go @@ -55,34 +55,15 @@ func (task *LLMStopTask) OnInit(ctx context.Context, obj db.IStandaloneModel, bo } task.SetStage("OnStopComplete", nil) - taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) { - s := auth.GetSession(ctx, task.UserCred, "") + s := auth.GetSession(ctx, task.UserCred, "") + err = s.WithTaskCallback(task.GetId(), func() error { _, err = compute.Servers.PerformAction(s, llm.SvrId, "stop", nil) - if err != nil { - task.taskFailed(ctx, llm, err.Error()) - return nil, errors.Wrap(err, "server perform stop") - } - _, err := llm.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_READY}, 600) - if err != nil { - if errors.Cause(err) == errors.ErrTimeout { - params := computeapi.ServerStopInput{ - IsForce: true, - TimeoutSecs: 10, - } - _, err = compute.Servers.PerformAction(s, llm.SvrId, "stop", jsonutils.Marshal(params)) - if err != nil { - return nil, errors.Wrap(err, "server perform stop by force") - } - _, err := llm.WaitServerStatus(ctx, task.UserCred, []string{computeapi.VM_READY}, 600) - if err != nil { - return nil, errors.Wrap(err, "WaitServerStatus 2") - } - } else { - return nil, errors.Wrap(err, "WaitServerStatus") - } - } - return nil, nil + return err }) + if err != nil { + task.taskFailed(ctx, llm, err.Error()) + return + } } func (task *LLMStopTask) OnStopComplete(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) { diff --git a/scripts/download_ollama_models.sh b/scripts/download_ollama_models.sh new file mode 100644 index 0000000000..1ebc3cc457 --- /dev/null +++ b/scripts/download_ollama_models.sh @@ -0,0 +1,137 @@ +#!/bin/bash + +# 脚本: Ollama 模型安全下载器 +# 描述: 此脚本根据输入的模型名称（例如 "qwen3:8b"）安全地下载 Ollama 模型文件。 +# 它使用临时文件进行下载，并在成功后重命名，以防止文件损坏。 +# +# 用法: ./download_model.sh <模型名称:标签> +# 示例: ./download_model.sh qwen3:8b + +# --- 配置 --- + +# Ollama 注册表的基础 URL +LLM_OLLAMA_LIBRARY_BASE_URL="https://registry.ollama.ai/v2/library" +# 模型在主机上保存的基础路径 +LLM_OLLAMA_HOST_PATH="/opt/ollama-models" +# 主机上的清单目录 +LLM_OLLAMA_HOST_MANIFESTS_DIR="/manifests" +# 主机上的 blob 目录 +LLM_OLLAMA_HOST_BLOBS_DIR="/blobs" + +# --- 脚本 --- + +# 检查是否提供了模型名称作为参数 +if [ -z "$1" ]; then + echo "错误：未提供模型名称。" + echo "用法: $0 <模型名称:标签>" + exit 1 +fi + +# 从输入参数中解析模型名称和标签 +MODEL_FULL_NAME=$1 +MODEL_NAME=$(echo "$MODEL_FULL_NAME" | cut -d':' -f1) +MODEL_TAG=$(echo "$MODEL_FULL_NAME" | cut -d':' -f2) + +# 检查模型名称和标签是否已成功解析 +if [ "$MODEL_NAME" == "$MODEL_TAG" ] || [ -z "$MODEL_TAG" ]; then + echo "错误：模型名称格式无效。应为 '名称:标签' (例如 'qwen3:8b')。" + exit 1 +fi + +echo "开始安全下载模型: $MODEL_FULL_NAME" +echo "----------------------------------------" + +# --- 1. 下载清单文件 --- + +# 创建清单文件要保存的目录 (如果不存在) +MANIFEST_DIR="$LLM_OLLAMA_HOST_PATH$LLM_OLLAMA_HOST_MANIFESTS_DIR" +mkdir -p "$MANIFEST_DIR" + +# 构造清单文件的 URL、最终路径和临时路径 +MANIFEST_SUFFIX_URL="$MODEL_NAME/manifests/$MODEL_TAG" +MANIFEST_URL="$LLM_OLLAMA_LIBRARY_BASE_URL/$MANIFEST_SUFFIX_URL" +MANIFEST_FILE_PATH="$MANIFEST_DIR/$MODEL_NAME-$MODEL_TAG" +MANIFEST_FILE_PATH_TMP="${MANIFEST_FILE_PATH}.tmp" + +echo "步骤 1: 正在从 $MANIFEST_URL 下载清单..." + +# 检查最终文件是否已存在，如果存在则跳过 +if [ -f "$MANIFEST_FILE_PATH" ]; then + echo "清单文件已存在，跳过下载。" +else + # 使用 wget 下载到临时文件 + wget --quiet --show-progress -O "$MANIFEST_FILE_PATH_TMP" "$MANIFEST_URL" + # 检查 wget 的退出状态 + if [ $? -eq 0 ]; then + # 如果成功，重命名临时文件 + mv "$MANIFEST_FILE_PATH_TMP" "$MANIFEST_FILE_PATH" + echo "清单已成功下载到: $MANIFEST_FILE_PATH" + else + # 如果失败，打印错误并删除临时文件 + echo "错误：下载清单失败。请检查模型名称是否正确以及网络连接。" + rm -f "$MANIFEST_FILE_PATH_TMP" + exit 1 + fi +fi + +echo "----------------------------------------" + +# --- 2. 从清单中提取 Blob 的摘要 (digest) --- + +echo "步骤 2: 正在从清单文件中解析 blob 摘要..." +# 使用 grep 和 sed 通过正则表达式提取所有 blob 的 digest +BLOBS=$(grep -o '"digest":"sha256:[^"]*' "$MANIFEST_FILE_PATH" | sed 's/"digest":"//') + +if [ -z "$BLOBS" ]; then + echo "警告：在清单文件中未找到任何 blob 摘要。" + exit 0 +fi + +echo "已找到以下 blobs:" +echo "$BLOBS" +echo "----------------------------------------" + +# --- 3. 下载所有 Blob 文件 --- + +echo "步骤 3: 正在下载所有 blob 文件..." +# 创建 blob 文件要保存的目录 (如果不存在) +BLOBS_DIR="$LLM_OLLAMA_HOST_PATH$LLM_OLLAMA_HOST_BLOBS_DIR" +mkdir -p "$BLOBS_DIR" + +# 逐行遍历所有提取出的 blob 摘要 +for BLOB in $BLOBS; do + # 将 blob 摘要中的 "sha256:" 替换为 "sha256-" 以用作文件名 + BLOB_FILENAME=$(echo "$BLOB" | sed 's/sha256:/sha256-/') + BLOB_FILE_PATH="$BLOBS_DIR/$BLOB_FILENAME" + BLOB_FILE_PATH_TMP="${BLOB_FILE_PATH}.tmp" + + # 如果最终文件已经存在，则跳过下载 + if [ -f "$BLOB_FILE_PATH" ]; then + echo " 文件已存在, 跳过下载: $BLOB_FILENAME" + continue + fi + + # 构造 blob 的下载 URL + BLOB_URL="$LLM_OLLAMA_LIBRARY_BASE_URL/$MODEL_NAME/blobs/$BLOB" + + echo " 正在下载 $BLOB..." + # 使用 wget 下载 blob 到临时文件 + wget --quiet --show-progress -O "$BLOB_FILE_PATH_TMP" "$BLOB_URL" + + # 检查 wget 的退出状态 + if [ $? -eq 0 ]; then + # 如果成功，重命名临时文件 + mv "$BLOB_FILE_PATH_TMP" "$BLOB_FILE_PATH" + echo " 已成功保存到: $BLOB_FILE_PATH" + else + # 如果失败，打印错误并删除临时文件 + echo " 错误：下载 blob $BLOB 失败。已从 $BLOB_URL 尝试下载。" + rm -f "$BLOB_FILE_PATH_TMP" + # 如果希望在任何一个 blob 下载失败时立即中止整个脚本，请取消下一行的注释 + # exit 1 + fi +done + +echo "----------------------------------------" +echo "所有任务已完成。" +echo "模型 '$MODEL_FULL_NAME' 已成功下载到 '$LLM_OLLAMA_HOST_PATH'。" \ No newline at end of file