1Panel/agent/app/service/container_image_check_update.go

package service

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/1Panel-dev/1Panel/agent/app/model"
	"github.com/1Panel-dev/1Panel/agent/app/task"
	"github.com/1Panel-dev/1Panel/agent/buserr"
	"github.com/1Panel-dev/1Panel/agent/global"
	"github.com/1Panel-dev/1Panel/agent/i18n"
	"github.com/1Panel-dev/1Panel/agent/utils/docker"
	"github.com/1Panel-dev/1Panel/agent/utils/dockerhub"
	"github.com/1Panel-dev/1Panel/agent/utils/req_helper"
	"github.com/docker/cli/cli/config"
	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/image"
	"github.com/docker/docker/client"
	"github.com/docker/docker/errdefs"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const imageCheckUpdateSetting = "ContainerImageUpdateEnabled"

type ImageCheckUpdateStatus struct {
	TaskID     string `json:"taskID"`
	Enabled    bool   `json:"enabled"`
	Running    bool   `json:"running"`
	Checked    int    `json:"checked"`
	Updates    int    `json:"updates"`
	Failed     int    `json:"failed"`
	Skipped    int    `json:"skipped"`
	Error      string `json:"error"`
	FinishedAt int64  `json:"finishedAt"`
}

var imageCheckUpdateState struct {
	sync.Mutex
	status ImageCheckUpdateStatus
}

func GetContainerImageCheckUpdateStatus() (ImageCheckUpdateStatus, error) {
	var setting model.Setting
	if err := global.DB.Where("key = ?", imageCheckUpdateSetting).First(&setting).Error; err != nil {
		return ImageCheckUpdateStatus{}, err
	}
	imageCheckUpdateState.Lock()
	defer imageCheckUpdateState.Unlock()
	status := imageCheckUpdateState.status
	status.Enabled = setting.Value == "Enable"
	return status, nil
}
func SetContainerImageCheckUpdateSchedule(enabled bool) error {
	value := "Disable"
	if enabled {
		_, err := req_helper.LoadRuntimeProxyURL()
		if errors.Is(err, req_helper.ErrProxyRequired) {
			return buserr.New("ErrImageCheckProxyRequired")
		}
		if err != nil {
			return err
		}
		value = "Enable"
	}
	return global.DB.Model(&model.Setting{}).Where("key = ?", imageCheckUpdateSetting).Update("value", value).Error
}

// StartContainerImageCheckUpdate coalesces manual and scheduled runs on this node.
func StartContainerImageCheckUpdate() (string, error) {
	imageCheckUpdateState.Lock()
	defer imageCheckUpdateState.Unlock()
	if imageCheckUpdateState.status.Running {
		return imageCheckUpdateState.status.TaskID, nil
	}
	taskItem, err := task.NewTask(i18n.GetMsgByKey("ContainerImageCheckUpdate"), task.TaskScan, task.TaskScopeImage, "", 0)
	if err != nil {
		return "", err
	}
	// Persist before returning the ID so the log drawer can immediately find it.
	if err := global.TaskDB.Save(taskItem.Task).Error; err != nil {
		_ = taskItem.Logger.Out.(io.Closer).Close()
		taskItem.DeleteLogFile()
		if cancel, ok := global.LoadTaskCancel(taskItem.TaskID); ok {
			cancel()
		}
		global.RemoveTaskCancel(taskItem.TaskID)
		return "", err
	}
	imageCheckUpdateState.status = ImageCheckUpdateStatus{Running: true, TaskID: taskItem.TaskID}
	go func() {
		result := ImageCheckUpdateStatus{TaskID: taskItem.TaskID}
		defer func() {
			result.FinishedAt = time.Now().UnixMilli()
			imageCheckUpdateState.Lock()
			imageCheckUpdateState.status = result
			imageCheckUpdateState.Unlock()
		}()
		// The action owns its deadline so the task runner waits for cleanup and logs.
		taskItem.AddSubTaskWithOps(i18n.GetMsgByKey("ContainerImageCheckUpdate"), func(t *task.Task) (err error) {
			defer func() {
				if recovered := recover(); recovered != nil {
					err = fmt.Errorf("image update check failed: %v", recovered)
				}
				t.Log(i18n.GetMsgWithMap("ContainerImageCheckSummary", map[string]interface{}{
					"checked": result.Checked, "updates": result.Updates, "failed": result.Failed, "skipped": result.Skipped,
				}))
			}()
			ctx, cancel := context.WithTimeout(t.TaskCtx, 30*time.Minute)
			defer cancel()
			if err = runContainerImageCheckUpdate(ctx, &result, t.Log); err != nil {
				return err
			}
			if result.Failed > 0 {
				return fmt.Errorf("%s", i18n.GetMsgWithMap("ContainerImageCheckFailures", map[string]interface{}{"count": result.Failed}))
			}
			return nil
		}, nil, 0, 0)
		if err := taskItem.Execute(); err != nil {
			result.Error = err.Error()
			global.LOG.Errorf("container image check: %s", err)
		}
	}()
	return taskItem.TaskID, nil
}

