opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-08-30 17:13:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(llm): don't rely on polling (#23709)

Browse Source
This commit is contained in:
cwz_eikoh
2025-11-07 10:19:37 +08:00
committed by GitHub
parent 9bb6e3eb1b
commit bf4f1755c8
13 changed files with 326 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/llm/models/dify.go
+1 -3
View File
@@ -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")
pkg/llm/models/dify_containers.go
+29
View File
@@ -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,
pkg/llm/models/llm.go
+1 -3
View File
@@ -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")
pkg/llm/models/llm_base.go
+3 -3
View File
@@ -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")
}
pkg/llm/tasks/dify/dify_create_task.go
+29 -24
View File
@@ -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
}
pkg/llm/tasks/dify/dify_delete_task.go
+13 -9
View File
@@ -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) {
pkg/llm/tasks/dify/dify_start_task.go
+17 -18
View File
@@ -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
pkg/llm/tasks/dify/dify_stop_task.go
+35 -26
View File
@@ -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) {
pkg/llm/tasks/llm/llm_create_task.go
+31 -23
View File
@@ -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
}
pkg/llm/tasks/llm/llm_delete_task.go
+6 -11
View File
@@ -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) {
pkg/llm/tasks/llm/llm_start_task.go
+17 -18
View File
@@ -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
pkg/llm/tasks/llm/llm_stop_task.go
+7 -26
View File
@@ -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) {
scripts/download_ollama_models.sh
+137
View File
@@ -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'。"