yms-daemon/internal/backendexecutor/executor.go

// Package backendexecutor 负责准备并启动一个显式命名的后端容器。
// 网关流量切换被刻意排除在本包职责之外。
package backendexecutor

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/distribution/reference"
	opencontainersdigest "github.com/opencontainers/go-digest"

	"yms-daemon/internal/containerengine"
	"yms-daemon/internal/healthcheck"
	"yms-daemon/internal/transaction"
)

const (
	// healthPath 后端 Actuator 健康检查端点的固定路径。
	healthPath = "/yms/actuator/health"
	// healthTimeout 等待后端容器健康的整体超时时间。
	healthTimeout = 120 * time.Second
	// healthInterval Actuator 健康检查的轮询间隔。
	healthInterval = time.Second
	// containerStopTimeoutSeconds 停止后端容器时允许的最长等待秒数。
	containerStopTimeoutSeconds = 150 * 60
	// hostNetworkMode 表示后端容器使用宿主机网络命名空间。
	hostNetworkMode = "host"
	// bindMountType 表示后端容器的挂载类型为绑定挂载。
	bindMountType = "bind"
	// stepLoadImage 从归档加载后端镜像的持久化步骤键。
	stepLoadImage = "backend.image.load"
	// stepPullImage 从仓库拉取后端镜像的持久化步骤键。
	stepPullImage = "backend.image.pull"
	// stepRemoveContainer 移除非活动后端容器的持久化步骤键。
	stepRemoveContainer = "backend.container.remove-inactive"
	// stepCreateContainer 创建非活动后端容器的持久化步骤键。
	stepCreateContainer = "backend.container.create"
	// stepStartContainer 启动非活动后端容器的持久化步骤键。
	stepStartContainer = "backend.container.start"
	// stepCheckHealth 等待后端 Actuator 健康的持久化步骤键。
	stepCheckHealth = "backend.container.health"
)

const (
	// ImageAcquisitionLoad 表示通过本地归档加载方式获取后端镜像。
	ImageAcquisitionLoad = "load"
	// ImageAcquisitionPull 表示通过远程仓库拉取方式获取后端镜像。
	ImageAcquisitionPull = "pull"
	// ImageAcquisitionPresent 表示镜像已在本地，不执行拉取或加载，只核对镜像身份。
	ImageAcquisitionPresent = "present"
)

// Request 携带 update 包与本地部署配置提供的精确取值。
// ImageReference 不透明值：执行器从不解析其标签中的任何含义。
type Request struct {
	// ImageAcquisition 指定镜像获取方式，取值为 ImageAcquisitionLoad 或 ImageAcquisitionPull。
	ImageAcquisition string
	// ArchivePath 当 ImageAcquisition 为 ImageAcquisitionLoad 时本地镜像归档的绝对路径。
	ArchivePath string
	// ImageReference 后端镜像的精确引用，执行器不解析其语义。
	ImageReference string
	// DisplayImageReference 容器配置中用于可读展示的镜像引用。非空时仅影响 docker ps 的显示，
	// 实际镜像身份仍必须由 ExpectedImageDigest 校验通过。
	DisplayImageReference string
	// ExpectedImageDigest 后端镜像期望的清单摘要。
	ExpectedImageDigest string
	// Platform 后端镜像显式指定的操作系统与架构。
	Platform containerengine.Platform
	// ContainerName 后端容器的精确名称。
	ContainerName string
	// Port 后端容器的监听端口，仅允许 8080 或 8081。
	Port int
	// PortEnvironmentKey 注入端口值的环境变量键。
	PortEnvironmentKey string
	// ConfigSource 宿主机上后端配置文件的绝对路径。
	ConfigSource string
	// ConfigTarget 后端配置在容器内的绝对挂载路径。
	ConfigTarget string
	// TmpSource 宿主机上后端临时目录的绝对路径。
	TmpSource string
	// TmpTarget 后端临时目录在容器内的绝对挂载路径。
	TmpTarget string
	// ConfigEnvironmentKey 注入配置位置的环境变量键。
	ConfigEnvironmentKey string
	// ConfigLocation 后端配置在容器内的位置取值。
	ConfigLocation string
	// RestartPolicy 后端容器的重启策略。
	RestartPolicy containerengine.RestartPolicy
	// HealthEndpoint 后端 Actuator 健康检查的 HTTP 端点。
	HealthEndpoint string
	// StartLog 表示是否读取并回传容器启动日志。
	StartLog bool
	// LogReporter 用于回传容器启动日志的每一行。
	LogReporter func(string)
}