// These caches live only for one check. Remote results do not depend on a
// container's local image ID, which is compared separately for every baseline.
type imageCheckUpdateRepository struct {
	client     imageCheckUpdateHubClient
	err        error
	tags       []string
	tagsLoaded bool
	tagsErr    error
}
type imageCheckUpdateTargetKey struct {
	repository, tag string
	platform        dockerhub.Platform
}
type imageCheckUpdateTargetResult struct {
	target     dockerhub.Target
	compatible bool
	err        error
}
type imageCheckUpdateResult struct {
	record model.ContainerImageCheckUpdate
	err    error
}
type imageCheckUpdateCache struct {
	username, password string
	httpClient         *http.Client
	repositories       map[string]*imageCheckUpdateRepository
	targets            map[imageCheckUpdateTargetKey]imageCheckUpdateTargetResult
}

func runContainerImageCheckUpdate(ctx context.Context, result *ImageCheckUpdateStatus, logger func(string)) error {
	httpClient, err := req_helper.NewRuntimeHTTPClient()
	if errors.Is(err, req_helper.ErrProxyRequired) {
		return buserr.New("ErrImageCheckProxyRequired")
	}
	if err != nil {
		return err
	}
	if httpClient.Transport != nil {
		defer httpClient.CloseIdleConnections()
	}
	log := func(key string, values map[string]interface{}) {
		if logger != nil {
			logger(i18n.GetMsgWithMap(key, values))
		}
	}
	cli, err := newImageCheckUpdateDockerClient()
	if err != nil {
		return err
	}
	defer cli.Close()
	containers, err := cli.ContainerList(ctx, container.ListOptions{All: true})
	if err != nil {
		return err
	}
	log("ContainerImageCheckScope", map[string]interface{}{"count": len(containers)})
	// Load old results once, rather than issuing one SELECT for every failed check.
	var previousRows []model.ContainerImageCheckUpdate
	if err := global.DB.WithContext(ctx).Find(&previousRows).Error; err != nil {
		return err
	}
	previous := make(map[string]model.ContainerImageCheckUpdate, len(previousRows))
	for _, row := range previousRows {
		previous[row.ContainerID] = row
	}
	cache := imageCheckUpdateCache{httpClient: httpClient, repositories: map[string]*imageCheckUpdateRepository{}, targets: map[imageCheckUpdateTargetKey]imageCheckUpdateTargetResult{}}
	cfg := config.LoadDefaultConfigFile(io.Discard)
	if cfg != nil {
		for _, host := range []string{"https://index.docker.io/v1/", "docker.io", "registry-1.docker.io"} {
			auth, e := cfg.GetAuthConfig(host)
			if e == nil && auth.Username != "" {
				cache.username, cache.password = auth.Username, auth.Password
				break
			}
		}
	}
	checks := map[string]imageCheckUpdateResult{}
	ids := make([]string, 0, len(containers))
	names := make(map[string]string, len(containers))
	pending := make([]model.ContainerImageCheckUpdate, 0, 100)
	for _, item := range containers {
		if err := ctx.Err(); err != nil {
			return err
		}
		ids = append(ids, item.ID)
		name := item.ID
		if len(item.Names) > 0 {
			name = strings.TrimPrefix(item.Names[0], "/")
		}
		names[item.ID] = name
		log("ContainerImageCheckItem", map[string]interface{}{"name": name, "image": item.Image})
		repository, tag, supported := dockerhub.Reference(item.Image)
		key := repository + ":" + tag + "@" + item.ImageID
		entry, found := checks[key]
		if !found {
			entry.record, entry.err = checkContainerImageUpdate(ctx, cli, item, repository, tag, supported, &cache)
			checks[key] = entry
		}
		record := entry.record
		record.ContainerID, record.ImageName = item.ID, item.Image
		if entry.err != nil {
			record.CheckStatus, record.CheckError = "failed", entry.err.Error()
			old := previous[item.ID]
			if old.ImageID == item.ImageID && old.ImageName == item.Image {
				record.LatestVersion, record.TargetImage, record.TargetImageID, record.HasUpdate = old.LatestVersion, old.TargetImage, old.TargetImageID, old.HasUpdate
			}
		}
		pending = append(pending, record)
		if len(pending) == 100 {
			if err := saveContainerImageCheckUpdates(ctx, cli, pending, names, result, log); err != nil {
				return err
			}
			pending = pending[:0]
		}
	}
	if err := saveContainerImageCheckUpdates(ctx, cli, pending, names, result, log); err != nil {
		return err
	}
	query := global.DB.WithContext(ctx).Where("1 = 1")
	if len(ids) > 0 {
		query = query.Where("container_id NOT IN ?", ids)
	}
	return query.Delete(&model.ContainerImageCheckUpdate{}).Error
}

