feat(container): support local image uploads and improve import task …

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) {

agent/app/service/app.go

@@ -550,14 +550,19 @@ 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
 	}
 
 	if err = createLink(context.Background(), installTask, app, appInstall, req.Params); err != nil {
 		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
 		}

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 {

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
+}

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)
+		}
+	}
+}

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
 		}