// Executor 在新容器恢复健康后，将持久化事务推进到 StateSwitching 状态。
type Executor struct {
	// store 持久化事务的存储。
	store *transaction.Store
	// coordinator 负责事务的独占执行与步骤执行。
	coordinator *transaction.Coordinator
	// engine 底层容器引擎。
	engine containerengine.Engine
	// checker Actuator 健康检查器。
	checker *healthcheck.ActuatorChecker
}

// New 构造一个 Executor，并校验所有必要依赖非空。
// store 为事务存储，coordinator 为事务协调器，engine 为容器引擎，httpClient 用于健康检查。
// 任一必要依赖为 nil 时返回错误；健康检查器构造失败时返回该错误。
func New(store *transaction.Store, coordinator *transaction.Coordinator, engine containerengine.Engine, httpClient *http.Client) (*Executor, error) {
	if store == nil {
		return nil, errors.New("transaction store is required")
	}
	if coordinator == nil {
		return nil, errors.New("transaction coordinator is required")
	}
	if engine == nil {
		return nil, errors.New("container engine is required")
	}
	checker, err := healthcheck.NewActuatorChecker(httpClient, healthInterval)
	if err != nil {
		return nil, err
	}
	return &Executor{store: store, coordinator: coordinator, engine: engine, checker: checker}, nil
}

// Run 从事务的持久化状态恢复执行。它不会切换网关流量。
// transactionID 待执行的事务标识，request 是本次执行携带的请求参数。
// transactionID 为空时返回错误；其余错误来自事务协调器的独占执行。
func (e *Executor) Run(ctx context.Context, transactionID string, request Request) error {
	if strings.TrimSpace(transactionID) == "" {
		return errors.New("transaction ID is required")
	}
	return e.coordinator.RunExclusive(ctx, func(ctx context.Context) error {
		return e.run(ctx, transactionID, request)
	})
}

// run 在独占执行上下文内的实现，循环读取事务状态并按状态推进，
// 直到事务进入 StateSwitching 状态后返回。每个状态分支处理失败或推进错误时立即返回。
func (e *Executor) run(ctx context.Context, transactionID string, request Request) error {
	for {
		record, err := e.store.Transaction(ctx, transactionID)
		if err != nil {
			return err
		}
		switch record.State {
		case transaction.StateCreated:
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateValidating, "backend container validation started"); err != nil {
				return err
			}
		case transaction.StateValidating:
			if err := e.validate(ctx, request); err != nil {
				_, transitionErr := e.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error())
				return errors.Join(err, transitionErr)
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StatePrepared, "backend container inputs validated"); err != nil {
				return err
			}
		case transaction.StatePrepared:
			if err := e.prepare(ctx, transactionID, request); err != nil {
				return e.failUnlessRecoverable(ctx, transactionID, err)
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateStarting, "backend container prepared"); err != nil {
				return err
			}
		case transaction.StateStarting:
			// 重放 PREPARED 步骤会核对持久化意图，阻止恢复时换入另一组请求参数。
			if err := e.prepare(ctx, transactionID, request); err != nil {
				return err
			}
			if err := e.startAndCheck(ctx, transactionID, request); err != nil {
				return err
			}
			if _, err := e.store.Transition(ctx, transactionID, transaction.StateSwitching, "backend container is healthy"); err != nil {
				return err
			}
		case transaction.StateSwitching:
			return nil
		default:
			return fmt.Errorf("backend container executor cannot run transaction %s in state %s", transactionID, record.State)
		}
	}
}