func checkContainerImageUpdate(ctx context.Context, cli imageCheckUpdateDockerClient, item container.Summary, repository, tag string, supported bool, cache *imageCheckUpdateCache) (model.ContainerImageCheckUpdate, error) {
	record := model.ContainerImageCheckUpdate{ImageID: item.ImageID, ImageName: item.Image, CheckStatus: "success", CheckedAt: time.Now()}
	if !supported {
		record.CheckStatus = "skipped"
		return record, nil
	}
	inspected, err := cli.ImageInspect(ctx, item.ImageID)
	if err != nil {
		return record, err
	}
	repo, loaded := cache.repositories[repository]
	if !loaded {
		repo = &imageCheckUpdateRepository{}
		repo.client, repo.err = newImageCheckUpdateHubClient(ctx, repository, cache.username, cache.password, cache.httpClient)
		cache.repositories[repository] = repo
	}
	if repo.err != nil {
		return record, repo.err
	}
	if dockerhub.NeedsTags(tag) {
		if !repo.tagsLoaded {
			repo.tags, repo.tagsErr = repo.client.Tags(ctx)
			repo.tagsLoaded = true
		}
		if repo.tagsErr != nil {
			return record, repo.tagsErr
		}
	}
	platform := dockerhub.Platform{OS: inspected.Os, Architecture: inspected.Architecture, Variant: inspected.Variant}
	for _, candidate := range dockerhub.Candidates(tag, repo.tags) {
		key := imageCheckUpdateTargetKey{repository: repository, tag: candidate, platform: platform}
		remote, loaded := cache.targets[key]
		if !loaded {
			remote.target, remote.compatible, remote.err = repo.client.Resolve(ctx, candidate, platform)
			cache.targets[key] = remote
		}
		if remote.err != nil {
			return record, remote.err
		}
		if !remote.compatible {
			continue
		}
		record.LatestVersion = remote.target.Tag
		record.TargetImage = "docker.io/" + repository + ":" + remote.target.Tag
		record.TargetImageID = remote.target.ConfigDigest
		sameImage := inspected.ID == remote.target.ConfigDigest
		if inspected.Descriptor != nil {
			// The containerd store uses the pulled index/manifest as its image ID,
			// not the config digest. Persist the ID that pulling this tag will produce
			// so post-upgrade reconciliation uses the same identity as Docker.
			if remote.target.ReferenceDigest == "" {
				return record, fmt.Errorf("Docker Hub manifest response is missing its digest")
			}
			record.TargetImageID = remote.target.ReferenceDigest
			sameImage = inspected.ID == remote.target.ReferenceDigest || inspected.ID == remote.target.Digest
		}
		// Both versioned and floating tags can be republished with new image content.
		record.HasUpdate = remote.target.Tag != tag || !sameImage
		return record, nil
	}
	return record, fmt.Errorf("no compatible image for %s/%s", platform.OS, platform.Architecture)
}

