package backendexecutor import ( "context" "encoding/json" "errors" "fmt" "os" "slices" "sync" "time" "yms-daemon/internal/containerengine" "yms-daemon/internal/healthcheck" "yms-daemon/internal/transaction" ) // loadImageOperation 表示“从本地归档加载并校验后端镜像”的事务操作。 type loadImageOperation struct { // engine 执行加载操作的容器引擎。 engine containerengine.Engine // archivePath 本地镜像归档的绝对路径。 archivePath string // imageReference 镜像加载后的目标精确引用。 imageReference string // expectedDigest 加载镜像后期望的清单摘要。 expectedDigest string // platform 镜像的期望平台。 platform containerengine.Platform } // pullImageOperation 表示“从远程仓库拉取并校验后端镜像”的事务操作。 type pullImageOperation struct { // engine 执行拉取操作的容器引擎。 engine containerengine.Engine // imageReference 待拉取的镜像精确引用。 imageReference string // expectedDigest 拉取镜像后期望的清单摘要。 expectedDigest string // platform 镜像的期望平台。 platform containerengine.Platform } // Apply 调用容器引擎按 imageReference 拉取后端镜像。 func (o *pullImageOperation) Apply(ctx context.Context) error { return o.engine.PullImage(ctx, o.imageReference) } // Inspect 检查拉取操作是否已生效:镜像不存在时为未应用;存在时依据摘要与平台判定, // 匹配则返回已应用,否则返回未应用。无法判定或解析失败时返回错误。 func (o *pullImageOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { image, err := o.engine.InspectImage(ctx, o.imageReference) if errors.Is(err, containerengine.ErrNotFound) { return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil } if err != nil { return transaction.Inspection{}, err } matches, err := imageMatches(image, o.expectedDigest, o.platform) if err != nil { return transaction.Inspection{}, err } result := resultJSON(image) if !matches { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil } // Apply 打开本地镜像归档并调用容器引擎加载后端镜像。 func (o *loadImageOperation) Apply(ctx context.Context) error { archive, err := os.Open(o.archivePath) if err != nil { return fmt.Errorf("open image archive: %w", err) } defer archive.Close() return o.engine.LoadImage(ctx, archive) } // Inspect 检查加载操作是否已生效:镜像不存在时为未应用;存在时依据摘要与平台判定, // 匹配则返回已应用,否则返回未应用。结果携带镜像标识、仓库摘要、描述符摘要与平台信息。 func (o *loadImageOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { image, err := o.engine.InspectImage(ctx, o.imageReference) if errors.Is(err, containerengine.ErrNotFound) { return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil } if err != nil { return transaction.Inspection{}, err } matches, err := imageMatches(image, o.expectedDigest, o.platform) if err != nil { return transaction.Inspection{}, err } result := resultJSON(struct { ImageID string `json:"imageId"` RepoDigests []string `json:"repoDigests"` DescriptorDigest string `json:"descriptorDigest"` Platform containerengine.Platform `json:"platform"` }{image.ID, image.RepoDigests, image.DescriptorDigest, image.Platform}) if !matches { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil } // createContainerOperation 表示“创建非活动后端容器”的事务操作。 type createContainerOperation struct { // engine 执行创建操作的容器引擎。 engine containerengine.Engine // expectedImage 创建容器时期望使用的镜像。 expectedImage containerengine.Image // spec 待创建容器的完整规格。 spec containerengine.ContainerSpec } // removeContainerOperation 表示“移除非活动后端容器”的事务操作。 type removeContainerOperation struct { // engine 执行移除操作的容器引擎。 engine containerengine.Engine // name 待移除容器的名称。 name string } // Apply 强制移除指定名称的非活动容器,容器不存在时视为成功。 func (o *removeContainerOperation) Apply(ctx context.Context) error { err := o.engine.RemoveContainer(ctx, o.name, true) if errors.Is(err, containerengine.ErrNotFound) { return nil } return err } // Inspect 检查移除操作是否已生效:容器不存在即为已应用,存在则为未应用并返回容器信息。 func (o *removeContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { record, err := o.engine.InspectContainer(ctx, o.name) if errors.Is(err, containerengine.ErrNotFound) { return transaction.Inspection{Status: transaction.InspectionApplied}, nil } if err != nil { return transaction.Inspection{}, err } return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: containerResult(record)}, nil } // Apply 调用容器引擎按规格创建非活动后端容器。 func (o *createContainerOperation) Apply(ctx context.Context) error { _, err := o.engine.CreateContainer(ctx, o.spec) return err } // Inspect 检查创建操作是否已生效:容器不存在时为未应用;存在时若与期望镜像及规格完全匹配则为已应用, // 否则为未知状态(存在冲突),以便上层据此重放或报错。 func (o *createContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { record, err := o.engine.InspectContainer(ctx, o.spec.Name) if errors.Is(err, containerengine.ErrNotFound) { return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil } if err != nil { return transaction.Inspection{}, err } result := containerResult(record) if !containerMatches(record, o.expectedImage.ID, o.spec) { return transaction.Inspection{Status: transaction.InspectionUnknown, Result: result}, nil } return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil } // startContainerOperation 表示“启动非活动后端容器”的事务操作。 type startContainerOperation struct { // engine 执行启动操作的容器引擎。 engine containerengine.Engine // name 待启动容器的名称。 name string } // Apply 调用容器引擎启动指定名称的后端容器。 func (o *startContainerOperation) Apply(ctx context.Context) error { return o.engine.StartContainer(ctx, o.name) } // Inspect 检查启动操作是否已生效:容器不存在时为未应用;存在且处于运行且非 Dead 状态时为已应用, // 否则为未应用。结果携带容器运行状态信息。 func (o *startContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { record, err := o.engine.InspectContainer(ctx, o.name) if errors.Is(err, containerengine.ErrNotFound) { return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil } if err != nil { return transaction.Inspection{}, err } result := containerResult(record) if record.Running && !record.Dead { return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil } return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } // healthOperation 表示“等待后端 Actuator 健康”的事务操作。 type healthOperation struct { // engine 用于查询容器运行状态。 engine containerengine.Engine // checker Actuator 健康检查器。 checker *healthcheck.ActuatorChecker // name 待检查容器的名称。 name string // endpoint 健康检查的 HTTP 端点。 endpoint string // timeout 等待健康的整体超时时间。 timeout time.Duration // mu 保护 confirmed 与 confirmedReport 的并发访问。 mu sync.Mutex // confirmedReport 已确认的健康报告。 confirmedReport healthcheck.ActuatorReport // confirmed 表示健康检查结果是否已经确认。 confirmed bool } // Apply 调用健康检查器等待后端健康,成功后加锁保存已确认的健康报告并标记为已确认。 func (o *healthOperation) Apply(ctx context.Context) error { report, err := o.checker.Wait(ctx, o.endpoint, o.timeout, o.running) if err != nil { return err } o.mu.Lock() o.confirmedReport = report o.confirmed = true o.mu.Unlock() return nil } // Inspect 检查健康操作是否已生效:若先前已确认则直接返回已确认结果, // 否则执行一次即时健康检查并返回检查结论。 func (o *healthOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { o.mu.Lock() if o.confirmed { report := o.confirmedReport o.mu.Unlock() return healthInspection(report, true, nil) } o.mu.Unlock() report, ready, err := o.checker.Check(ctx, o.endpoint, o.running) return healthInspection(report, ready, err) } // running 查询后端容器是否处于运行且非 Dead 状态,容器不存在时返回 false 且无错误。 func (o *healthOperation) running(ctx context.Context) (bool, error) { record, err := o.engine.InspectContainer(ctx, o.name) if errors.Is(err, containerengine.ErrNotFound) { return false, nil } if err != nil { return false, err } return record.Running && !record.Dead, nil } // healthInspection 将健康检查结果转换为事务检查结论:工作负载停止、检查出错或未就绪时均为未应用, // 仅当 ready 为 true 且无错误时才判定为已应用。结果始终携带健康报告的 JSON 序列化。 func healthInspection(report healthcheck.ActuatorReport, ready bool, err error) (transaction.Inspection, error) { result := resultJSON(report) if errors.Is(err, healthcheck.ErrWorkloadStopped) { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } if err != nil { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } if !ready { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil } // containerMatches 判断实际容器记录是否与期望镜像标识及期望规格完全匹配, // 包括镜像、网络模式、重启策略、用户、停止超时以及环境变量与挂载集合的逐项包含关系。 func containerMatches(record containerengine.Container, expectedImageID string, spec containerengine.ContainerSpec) bool { if record.ImageID != expectedImageID || record.NetworkMode != spec.NetworkMode || record.RestartPolicy != spec.RestartPolicy || record.User != spec.User || record.StopTimeoutSeconds != spec.StopTimeoutSeconds { return false } for _, expected := range spec.Environment { if !slices.Contains(record.Environment, expected) { return false } } for _, expected := range spec.Mounts { if !slices.Contains(record.Mounts, expected) { return false } } return true } // containerResult 将容器记录的标识、镜像、运行、Dead 与状态字段序列化为 JSON RawMessage。 func containerResult(record containerengine.Container) json.RawMessage { return resultJSON(struct { ID string `json:"id"` ImageID string `json:"imageId"` Running bool `json:"running"` Dead bool `json:"dead"` Status string `json:"status"` }{record.ID, record.ImageID, record.Running, record.Dead, record.Status}) } // 以下编译期断言确保各操作类型均实现 transaction.Operation 接口。 var _ transaction.Operation = (*loadImageOperation)(nil) var _ transaction.Operation = (*pullImageOperation)(nil) var _ transaction.Operation = (*removeContainerOperation)(nil) var _ transaction.Operation = (*createContainerOperation)(nil) var _ transaction.Operation = (*startContainerOperation)(nil) var _ transaction.Operation = (*healthOperation)(nil)