// failUnlessRecoverable 判断 prepare 阶段的错误是否可恢复。
// 若 cause 是 UncertainStepError 或 ErrStepConflict，则直接原样返回（保留不确定性以便重放恢复）；
// 否则将事务标记为 StateFailed 并合并返回 cause 与状态迁移错误。
func (e *Executor) failUnlessRecoverable(ctx context.Context, transactionID string, cause error) error {
	var uncertain *transaction.UncertainStepError
	if errors.As(cause, &uncertain) || errors.Is(cause, transaction.ErrStepConflict) {
		return cause
	}
	_, transitionErr := e.store.Transition(ctx, transactionID, transaction.StateFailed, cause.Error())
	return errors.Join(cause, transitionErr)
}

// validate 对请求输入做运行时校验：先校验请求字段，再根据镜像获取方式校验镜像归档，
// 随后校验配置文件为普通文件、临时目录为直接目录，最后对容器引擎执行 Ping 探活。
func (e *Executor) validate(ctx context.Context, request Request) error {
	if err := validateRequest(request); err != nil {
		return err
	}
	if request.ImageAcquisition == ImageAcquisitionLoad {
		if err := regularFile(request.ArchivePath, "image archive"); err != nil {
			return err
		}
	}
	if err := regularFile(request.ConfigSource, "backend configuration"); err != nil {
		return err
	}
	if err := directDirectory(request.TmpSource, "backend temporary directory"); err != nil {
		return err
	}
	if err := e.engine.Ping(ctx); err != nil {
		return err
	}
	return nil
}

// prepare 根据镜像获取方式执行对应的加载或拉取步骤，随后核对镜像、移除旧的非活动容器，
// 并创建新的非活动后端容器。所有步骤均通过事务协调器持久化执行以保证可恢复。
func (e *Executor) prepare(ctx context.Context, transactionID string, request Request) error {
	switch request.ImageAcquisition {
	case ImageAcquisitionLoad:
		operation := &loadImageOperation{engine: e.engine, archivePath: request.ArchivePath, imageReference: request.ImageReference, expectedDigest: request.ExpectedImageDigest, platform: request.Platform}
		if _, err := e.coordinator.ExecuteStep(ctx, transactionID, loadIntent(request), operation); err != nil {
			return err
		}
	case ImageAcquisitionPull:
		operation := &pullImageOperation{engine: e.engine, imageReference: request.ImageReference, expectedDigest: request.ExpectedImageDigest, platform: request.Platform}
		if _, err := e.coordinator.ExecuteStep(ctx, transactionID, pullIntent(request), operation); err != nil {
			return err
		}
	case ImageAcquisitionPresent:
		// 镜像已在本地，无外部副作用，跳过拉取，后续 InspectImage 时核对摘要。
	}

	image, err := e.engine.InspectImage(ctx, request.ImageReference)
	if err != nil {
		return fmt.Errorf("inspect prepared backend image: %w", err)
	}
	if request.ImageAcquisition == ImageAcquisitionPresent {
		matches, err := imageMatches(image, request.ExpectedImageDigest, request.Platform)
		if err != nil {
			return err
		}
		if !matches {
			return fmt.Errorf("present backend image %s does not match expected digest %s", request.ImageReference, request.ExpectedImageDigest)
		}
	}
	removeOperation := &removeContainerOperation{engine: e.engine, name: request.ContainerName}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, removeIntent(request), removeOperation); err != nil {
		return err
	}
	createOperation := &createContainerOperation{
		engine:        e.engine,
		expectedImage: image,
		spec:          containerSpec(request),
	}
	_, err = e.coordinator.ExecuteStep(ctx, transactionID, createIntent(request, image.ID), createOperation)
	return err
}

