feat(llm): async instant model delete and improve import status tracking (#25073)

fix(aiproxy): set OpenAI handler process timeouts and skip upstream error
response when request context is already cancelled.

pkg/aiproxy/handlers/chat_completions.go

@@ -54,7 +54,10 @@ func upstreamErrorStatusCode(uerr *upstream.Error) int {
 	return uerr.StatusCode
 }
 
-func writeUpstreamError(w http.ResponseWriter, uerr *upstream.Error) {
+func writeUpstreamError(ctx context.Context, w http.ResponseWriter, uerr *upstream.Error) {
+	if ctx.Err() != nil {
+		return
+	}
 	status := http.StatusBadGateway
 	if uerr != nil && uerr.StatusCode > 0 {
 		status = uerr.StatusCode
@@ -164,7 +167,7 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 	if !isStream {
 		resp, uerr := chatCompletionWithKeyFailover(ctx, up, dict, isStream, timeout)
 		if uerr != nil {
-			writeUpstreamError(w, uerr)
+			writeUpstreamError(ctx, w, uerr)
 			return
 		}
 		body := resp.Body
@@ -179,7 +182,7 @@ func chatCompletionsHandler(ctx context.Context, w http.ResponseWriter, r *http.
 
 	ch, uerr := chatCompletionStreamWithKeyFailover(ctx, up, dict, isStream, prov, timeout)
 	if uerr != nil {
-		writeUpstreamError(w, uerr)
+		writeUpstreamError(ctx, w, uerr)
 		return
 	}
 

pkg/aiproxy/handlers/completions.go

@@ -104,7 +104,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	if !isStream {
 		resp, uerr := completionsWithKeyFailover(ctx, up, dict, isStream, timeout)
 		if uerr != nil {
-			writeUpstreamError(w, uerr)
+			writeUpstreamError(ctx, w, uerr)
 			return
 		}
 		out := resp.Body
@@ -119,7 +119,7 @@ func completionsHandler(ctx context.Context, w http.ResponseWriter, r *http.Requ
 
 	ch, uerr := completionsStreamWithKeyFailover(ctx, up, dict, isStream, compProv, timeout)
 	if uerr != nil {
-		writeUpstreamError(w, uerr)
+		writeUpstreamError(ctx, w, uerr)
 		return
 	}
 

pkg/aiproxy/handlers/embeddings.go

@@ -80,7 +80,7 @@ func embeddingsHandler(ctx context.Context, w http.ResponseWriter, r *http.Reque
 
 	resp, uerr := embeddingsWithKeyFailover(ctx, up, dict, 60*time.Second)
 	if uerr != nil {
-		writeUpstreamError(w, uerr)
+		writeUpstreamError(ctx, w, uerr)
 		return
 	}
 	out := resp.Body

pkg/aiproxy/handlers/handlers.go

@@ -15,6 +15,8 @@
 package handlers
 
 import (
+	"time"
+
 	"yunion.io/x/onecloud/pkg/aiproxy/models"
 	"yunion.io/x/onecloud/pkg/aiproxy/options"
 	"yunion.io/x/onecloud/pkg/appsrv"
@@ -24,7 +26,11 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
 )
 
-const openaiCompatAPIPrefix = "/ai/openai/v1"
+const (
+	openaiCompatAPIPrefix     = "/ai/openai/v1"
+	openaiLongProcessTimeout  = 2 * time.Hour
+	openaiShortProcessTimeout = 5 * time.Minute
+)
 
 func InitHandlers(app *appsrv.Application, isSlave bool) {
 	db.InitAllManagers()
@@ -36,10 +42,14 @@ func InitHandlers(app *appsrv.Application, isSlave bool) {
 
 	db.AddScopeResourceCountHandler("", app)
 
-	app.AddHandler2("POST", openaiCompatAPIPrefix+"/chat/completions", chatCompletionsHandler, nil, "aiproxy_openai_v1_chat_completions", nil)
-	app.AddHandler2("POST", openaiCompatAPIPrefix+"/completions", completionsHandler, nil, "aiproxy_openai_v1_completions", nil)
-	app.AddHandler2("POST", openaiCompatAPIPrefix+"/embeddings", embeddingsHandler, nil, "aiproxy_openai_v1_embeddings", nil)
-	app.AddHandler2("POST", openaiCompatAPIPrefix+"/images/generations", imagesGenerationsHandler, nil, "aiproxy_openai_v1_images_generations", nil)
+	app.AddHandler2("POST", openaiCompatAPIPrefix+"/chat/completions", chatCompletionsHandler, nil, "aiproxy_openai_v1_chat_completions", nil).
+		SetProcessTimeout(openaiLongProcessTimeout)
+	app.AddHandler2("POST", openaiCompatAPIPrefix+"/completions", completionsHandler, nil, "aiproxy_openai_v1_completions", nil).
+		SetProcessTimeout(openaiLongProcessTimeout)
+	app.AddHandler2("POST", openaiCompatAPIPrefix+"/embeddings", embeddingsHandler, nil, "aiproxy_openai_v1_embeddings", nil).
+		SetProcessTimeout(openaiShortProcessTimeout)
+	app.AddHandler2("POST", openaiCompatAPIPrefix+"/images/generations", imagesGenerationsHandler, nil, "aiproxy_openai_v1_images_generations", nil).
+		SetProcessTimeout(openaiShortProcessTimeout)
 	app.AddHandler2("GET", openaiCompatAPIPrefix+"/models", modelsHandler, nil, "aiproxy_openai_v1_models", nil)
 	app.AddHandler2("GET", openaiCompatAPIPrefix+"/models/<model>", modelRetrieveHandler, nil, "aiproxy_openai_v1_models_retrieve", nil)
 

pkg/aiproxy/handlers/images_generations.go

@@ -80,7 +80,7 @@ func imagesGenerationsHandler(ctx context.Context, w http.ResponseWriter, r *htt
 
 	resp, uerr := imagesGenerationsWithKeyFailover(ctx, up, dict, 180*time.Second)
 	if uerr != nil {
-		writeUpstreamError(w, uerr)
+		writeUpstreamError(ctx, w, uerr)
 		return
 	}
 	out := resp.Body

pkg/aiproxy/models/chat_upstream.go

@@ -21,7 +21,6 @@ import (
 	"strings"
 
 	"yunion.io/x/jsonutils"
-	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/util/rbacscope"
 
@@ -233,7 +232,6 @@ func ResolveChatUpstream(ctx context.Context, userCred mcclient.TokenCredential,
 	if err != nil {
 		return nil, err
 	}
-	log.Infof("=========request model: %s, pick routing: %s", reqModel, jsonutils.Marshal(routing).PrettyString())
 	if routing == nil {
 		return nil, errors.Wrap(httperrors.ErrNotFound, "no ai_routing matched for virtual key project on this aiproxy node")
 	}

pkg/apis/llm/instantmodel_const.go

@@ -3,6 +3,8 @@ package llm
 import "time"
 
 const (
+	INSTANT_MODEL_STATUS_PACKAGING = "packaging"
+
 	InstantModelImportDownloadProgressEnd float32 = 90
 	InstantModelImportArchiveProgress     float32 = 95
 	InstantModelImportUploadProgress      float32 = 98

pkg/llm/models/instantmodel.go

@@ -794,6 +794,34 @@ func (model *SInstantModel) ValidateDeleteCondition(ctx context.Context, info js
 	return nil
 }
 
+func (model *SInstantModel) CustomizeDelete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error {
+	return model.StartDeleteTask(ctx, userCred, query, "")
+}
+
+func (model *SInstantModel) StartDeleteTask(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, parentTaskId string) error {
+	model.SetStatus(ctx, userCred, commonapis.STATUS_DELETING, "")
+	params := jsonutils.NewDict()
+	if query != nil && jsonutils.QueryBoolean(query, "purge", false) {
+		params.Set("purge", jsonutils.JSONTrue)
+	}
+	if len(model.ImageId) > 0 {
+		params.Set("image_id", jsonutils.NewString(model.ImageId))
+	}
+	task, err := taskman.TaskManager.NewTask(ctx, "LLMInstantModelDeleteTask", model, userCred, params, parentTaskId, "", nil)
+	if err != nil {
+		return errors.Wrap(err, "NewTask LLMInstantModelDeleteTask")
+	}
+	return task.ScheduleRun(nil)
+}
+
+func (model *SInstantModel) Delete(ctx context.Context, userCred mcclient.TokenCredential) error {
+	return nil
+}
+
+func (model *SInstantModel) RealDelete(ctx context.Context, userCred mcclient.TokenCredential) error {
+	return model.SSharableVirtualResourceBase.Delete(ctx, userCred)
+}
+
 func (model *SInstantModel) ValidateUpdateCondition(ctx context.Context) error {
 	if model.Enabled.IsTrue() {
 		return errors.Wrap(errors.ErrInvalidStatus, "cannot update when enabled")
@@ -1140,6 +1168,29 @@ func sanitizeInstantModelImportCacheComponent(s string) string {
 	return strings.Trim(b.String(), "-")
 }
 
+func (model *SInstantModel) updateImportStatus(ctx context.Context, userCred mcclient.TokenCredential, status string, reason string) error {
+	if model.Status == status {
+		return nil
+	}
+	oldStatus := model.Status
+	_, err := db.Update(model, func() error {
+		model.Status = status
+		return nil
+	})
+	if err != nil {
+		return errors.Wrap(err, "Update")
+	}
+	db.CallStatusChanegdNotifyHook(ctx, userCred, oldStatus, status, model)
+	if userCred != nil {
+		notes := fmt.Sprintf("%s=>%s", oldStatus, status)
+		if len(reason) > 0 {
+			notes = fmt.Sprintf("%s: %s", notes, reason)
+		}
+		db.OpsLog.LogEvent(model, db.ACT_UPDATE_STATUS, notes, userCred)
+	}
+	return nil
+}
+
 func (model *SInstantModel) DoImport(ctx context.Context, userCred mcclient.TokenCredential, s *mcclient.ClientSession, input apis.InstantModelImportInput) (tmpDir string, err error) {
 	progress := newInstantModelImportProgressUpdater(model)
 	progress.set(0, true)
@@ -1174,6 +1225,11 @@ func (model *SInstantModel) DoImport(ctx context.Context, userCred mcclient.Toke
 	progress.set(apis.InstantModelImportDownloadProgressEnd, true)
 	log.Infof("Downloaded model %s:%s with modelId: %s to %s", input.ModelName, input.ModelTag, modelId, tmpDir)
 
+	if err = model.updateImportStatus(ctx, userCred, apis.INSTANT_MODEL_STATUS_PACKAGING, "packaging model files"); err != nil {
+		err = errors.Wrap(err, "updateImportStatus packaging")
+		return
+	}
+
 	// create tar.gz archive from downloaded files
 	imagePath := fmt.Sprintf("%s/model.tgz", tmpDir)
 	_ = os.Remove(imagePath)
@@ -1183,6 +1239,11 @@ func (model *SInstantModel) DoImport(ctx context.Context, userCred mcclient.Toke
 	}
 	progress.set(apis.InstantModelImportArchiveProgress, true)
 
+	if err = model.updateImportStatus(ctx, userCred, imageapi.IMAGE_STATUS_SAVING, "uploading model archive"); err != nil {
+		err = errors.Wrap(err, "updateImportStatus saving")
+		return
+	}
+
 	// upload the image
 	imageId, err := func() (string, error) {
 		imgFile, err := os.Open(imagePath)
@@ -1205,6 +1266,8 @@ func (model *SInstantModel) DoImport(ctx context.Context, userCred mcclient.Toke
 		imgParams.GenerateName = fmt.Sprintf("%s-%s", safeModelName, strings.TrimSpace(input.ModelTag))
 		imgParams.DiskFormat = "tgz"
 		imgParams.Size = &imgFileSize
+		protected := false
+		imgParams.Protected = &protected
 		imgParams.Properties = map[string]string{
 			"llm_type":   string(input.LlmType),
 			"model_name": input.ModelName,
@@ -1238,7 +1301,6 @@ func (model *SInstantModel) DoImport(ctx context.Context, userCred mcclient.Toke
 		model.ModelId = modelId
 		model.ImageId = imageId
 		model.Mounts = mounts
-		model.Status = imageapi.IMAGE_STATUS_SAVING
 		if shouldAutoRenameInstantModelImportName(model.Name, model.LlmType, input.ModelName, input.ModelTag) {
 			suffix := extractInstantModelImportNameSuffix(model.Name)
 			model.Name = buildInstantModelFinalName(model.LlmType, modelId, input.ModelTag, suffix)

pkg/llm/tasks/llm/llm_instant_model_delete_task.go

@@ -0,0 +1,145 @@
+package llm
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/httputils"
+
+	commonapis "yunion.io/x/onecloud/pkg/apis"
+	imageapi "yunion.io/x/onecloud/pkg/apis/image"
+	"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/mcclient"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	imagemodules "yunion.io/x/onecloud/pkg/mcclient/modules/image"
+	"yunion.io/x/onecloud/pkg/util/logclient"
+)
+
+const (
+	instantModelDeletePollInterval    = 30 * time.Second
+	instantModelDeletePollMaxAttempts = 60
+)
+
+type LLMInstantModelDeleteTask struct {
+	taskman.STask
+}
+
+func init() {
+	taskman.RegisterTask(LLMInstantModelDeleteTask{})
+}
+
+func (task *LLMInstantModelDeleteTask) taskFailed(ctx context.Context, model *models.SInstantModel, err error) {
+	model.SetStatus(ctx, task.UserCred, commonapis.STATUS_DELETE_FAILED, err.Error())
+	db.OpsLog.LogEvent(model, db.ACT_DELETE_FAIL, err, task.UserCred)
+	logclient.AddActionLogWithStartable(task, model, logclient.ACT_DELETE, err, task.UserCred, false)
+	task.SetStageFailed(ctx, jsonutils.NewString(err.Error()))
+}
+
+func (task *LLMInstantModelDeleteTask) getImageId(model *models.SInstantModel) string {
+	if imageId, _ := task.Params.GetString("image_id"); len(imageId) > 0 {
+		return imageId
+	}
+	return model.ImageId
+}
+
+func (task *LLMInstantModelDeleteTask) isImageNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	if httputils.ErrorCode(err) == 404 {
+		return true
+	}
+	return strings.Contains(err.Error(), "ResourceNotFoundError")
+}
+
+func (task *LLMInstantModelDeleteTask) unprotectGlanceImage(s *mcclient.ClientSession, imageId string) error {
+	protected := false
+	updateInput := imageapi.ImageUpdateInput{
+		Protected: &protected,
+	}
+	_, err := imagemodules.Images.Update(s, imageId, jsonutils.Marshal(updateInput))
+	if err != nil && !task.isImageNotFound(err) {
+		return errors.Wrapf(err, "unprotect glance image %s", imageId)
+	}
+	return nil
+}
+
+func (task *LLMInstantModelDeleteTask) waitImageDeleted(ctx context.Context, imageId string) error {
+	s := auth.GetAdminSession(ctx, options.Options.Region)
+	var lastErr error
+	for i := 0; i < instantModelDeletePollMaxAttempts; i++ {
+		_, err := imagemodules.Images.Get(s, imageId, nil)
+		if task.isImageNotFound(err) {
+			return nil
+		}
+		if err != nil {
+			lastErr = err
+		}
+		time.Sleep(instantModelDeletePollInterval)
+	}
+	if lastErr != nil {
+		return errors.Wrapf(lastErr, "wait glance image %s deleted", imageId)
+	}
+	return errors.Errorf("wait glance image %s deleted timeout", imageId)
+}
+
+func (task *LLMInstantModelDeleteTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) {
+	model := obj.(*models.SInstantModel)
+	model.SetStatus(ctx, task.UserCred, commonapis.STATUS_DELETING, "start delete")
+
+	imageId := task.getImageId(model)
+	if len(imageId) == 0 {
+		task.OnImageDeleteComplete(ctx, model, nil)
+		return
+	}
+
+	s := auth.GetAdminSession(ctx, options.Options.Region)
+	_, err := imagemodules.Images.Get(s, imageId, nil)
+	if task.isImageNotFound(err) {
+		task.OnImageDeleteComplete(ctx, model, nil)
+		return
+	}
+	if err != nil {
+		task.taskFailed(ctx, model, errors.Wrapf(err, "get glance image %s", imageId))
+		return
+	}
+
+	task.SetStage("OnImageDeleteComplete", nil)
+	taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) {
+		if err := task.unprotectGlanceImage(s, imageId); err != nil {
+			return nil, err
+		}
+		deleteParams := jsonutils.NewDict()
+		deleteParams.Set("override_pending_delete", jsonutils.JSONTrue)
+		if jsonutils.QueryBoolean(task.Params, "purge", false) {
+			deleteParams.Set("purge", jsonutils.JSONTrue)
+		}
+		_, err := imagemodules.Images.DeleteWithParam(s, imageId, deleteParams, nil)
+		if err != nil && !task.isImageNotFound(err) {
+			return nil, errors.Wrapf(err, "delete glance image %s", imageId)
+		}
+		if err := task.waitImageDeleted(ctx, imageId); err != nil {
+			return nil, err
+		}
+		return nil, nil
+	})
+}
+
+func (task *LLMInstantModelDeleteTask) OnImageDeleteCompleteFailed(ctx context.Context, model *models.SInstantModel, err jsonutils.JSONObject) {
+	task.taskFailed(ctx, model, errors.Error(err.String()))
+}
+
+func (task *LLMInstantModelDeleteTask) OnImageDeleteComplete(ctx context.Context, model *models.SInstantModel, body jsonutils.JSONObject) {
+	err := model.RealDelete(ctx, task.GetUserCred())
+	if err != nil {
+		task.taskFailed(ctx, model, err)
+		return
+	}
+	task.SetStageComplete(ctx, nil)
+}

pkg/llm/tasks/llm/llm_sku_create_task.go

@@ -63,6 +63,10 @@ func (task *LLMSkuCreateTask) OnInit(ctx context.Context, obj db.IStandaloneMode
 	if err := task.SaveParams(extra); err != nil {
 		log.Warningf("LLMSkuCreateTask persist imported instant model id: %s", err)
 	}
+	if err := sku.AttachMountedModel(ctx, task.UserCred, instantModel.GetId()); err != nil {
+		task.taskFailed(ctx, sku, errors.Wrapf(err, "attach InstantModel %s to SKU", instantModel.GetId()))
+		return
+	}
 }
 
 func (task *LLMSkuCreateTask) OnInstantModelReady(ctx context.Context, sku *models.SLLMSku, body jsonutils.JSONObject) {