// Revalidate immediately before each small write batch. Never hold a database
// transaction open while waiting for Docker or Docker Hub.
func saveContainerImageCheckUpdates(ctx context.Context, cli imageCheckUpdateDockerClient, pending []model.ContainerImageCheckUpdate, names map[string]string, result *ImageCheckUpdateStatus, log func(string, map[string]interface{})) error {
	records := make([]model.ContainerImageCheckUpdate, 0, len(pending))
	for _, record := range pending {
		current, err := cli.ContainerInspect(ctx, record.ContainerID)
		if err != nil && !errdefs.IsNotFound(err) {
			return err
		}
		if err != nil || current.Image != record.ImageID {
			result.Skipped++
			log("ContainerImageCheckChanged", map[string]interface{}{"name": names[record.ContainerID]})
			continue
		}
		records = append(records, record)
	}
	if len(records) == 0 {
		return nil
	}
	if err := global.DB.WithContext(ctx).Clauses(clause.OnConflict{UpdateAll: true}).CreateInBatches(&records, 100).Error; err != nil {
		return err
	}
	// Count and announce only results that were successfully persisted.
	for _, record := range records {
		values := map[string]interface{}{"name": names[record.ContainerID], "image": record.ImageName}
		switch record.CheckStatus {
		case "failed":
			result.Failed++
			values["error"] = record.CheckError
			log("ContainerImageCheckFailed", values)
		case "skipped":
			result.Skipped++
			log("ContainerImageCheckSkipped", values)
		default:
			result.Checked++
			if record.HasUpdate {
				result.Updates++
				values["target"] = record.TargetImage
				log("ContainerImageCheckAvailable", values)
			} else {
				log("ContainerImageCheckCurrent", values)
			}
		}
	}
	return nil
}

// sameContainerImage guards cached results after external image changes.
func sameContainerImage(stored, displayed string) bool {
	return strings.TrimPrefix(stored, "sha256:") == strings.TrimPrefix(displayed, "sha256:")
}

// Reconcile only after Docker confirms the replacement container uses the target.
func reconcileContainerImageCheckUpdate(oldID, newID, imageID, imageName string) error {
	return global.DB.Transaction(func(tx *gorm.DB) error {
		var previous model.ContainerImageCheckUpdate
		if err := tx.Where("container_id = ?", oldID).Find(&previous).Error; err != nil {
			return err
		}
		if previous.ContainerID == "" {
			return nil
		}
		if err := tx.Where("container_id = ?", oldID).Delete(&model.ContainerImageCheckUpdate{}).Error; err != nil {
			return err
		}
		if previous.TargetImageID != imageID {
			return nil
		}
		previous.ContainerID = newID
		previous.ImageID = imageID
		previous.ImageName = imageName
		previous.HasUpdate = false
		previous.CheckStatus = "success"
		previous.CheckError = ""
		previous.CheckedAt = time.Now()
		return tx.Clauses(clause.OnConflict{UpdateAll: true}).Create(&previous).Error
	})
}

type imageCheckUpdateDockerClient interface {
	ContainerList(context.Context, container.ListOptions) ([]container.Summary, error)
	ContainerInspect(context.Context, string) (container.InspectResponse, error)
	ImageInspect(context.Context, string, ...client.ImageInspectOption) (image.InspectResponse, error)
	Close() error
}
type imageCheckUpdateHubClient interface {
	Tags(context.Context) ([]string, error)
	Resolve(context.Context, string, dockerhub.Platform) (dockerhub.Target, bool, error)
}

var newImageCheckUpdateDockerClient = func() (imageCheckUpdateDockerClient, error) { return docker.NewDockerClient() }
var newImageCheckUpdateHubClient = func(ctx context.Context, repository, username, password string, httpClient *http.Client) (imageCheckUpdateHubClient, error) {
	return dockerhub.New(ctx, repository, username, password, httpClient)
}