// startAndCheck 先启动非活动后端容器，若请求要求读取启动日志则在后台跟随回传，
// 随后执行健康检查步骤等待后端 Actuator 健康，健康检查结束后停止日志跟随。
func (e *Executor) startAndCheck(ctx context.Context, transactionID string, request Request) error {
	startOperation := &startContainerOperation{engine: e.engine, name: request.ContainerName}
	if _, err := e.coordinator.ExecuteStep(ctx, transactionID, startIntent(request), startOperation); err != nil {
		return err
	}
	var cancelLog context.CancelFunc
	if request.StartLog && request.LogReporter != nil {
		var logCtx context.Context
		logCtx, cancelLog = context.WithCancel(ctx)
		go e.streamContainerLogs(logCtx, request.ContainerName, request.LogReporter)
	}
	healthOperation := &healthOperation{
		engine:   e.engine,
		checker:  e.checker,
		name:     request.ContainerName,
		endpoint: request.HealthEndpoint,
		timeout:  healthTimeout,
	}
	_, err := e.coordinator.ExecuteStep(ctx, transactionID, healthIntent(request), healthOperation)
	if cancelLog != nil {
		cancelLog()
	}
	return err
}

// streamContainerLogs 跟随读取指定容器的日志并逐行上报，直到 ctx 被取消。
// 它由 startAndCheck 在后台 goroutine 中调用，健康检查期间持续吐出容器启动日志。
func (e *Executor) streamContainerLogs(ctx context.Context, name string, report func(string)) {
	logs, err := e.engine.ContainerLogs(ctx, name)
	if err != nil {
		report("unable to read container startup logs: " + err.Error())
		return
	}
	defer logs.Close()
	scanner := bufio.NewScanner(logs)
	for scanner.Scan() {
		report(scanner.Text())
	}
}

// validateRequest 对请求字段做静态校验，确保所有取值精确且自洽。
// 任一字段不符合要求时返回描述性错误。
func validateRequest(request Request) error {
	switch request.ImageAcquisition {
	case ImageAcquisitionLoad:
		if !filepath.IsAbs(request.ArchivePath) {
			return errors.New("image archive path must be absolute")
		}
	case ImageAcquisitionPull:
		if request.ArchivePath != "" {
			return errors.New("pull image acquisition does not accept an archive path")
		}
	case ImageAcquisitionPresent:
		if request.ArchivePath != "" {
			return errors.New("present image acquisition does not accept an archive path")
		}
	default:
		return fmt.Errorf("unsupported image acquisition: %q", request.ImageAcquisition)
	}
	if request.ImageReference == "" || strings.TrimSpace(request.ImageReference) != request.ImageReference {
		return errors.New("exact image reference is required")
	}
	if _, err := opencontainersdigest.Parse(request.ExpectedImageDigest); err != nil {
		return fmt.Errorf("invalid expected image digest: %w", err)
	}
	if request.Platform.OS == "" || request.Platform.Architecture == "" {
		return errors.New("explicit image operating system and architecture are required")
	}
	if request.ContainerName == "" || strings.TrimSpace(request.ContainerName) != request.ContainerName {
		return errors.New("exact container name is required")
	}
	if request.Port != 8080 && request.Port != 8081 {
		return fmt.Errorf("backend container port must be 8080 or 8081: %d", request.Port)
	}
	if request.PortEnvironmentKey == "" || strings.Contains(request.PortEnvironmentKey, "=") || strings.TrimSpace(request.PortEnvironmentKey) != request.PortEnvironmentKey {
		return errors.New("exact port environment key is required")
	}
	if !filepath.IsAbs(request.ConfigSource) || !filepath.IsAbs(request.ConfigTarget) {
		return errors.New("backend configuration source and target must be absolute paths")
	}
	if !filepath.IsAbs(request.TmpSource) || !filepath.IsAbs(request.TmpTarget) {
		return errors.New("backend temporary source and target must be absolute paths")
	}
	if request.ConfigEnvironmentKey == "" || strings.Contains(request.ConfigEnvironmentKey, "=") || strings.TrimSpace(request.ConfigEnvironmentKey) != request.ConfigEnvironmentKey {
		return errors.New("exact backend configuration environment key is required")
	}
	if request.ConfigLocation == "" || strings.TrimSpace(request.ConfigLocation) != request.ConfigLocation {
		return errors.New("exact backend configuration location is required")
	}
	if err := validateRestartPolicy(request.RestartPolicy); err != nil {
		return err
	}
	parsed, err := url.ParseRequestURI(request.HealthEndpoint)
	if err != nil || parsed.Scheme != "http" || parsed.Host == "" || parsed.Path != healthPath {
		return fmt.Errorf("health endpoint must be an HTTP URL with exact path %s", healthPath)
	}
	if parsed.Port() != strconv.Itoa(request.Port) {
		return fmt.Errorf("health endpoint port must equal backend container port %d", request.Port)
	}
	return nil
}

