diff --git a/agent/app/repo/task.go b/agent/app/repo/task.go index b5dea011fcd7..d0450c78a911 100644 --- a/agent/app/repo/task.go +++ b/agent/app/repo/task.go @@ -2,6 +2,7 @@ package repo import ( "context" + "time" "github.com/1Panel-dev/1Panel/agent/constant" "github.com/1Panel-dev/1Panel/agent/global" @@ -110,7 +111,9 @@ func (t TaskRepo) Update(ctx context.Context, task *model.Task) error { } func (t TaskRepo) UpdateRunningTaskToFailed() error { - return getTaskDb(t.WithByStatus(constant.StatusExecuting)).Model(&model.Task{}).Updates(map[string]interface{}{"status": constant.StatusFailed, "error_msg": "1Panel restart causes failure"}).Error + return getTaskDb(t.WithByStatus(constant.StatusExecuting)).Model(&model.Task{}).Updates(map[string]interface{}{ + "status": constant.StatusFailed, "error_msg": constant.InterruptedMsg, "end_at": time.Now(), + }).Error } func (t TaskRepo) CountExecutingTask() (int64, error) { diff --git a/agent/app/service/app.go b/agent/app/service/app.go index 45cfc365eb96..d33678ae4b31 100644 --- a/agent/app/service/app.go +++ b/agent/app/service/app.go @@ -550,6 +550,11 @@ func (a AppService) installWithHooks(req request.AppInstallCreate, executeScript installTask, err := task.NewTaskWithOps(appInstall.Name, task.TaskInstall, task.TaskScopeApp, req.TaskID, appInstall.ID) if err != nil { + appInstall.Status = constant.StatusInstallErr + appInstall.Message = err.Error() + if saveErr := appInstallRepo.Save(context.Background(), appInstall); saveErr != nil { + err = fmt.Errorf("%w; save failed install status: %v", err, saveErr) + } return } @@ -557,7 +562,7 @@ func (a AppService) installWithHooks(req request.AppInstallCreate, executeScript return } - installApp := func(t *task.Task) error { + installApp := func(t *task.Task) (err error) { if err = copyData(t, app, appDetail, appInstall, req); err != nil { return err } diff --git a/agent/app/service/app_install.go b/agent/app/service/app_install.go index 38ee2b9d8650..18f33e2dbf5d 100644 --- a/agent/app/service/app_install.go +++ b/agent/app/service/app_install.go @@ -339,6 +339,11 @@ func operateAppWithLifecycleScripts(install model.AppInstall, req request.AppIns operationTask, err := task.NewTaskWithOps(install.Name, taskType, task.TaskScopeApp, req.TaskID, install.ID) if err != nil { + install.Status = constant.StatusUpErr + install.Message = err.Error() + if saveErr := appInstallRepo.Save(context.Background(), &install); saveErr != nil { + return fmt.Errorf("%w; save failed operation status: %v", err, saveErr) + } return err } operation := string(req.Operate) @@ -567,11 +572,15 @@ func (a *AppInstallService) SyncAll(systemInit bool) error { return err } for _, i := range allList { - if i.Status == constant.StatusInstalling || i.Status == constant.StatusUpgrading || i.Status == constant.StatusRebuilding || i.Status == constant.StatusUninstalling { + if appInstallOperationInterruptedOnRestart(i.Status) { if systemInit { - i.Status = constant.StatusError - i.Message = "1Panel restart causes the task to terminate" - _ = appInstallRepo.Save(context.Background(), &i) + i.Status = appInstallOperationFailureStatus(i.Status) + i.Message = constant.InterruptedMsg + if err := appInstallRepo.Save(context.Background(), &i); err != nil { + return err + } + } else if err := syncAppInstallStatus(&i, false); err != nil { + global.LOG.Errorf("sync install app[%s] error,mgs: %s", i.Name, err.Error()) } continue } @@ -931,10 +940,34 @@ func (a *AppInstallService) GetParams(id uint) (*response.AppConfig, error) { } func syncAppInstallStatus(appInstall *model.AppInstall, force bool) error { + operation := "" + switch appInstall.Status { + case constant.StatusInstalling: + operation = task.TaskInstall + case constant.StatusUpgrading: + operation = task.TaskUpgrade + case constant.StatusRebuilding: + operation = task.TaskBuild + case constant.StatusUninstalling: + operation = task.TaskUninstall + case constant.StatusStarting, constant.StatusWaiting: + operation = task.TaskUpdate + case constant.StatusRestarting: + operation = task.TaskRestart + } + if operation != "" { + if err := reconcileAppInstallTaskFailure(appInstall, operation, global.DB, global.TaskDB); err != nil { + return err + } + } switch appInstall.Status { case constant.StatusInstalling, constant.StatusRebuilding, constant.StatusUpgrading, constant.StatusUninstalling, constant.StatusStarting, constant.StatusRestarting, constant.StatusWaiting: return nil + case constant.StatusInstallErr, constant.StatusUpgradeErr, constant.StatusUpErr: + if !force { + return nil + } } cli, err := docker.NewClient() if err != nil { diff --git a/agent/app/service/app_operation_status.go b/agent/app/service/app_operation_status.go new file mode 100644 index 000000000000..fb70f4ec71cd --- /dev/null +++ b/agent/app/service/app_operation_status.go @@ -0,0 +1,76 @@ +package service + +import ( + "errors" + + "github.com/1Panel-dev/1Panel/agent/app/model" + "github.com/1Panel-dev/1Panel/agent/constant" + "gorm.io/gorm" +) + +func appInstallOperationFailureStatus(status string) string { + switch status { + case constant.StatusInstalling: + return constant.StatusInstallErr + case constant.StatusUpgrading: + return constant.StatusUpgradeErr + case constant.StatusUninstalling: + return constant.StatusError + default: + return constant.StatusUpErr + } +} + +func appInstallOperationPending(status string) bool { + switch status { + case constant.StatusInstalling, constant.StatusRebuilding, constant.StatusUpgrading, constant.StatusUninstalling, + constant.StatusStarting, constant.StatusRestarting, constant.StatusWaiting: + return true + default: + return false + } +} + +func appInstallOperationInterruptedOnRestart(status string) bool { + // ReStarting can also describe a container's Docker restart policy. + return appInstallOperationPending(status) && status != constant.StatusRestarting +} + +func reconcileAppInstallTaskFailure(install *model.AppInstall, operation string, appDB, taskDB *gorm.DB) error { + if !appInstallOperationPending(install.Status) { + return nil + } + var operationTask model.Task + err := taskDB.Where("type = ? AND resource_id = ? AND operate = ?", "App", install.ID, operation). + Order("created_at DESC").First(&operationTask).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil + } + if err != nil { + return err + } + if operationTask.Status != constant.StatusFailed && operationTask.Status != constant.StatusCanceled { + return nil + } + // A previous operation must not turn a newly queued retry into a failure. + if operationTask.CreatedAt.Before(install.UpdatedAt) && + (operationTask.EndAt.IsZero() || operationTask.EndAt.Before(install.UpdatedAt)) { + return nil + } + message := operationTask.ErrorMsg + if message == "" { + message = "the application operation task failed" + } + status := appInstallOperationFailureStatus(install.Status) + result := appDB.Model(&model.AppInstall{}). + Where("id = ? AND status = ? AND updated_at = ?", install.ID, install.Status, install.UpdatedAt). + Updates(map[string]interface{}{"status": status, "message": message}) + if result.Error != nil { + return result.Error + } + if result.RowsAffected > 0 { + install.Status = status + install.Message = message + } + return nil +} diff --git a/agent/app/service/app_operation_status_test.go b/agent/app/service/app_operation_status_test.go new file mode 100644 index 000000000000..4cc44d0b0173 --- /dev/null +++ b/agent/app/service/app_operation_status_test.go @@ -0,0 +1,147 @@ +package service + +import ( + "path/filepath" + "testing" + "time" + + "github.com/1Panel-dev/1Panel/agent/app/model" + "github.com/1Panel-dev/1Panel/agent/constant" + "github.com/glebarez/sqlite" + "gorm.io/gorm" + "gorm.io/gorm/clause" + "gorm.io/gorm/logger" +) + +func operationStatusTestDB(t *testing.T) (*gorm.DB, *gorm.DB, model.AppInstall) { + t.Helper() + open := func(name string, schema interface{}) *gorm.DB { + db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), name)), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Silent), + }) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(schema); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + sqlDB, _ := db.DB() + if sqlDB != nil { + _ = sqlDB.Close() + } + }) + return db + } + appDB := open("apps.db", &model.AppInstall{}) + taskDB := open("tasks.db", &model.Task{}) + install := model.AppInstall{ + Name: "gb10-ds-vision", AppId: 1, AppDetailId: 1, + Version: "nvidia-gb10-dspark-0.1.1", ContainerName: "vllm-dspark-1", ServiceName: "vllm-dspark", + Status: constant.StatusStarting, Env: `{"MODEL_DIR":"/models/vision"}`, + } + if err := appDB.Omit(clause.Associations).Create(&install).Error; err != nil { + t.Fatal(err) + } + return appDB, taskDB, install +} + +func TestAppInstallFailedTaskRecoversPendingState(t *testing.T) { + for _, state := range []string{constant.StatusStarting, constant.StatusWaiting, constant.StatusRestarting, constant.StatusInstalling} { + t.Run(state, func(t *testing.T) { + appDB, taskDB, install := operationStatusTestDB(t) + install.Status = state + if err := appDB.Omit(clause.Associations).Save(&install).Error; err != nil { + t.Fatal(err) + } + task := model.Task{ID: "failed", Type: "App", Operate: "TaskUpdate", ResourceID: install.ID, + Status: constant.StatusFailed, ErrorMsg: "model startup script failed", CreatedAt: install.UpdatedAt.Add(time.Second), EndAt: install.UpdatedAt.Add(2 * time.Second)} + if err := taskDB.Create(&task).Error; err != nil { + t.Fatal(err) + } + if err := reconcileAppInstallTaskFailure(&install, task.Operate, appDB, taskDB); err != nil { + t.Fatal(err) + } + want := constant.StatusUpErr + if state == constant.StatusInstalling { + want = constant.StatusInstallErr + } + var stored model.AppInstall + if err := appDB.First(&stored, install.ID).Error; err != nil { + t.Fatal(err) + } + if install.Status != want || stored.Status != want || stored.Message != task.ErrorMsg || stored.Env != install.Env { + t.Fatalf("recovered state = %s/%s, message %q, env %q", install.Status, stored.Status, stored.Message, stored.Env) + } + }) + } +} + +func TestAppInstallPendingStateKeepsActiveAndUnrelatedTasks(t *testing.T) { + for _, taskStatus := range []string{constant.StatusExecuting, constant.StatusSuccess, "missing", "previous-failure", "other-resource"} { + t.Run(taskStatus, func(t *testing.T) { + appDB, taskDB, install := operationStatusTestDB(t) + task := model.Task{ID: "task", Type: "App", Operate: "TaskUpdate", ResourceID: install.ID, + Status: taskStatus, CreatedAt: install.UpdatedAt.Add(time.Second), EndAt: install.UpdatedAt.Add(2 * time.Second)} + if taskStatus == "previous-failure" { + task.Status = constant.StatusFailed + task.CreatedAt = install.UpdatedAt.Add(-2 * time.Hour) + task.EndAt = install.UpdatedAt.Add(-time.Hour) + } + if taskStatus == "other-resource" { + task.Status = constant.StatusFailed + task.ResourceID++ + } + if taskStatus != "missing" { + if err := taskDB.Create(&task).Error; err != nil { + t.Fatal(err) + } + } + if err := reconcileAppInstallTaskFailure(&install, task.Operate, appDB, taskDB); err != nil { + t.Fatal(err) + } + if install.Status != constant.StatusStarting { + t.Fatalf("unrelated or active task changed pending state to %s", install.Status) + } + }) + } +} + +func TestAppInstallRecoveryDoesNotOverwriteConcurrentRetry(t *testing.T) { + appDB, taskDB, install := operationStatusTestDB(t) + task := model.Task{ID: "failed", Type: "App", Operate: "TaskUpdate", ResourceID: install.ID, + Status: constant.StatusCanceled, ErrorMsg: "interrupted", CreatedAt: install.UpdatedAt.Add(time.Second), EndAt: install.UpdatedAt.Add(2 * time.Second)} + if err := taskDB.Create(&task).Error; err != nil { + t.Fatal(err) + } + if err := appDB.Callback().Update().Before("gorm:update").Register("new-operation", func(db *gorm.DB) { + if err := db.Exec("UPDATE app_installs SET updated_at = ?, message = ? WHERE id = ?", install.UpdatedAt.Add(time.Hour), "new attempt", install.ID).Error; err != nil { + t.Error(err) + } + }); err != nil { + t.Fatal(err) + } + if err := reconcileAppInstallTaskFailure(&install, task.Operate, appDB, taskDB); err != nil { + t.Fatal(err) + } + var stored model.AppInstall + if err := appDB.First(&stored, install.ID).Error; err != nil { + t.Fatal(err) + } + if stored.Status != constant.StatusStarting || stored.Message != "new attempt" { + t.Fatalf("recovery overwrote a newer operation: %s, %s", stored.Status, stored.Message) + } +} + +func TestAppInstallRestartRecoveryCoversLifecycleStates(t *testing.T) { + for _, state := range []string{constant.StatusStarting, constant.StatusWaiting, constant.StatusInstalling} { + if !appInstallOperationInterruptedOnRestart(state) { + t.Errorf("restart recovery omitted %s", state) + } + } + for _, state := range []string{constant.StatusRunning, constant.StatusStopped, constant.StatusRestarting} { + if appInstallOperationInterruptedOnRestart(state) { + t.Errorf("restart recovery changes a Docker runtime state: %s", state) + } + } +} diff --git a/agent/app/service/app_utils.go b/agent/app/service/app_utils.go index a5887d465df8..9dfd7c9aed27 100644 --- a/agent/app/service/app_utils.go +++ b/agent/app/service/app_utils.go @@ -1406,9 +1406,8 @@ func handleErr(install model.AppInstall, err error, out string) error { } func doNotNeedSync(installed model.AppInstall) bool { - return installed.Status == constant.StatusInstalling || installed.Status == constant.StatusRebuilding || installed.Status == constant.StatusUpgrading || - installed.Status == constant.StatusSyncing || installed.Status == constant.StatusUninstalling || installed.Status == constant.StatusInstallErr || - installed.Status == constant.StatusStarting || installed.Status == constant.StatusRestarting || installed.Status == constant.StatusWaiting + return appInstallOperationPending(installed.Status) || installed.Status == constant.StatusSyncing || + installed.Status == constant.StatusInstallErr || installed.Status == constant.StatusUpgradeErr || installed.Status == constant.StatusUpErr } func synAppInstall(containers map[string]container.Summary, appInstall *model.AppInstall, force bool) { @@ -1502,6 +1501,11 @@ func handleInstalled(appInstallList []model.AppInstall, updated, sync, checkUpda } for _, installed := range appInstallList { + if sync && appInstallOperationPending(installed.Status) { + if err := syncAppInstallStatus(&installed, false); err != nil { + return nil, err + } + } if updated && ignoreUpdate(installed) { continue } diff --git a/agent/app/service/image.go b/agent/app/service/image.go index b8f7e65875ee..768e9f99f4aa 100644 --- a/agent/app/service/image.go +++ b/agent/app/service/image.go @@ -5,7 +5,6 @@ import ( "context" "encoding/base64" "encoding/json" - "errors" "fmt" "io" "os" @@ -28,6 +27,7 @@ import ( "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/image" "github.com/docker/docker/api/types/registry" + "github.com/docker/docker/pkg/jsonmessage" "github.com/moby/go-archive" ) @@ -325,18 +325,16 @@ func (u *ImageService) ImageLoad(req dto.ImageLoad) error { } go func() { - client, err := docker.NewDockerClient() - if err != nil { - taskItem.Log("Failed to create Docker client: " + err.Error()) - return - } - defer client.Close() - for _, itemPath := range req.Paths { currentPath := itemPath itemName := path.Base(currentPath) taskItem.AddSubTask(i18n.GetWithName("TaskImport", itemName), func(t *task.Task) error { taskItem.Logf("----------------- %s -----------------", itemName) + client, err := docker.NewDockerClient() + if err != nil { + return err + } + defer client.Close() file, err := os.Open(currentPath) if err != nil { return err @@ -347,14 +345,7 @@ func (u *ImageService) ImageLoad(req dto.ImageLoad) error { return err } defer res.Body.Close() - content, err := io.ReadAll(res.Body) - if err != nil { - return err - } - if strings.Contains(string(content), "Error") { - return errors.New(string(content)) - } - return nil + return consumeImageLoadResponse(res.Body, taskItem.Log) }, nil) } _ = taskItem.Execute() @@ -362,6 +353,29 @@ func (u *ImageService) ImageLoad(req dto.ImageLoad) error { return nil } +// Docker may report load failures in a successful HTTP response's JSON stream. +func consumeImageLoadResponse(reader io.Reader, log func(string)) error { + decoder := json.NewDecoder(reader) + for { + var message jsonmessage.JSONMessage + if err := decoder.Decode(&message); err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + if message.Error != nil && message.Error.Message != "" { + return message.Error + } + if message.ErrorMessage != "" { + return errors.New(message.ErrorMessage) + } + if text := strings.TrimSpace(message.Stream); text != "" { + log(text) + } + } +} + func (u *ImageService) ImageSave(req dto.ImageSave) error { taskItem, err := task.NewTaskWithOps(req.Name, task.TaskExport, task.TaskScopeImage, req.TaskID, 1) if err != nil { diff --git a/agent/app/service/image_load_response.go b/agent/app/service/image_load_response.go new file mode 100644 index 000000000000..f8b2bba5414b --- /dev/null +++ b/agent/app/service/image_load_response.go @@ -0,0 +1,33 @@ +package service + +import ( + "encoding/json" + "errors" + "io" + "strings" + + "github.com/docker/docker/pkg/jsonmessage" +) + +// Docker may report load failures in a successful HTTP response's JSON stream. +func consumeImageLoadResponse(reader io.Reader, log func(string)) error { + decoder := json.NewDecoder(reader) + for { + var message jsonmessage.JSONMessage + if err := decoder.Decode(&message); err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + if message.Error != nil && message.Error.Message != "" { + return message.Error + } + if message.ErrorMessage != "" { + return errors.New(message.ErrorMessage) + } + if text := strings.TrimSpace(message.Stream); text != "" { + log(text) + } + } +} diff --git a/agent/app/service/task.go b/agent/app/service/task.go index 413fab60b482..a8136f681774 100644 --- a/agent/app/service/task.go +++ b/agent/app/service/task.go @@ -57,7 +57,10 @@ func (u *TaskLogService) ReadByLine(req request.TaskLogReadReq) (*response.FileL if req.TaskID != "" { opts = append(opts, taskRepo.WithByID(req.TaskID)) } else { - opts = append(opts, repo.WithOrderRuleBy("created_at", "desc"), repo.WithByType(req.TaskType), taskRepo.WithOperate(req.TaskOperate), taskRepo.WithResourceID(req.ResourceID)) + opts = append(opts, repo.WithOrderRuleBy("created_at", "desc"), repo.WithByType(req.TaskType), taskRepo.WithResourceID(req.ResourceID)) + if req.TaskOperate != "" { + opts = append(opts, taskRepo.WithOperate(req.TaskOperate)) + } } taskModel, err := taskRepo.GetFirst(opts...) if err != nil { diff --git a/agent/app/task/task.go b/agent/app/task/task.go index 044716c41169..3883f13501d4 100644 --- a/agent/app/task/task.go +++ b/agent/app/task/task.go @@ -87,7 +87,6 @@ const ( TaskScopeCronjob = "Cronjob" TaskScopeClam = "Clam" TaskScopeSystem = "System" - TaskScopeVm = "VirtualMachine" TaskScopeFirewall = "Firewall" TaskScopeAppStore = "AppStore" TaskScopeSnapshot = "Snapshot" diff --git a/frontend/src/api/interface/website.ts b/frontend/src/api/interface/website.ts index e5141f54fe42..5da5618f95b8 100644 --- a/frontend/src/api/interface/website.ts +++ b/frontend/src/api/interface/website.ts @@ -311,6 +311,7 @@ export namespace Website { email: string; url: string; type: string; + keyType: string; useProxy: boolean; } diff --git a/frontend/src/api/modules/container.ts b/frontend/src/api/modules/container.ts index e5176cfa515a..731b5f4bf056 100644 --- a/frontend/src/api/modules/container.ts +++ b/frontend/src/api/modules/container.ts @@ -144,8 +144,13 @@ export const imagePull = (params: Container.ImagePull) => { export const imagePush = (params: Container.ImagePush) => { return http.post(`/containers/image/push`, params); }; -export const imageLoad = (params: Container.ImageLoad) => { - return http.post(`/containers/image/load`, params, TimeoutEnum.T_10M); +export const imageLoad = (params: Container.ImageLoad, currentNode?: string) => { + return http.post( + `/containers/image/load`, + params, + TimeoutEnum.T_10M, + currentNode ? { CurrentNode: currentNode } : undefined, + ); }; export const imageSave = (params: Container.ImageSave) => { return http.post(`/containers/image/save`, params, TimeoutEnum.T_10M); diff --git a/frontend/src/components/table/TableSearch.vue b/frontend/src/components/table/TableSearch.vue index f965fb385536..401ad150df98 100644 --- a/frontend/src/components/table/TableSearch.vue +++ b/frontend/src/components/table/TableSearch.vue @@ -15,7 +15,7 @@ diff --git a/frontend/src/views/container/network/index.vue b/frontend/src/views/container/network/index.vue index 52b792af8ebc..1a59e14c4a23 100644 --- a/frontend/src/views/container/network/index.vue +++ b/frontend/src/views/container/network/index.vue @@ -95,10 +95,10 @@