// validateRestartPolicy 校验重启策略名称与其最大重试次数的组合是否合法。
// "no"、"always"、"unless-stopped" 不接受最大重试次数；"on-failure" 允许非负次数；其余名称不受支持。
func validateRestartPolicy(policy containerengine.RestartPolicy) error {
	switch policy.Name {
	case "no", "always", "unless-stopped":
		if policy.MaximumRetryCount != 0 {
			return fmt.Errorf("restart policy %s does not accept a maximum retry count", policy.Name)
		}
	case "on-failure":
		if policy.MaximumRetryCount < 0 {
			return errors.New("on-failure maximum retry count cannot be negative")
		}
	default:
		return fmt.Errorf("unsupported explicit restart policy: %q", policy.Name)
	}
	return nil
}

// regularFile 校验 path 指向一个普通文件，description 用于构造错误信息。
func regularFile(path, description string) error {
	info, err := os.Stat(path)
	if err != nil {
		return fmt.Errorf("inspect %s %s: %w", description, path, err)
	}
	if !info.Mode().IsRegular() {
		return fmt.Errorf("%s is not a regular file: %s", description, path)
	}
	return nil
}

// directDirectory 校验 path 指向一个目录且不是符号链接（即“直接目录”）。
func directDirectory(path, description string) error {
	info, err := os.Lstat(path)
	if err != nil {
		return fmt.Errorf("inspect %s %s: %w", description, path, err)
	}
	if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 {
		return fmt.Errorf("%s is not a direct directory: %s", description, path)
	}
	return nil
}

// containerSpec 根据请求构造后端容器的完整规格，包括名称、镜像引用、平台、环境变量、
// 宿主机网络模式、重启策略、绑定挂载、用户与停止超时。
func containerSpec(request Request) containerengine.ContainerSpec {
	imageReference := request.ImageReference
	if request.DisplayImageReference != "" {
		imageReference = request.DisplayImageReference
	}
	return containerengine.ContainerSpec{
		Name:           request.ContainerName,
		ImageReference: imageReference,
		Platform:       request.Platform,
		Environment: []string{
			request.PortEnvironmentKey + "=" + strconv.Itoa(request.Port),
			request.ConfigEnvironmentKey + "=" + request.ConfigLocation,
		},
		NetworkMode:   hostNetworkMode,
		RestartPolicy: request.RestartPolicy,
		Mounts: []containerengine.Mount{
			{Type: bindMountType, Source: request.ConfigSource, Target: request.ConfigTarget, ReadOnly: true},
			{Type: bindMountType, Source: request.TmpSource, Target: request.TmpTarget},
		},
		User:               "0:0",
		StopTimeoutSeconds: containerStopTimeoutSeconds,
	}
}

// pullIntent 构造拉取并校验后端镜像步骤的持久化意图。
func pullIntent(request Request) transaction.StepIntent {
	return intent(stepPullImage, "pull and verify backend image", struct {
		ImageReference string                   `json:"imageReference"`
		ImageDigest    string                   `json:"imageDigest"`
		Platform       containerengine.Platform `json:"platform"`
	}{request.ImageReference, request.ExpectedImageDigest, request.Platform})
}

// loadIntent 构造从归档加载并校验后端镜像步骤的持久化意图。
func loadIntent(request Request) transaction.StepIntent {
	return intent(stepLoadImage, "load and verify backend image", struct {
		ArchivePath    string                   `json:"archivePath"`
		ImageReference string                   `json:"imageReference"`
		ImageDigest    string                   `json:"imageDigest"`
		Platform       containerengine.Platform `json:"platform"`
	}{request.ArchivePath, request.ImageReference, request.ExpectedImageDigest, request.Platform})
}

// createIntent 构造创建非活动后端容器步骤的持久化意图，imageID 为已核对镜像的标识。
func createIntent(request Request, imageID string) transaction.StepIntent {
	return intent(stepCreateContainer, "create inactive backend container", struct {
		Spec    containerengine.ContainerSpec `json:"spec"`
		ImageID string                        `json:"imageId"`
	}{containerSpec(request), imageID})
}

// removeIntent 构造移除非活动后端容器步骤的持久化意图。
func removeIntent(request Request) transaction.StepIntent {
	return intent(stepRemoveContainer, "remove inactive backend container", struct {
		ContainerName string `json:"containerName"`
	}{request.ContainerName})
}

// startIntent 构造启动非活动后端容器步骤的持久化意图。
func startIntent(request Request) transaction.StepIntent {
	return intent(stepStartContainer, "start inactive backend container", struct {
		ContainerName string `json:"containerName"`
	}{request.ContainerName})
}

// healthIntent 构造等待后端 Actuator 健康步骤的持久化意图。
func healthIntent(request Request) transaction.StepIntent {
	return intent(stepCheckHealth, "wait for backend Actuator health", struct {
		ContainerName string        `json:"containerName"`
		Endpoint      string        `json:"endpoint"`
		Timeout       time.Duration `json:"timeout"`
	}{request.ContainerName, request.HealthEndpoint, healthTimeout})
}

// intent 将任意值序列化为 JSON 后封装成事务步骤意图。序列化失败视为内部错误并直接 panic。
func intent(key, name string, value any) transaction.StepIntent {
	payload, err := json.Marshal(value)
	if err != nil {
		panic(fmt.Sprintf("marshal internal step intent: %v", err))
	}
	return transaction.StepIntent{Key: key, Name: name, Intent: payload}
}

// imageMatches 判断已核对镜像是否与期望摘要及期望平台完全匹配。
// 当镜像平台与期望平台不一致时直接返回 false；否则依据镜像的描述符摘要或仓库摘要
// 中是否存在与期望摘要相等的证据来判定匹配。
func imageMatches(image containerengine.Image, expectedDigest string, expectedPlatform containerengine.Platform) (bool, error) {
	expected, err := opencontainersdigest.Parse(expectedDigest)
	if err != nil {
		return false, fmt.Errorf("parse expected image digest: %w", err)
	}
	if image.Platform != expectedPlatform {
		return false, nil
	}
	if image.DescriptorDigest != "" {
		actual, err := opencontainersdigest.Parse(image.DescriptorDigest)
		if err != nil {
			return false, fmt.Errorf("parse inspected image descriptor digest: %w", err)
		}
		if actual == expected {
			return true, nil
		}
	}
	for _, repoDigest := range image.RepoDigests {
		parsed, err := reference.ParseAnyReference(repoDigest)
		if err != nil {
			return false, fmt.Errorf("parse inspected repository digest %q: %w", repoDigest, err)
		}
		digested, ok := parsed.(reference.Digested)
		if !ok {
			return false, fmt.Errorf("inspected repository digest is not digest-qualified: %q", repoDigest)
		}
		if digested.Digest() == expected {
			return true, nil
		}
	}
	return false, nil
}

// resultJSON 将任意值序列化为 JSON RawMessage。序列化失败视为内部错误并直接 panic。
func resultJSON(value any) json.RawMessage {
	payload, err := json.Marshal(value)
	if err != nil {
		panic(fmt.Sprintf("marshal internal step result: %v", err))
	}
	return payload
}