package backendupdate import ( "bytes" "context" "crypto/rand" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "os" "path/filepath" "strings" "yms-daemon/internal/filestore" "yms-daemon/internal/hostnginx" "yms-daemon/internal/systemd" "yms-daemon/internal/transaction" ) // gatewayOperation 切换宿主 Nginx 后端上游的事务操作。Apply 应用 after 快照 // 并写入回执,Inspect 通过对比当前配置与 before/after 快照判断执行状态。 type gatewayOperation struct { // controller 宿主 Nginx 配置的读取与应用控制器。 controller gatewayController // before 切换前的 Nginx 配置快照。 before hostnginx.Snapshot // after 切换后的 Nginx 配置快照。 after hostnginx.Snapshot // receiptPath 切换成功后写入回执摘要的文件路径。 receiptPath string } // Apply 应用 after 快照到宿主 Nginx,并在成功后写入摘要回执文件。 // 回执文件用于 Inspect 区分“已应用”与“应用内容不一致”等状态。 func (o *gatewayOperation) Apply(ctx context.Context) error { if err := o.controller.Apply(ctx, o.after); err != nil { return err } return writeImmutableFile(o.receiptPath, []byte(snapshotDigest(o.after)), 0o600) } // Inspect 读取当前 Nginx 配置并与 after/before 快照对比,返回该操作是否已应用、 // 未应用或状态未知。对比同时校验端口与配置内容;回执文件的存在与摘要一致用于 // 强化“已应用”判定。 func (o *gatewayOperation) Inspect(context.Context) (transaction.Inspection, error) { current, err := o.controller.Read() if err != nil { return transaction.Inspection{}, err } receipt, receiptErr := os.ReadFile(o.receiptPath) if current.ActivePort == o.after.ActivePort && bytes.Equal(current.Content, o.after.Content) { if receiptErr == nil && string(receipt) == snapshotDigest(o.after) { return transaction.Inspection{Status: transaction.InspectionApplied, Result: gatewayResult(current)}, nil } if errors.Is(receiptErr, os.ErrNotExist) { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: gatewayResult(current)}, nil } if receiptErr != nil { return transaction.Inspection{}, fmt.Errorf("read host Nginx switch receipt: %w", receiptErr) } return transaction.Inspection{Status: transaction.InspectionUnknown, Result: gatewayResult(current)}, nil } if current.ActivePort == o.before.ActivePort && bytes.Equal(current.Content, o.before.Content) { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: gatewayResult(current)}, nil } return transaction.Inspection{Status: transaction.InspectionUnknown, Result: gatewayResult(current)}, nil } // pathOperation 将某文件系统路径调整到期望状态的事务操作。before 保存操作前 // 的路径状态,desired 保存期望达到的路径状态,Inspect 据此判断执行结果。 type pathOperation struct { // path 被操作的路径。 path string // before 操作前的路径状态,用于判断“未应用”。 before pathState // desired 期望达到的路径状态,用于判断“已应用”。 desired pathState } // Apply 将目标路径调整到 desired 状态。 func (o *pathOperation) Apply(context.Context) error { return applyPathState(o.path, o.desired) } // Inspect 判断路径当前是否已匹配 desired 或仍匹配 before,返回相应的执行状态。 func (o *pathOperation) Inspect(context.Context) (transaction.Inspection, error) { desired, err := pathMatches(o.path, o.desired) if err != nil { return transaction.Inspection{}, err } if desired { return transaction.Inspection{Status: transaction.InspectionApplied, Result: pathResult(o.path, o.desired)}, nil } before, err := pathMatches(o.path, o.before) if err != nil { return transaction.Inspection{}, err } if before { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: pathResult(o.path, o.before)}, nil } return transaction.Inspection{Status: transaction.InspectionUnknown}, nil } // unitStopOperation 停止某个 systemd 单元的事务操作。 type unitStopOperation struct { // units systemd 管理器。 units systemd.Manager // name 待停止的单元名。 name string } // Apply 停止指定 systemd 单元。 func (o *unitStopOperation) Apply(ctx context.Context) error { return o.units.Stop(ctx, o.name) } // Inspect 查询单元状态并判断是否已停止。inactive 与 failed 均视为已停止,因为 // systemd 可能把成功停止的遗留服务标记为 failed(其追踪的 JVM 以 SIGTERM 退出, // 状态码 143),二者都表示已无活动进程,即本步骤所需结果。 func (o *unitStopOperation) Inspect(ctx context.Context) (transaction.Inspection, error) { unit, err := o.units.Inspect(ctx, o.name) if err != nil { return transaction.Inspection{}, err } result, err := json.Marshal(unit) if err != nil { return transaction.Inspection{}, err } // systemd may leave a successfully stopped legacy service in failed when // its tracked JVM exits with SIGTERM (status 143). Both states // mean no process is active, which is the required result of this step. if unit.ActiveState == "inactive" || unit.ActiveState == "failed" { return transaction.Inspection{Status: transaction.InspectionApplied, Result: result}, nil } if unitRunning(unit) { return transaction.Inspection{Status: transaction.InspectionNotApplied, Result: result}, nil } return transaction.Inspection{Status: transaction.InspectionUnknown, Result: result}, nil } // snapshotPath 检查指定路径并把其当前状态快照为 pathState。若路径不存在则返回 // absent 状态;若为符号链接则记录其目标;若为普通文件则复制一份备份并记录其 // 尺寸与摘要身份,便于后续校验与恢复。 func snapshotPath(path string, backupPath string) (pathState, error) { info, err := os.Lstat(path) if errors.Is(err, os.ErrNotExist) { return pathState{Kind: pathKindAbsent}, nil } if err != nil { return pathState{}, fmt.Errorf("inspect compatibility path %s: %w", path, err) } if info.Mode()&os.ModeSymlink != 0 { target, err := os.Readlink(path) if err != nil { return pathState{}, fmt.Errorf("read compatibility link %s: %w", path, err) } return pathState{Kind: pathKindSymlink, Target: target}, nil } if !info.Mode().IsRegular() { return pathState{}, fmt.Errorf("compatibility path is neither a regular file nor symbolic link: %s", path) } identity, err := copyFileSnapshot(path, backupPath, info.Mode().Perm()) if err != nil { return pathState{}, err } return pathState{Kind: pathKindRegular, BackupPath: backupPath, Identity: identity, Mode: uint32(info.Mode().Perm())}, nil } // applyPathState 将目标路径调整到指定状态:absent 则删除,symlink 则原子地替换 // 为指向目标的新链接,regular 则校验备份后原子地恢复为普通文件。父目录必须是 // 直接目录。所有变更完成后都会同步父目录以确保持久化。 func applyPathState(path string, state pathState) error { parent := filepath.Dir(path) info, err := os.Lstat(parent) if err != nil { return fmt.Errorf("inspect path state parent %s: %w", parent, err) } if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { return fmt.Errorf("path state parent is not a direct directory: %s", parent) } switch state.Kind { case pathKindAbsent: if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { return fmt.Errorf("remove path state %s: %w", path, err) } return syncDirectory(parent) case pathKindSymlink: if !filepath.IsAbs(state.Target) { return errors.New("path state symbolic link target must be absolute") } temporary := filepath.Join(parent, ".yms-daemon-link-"+rand.Text()) if err := os.Symlink(state.Target, temporary); err != nil { return fmt.Errorf("create temporary compatibility link: %w", err) } defer os.Remove(temporary) if err := os.Rename(temporary, path); err != nil { return fmt.Errorf("replace compatibility link %s: %w", path, err) } return syncDirectory(parent) case pathKindRegular: if err := verifyFileIdentity(state.BackupPath, state.Identity); err != nil { return fmt.Errorf("verify compatibility file snapshot: %w", err) } return copyFileAtomic(state.BackupPath, path, os.FileMode(state.Mode)) default: return fmt.Errorf("unsupported path state kind: %q", state.Kind) } } // pathMatches 判断路径当前是否匹配指定状态:absent 状态要求路径不存在,symlink // 状态要求为指向相同目标的符号链接,regular 状态要求为权限位与身份摘要都一致的 // 普通文件。 func pathMatches(path string, state pathState) (bool, error) { info, err := os.Lstat(path) if errors.Is(err, os.ErrNotExist) { return state.Kind == pathKindAbsent, nil } if err != nil { return false, err } switch state.Kind { case pathKindAbsent: return false, nil case pathKindSymlink: if info.Mode()&os.ModeSymlink == 0 { return false, nil } target, err := os.Readlink(path) return target == state.Target, err case pathKindRegular: if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 || uint32(info.Mode().Perm()) != state.Mode { return false, nil } if err := verifyFileIdentity(path, state.Identity); err != nil { return false, nil } return true, nil default: return false, fmt.Errorf("unsupported path state kind: %q", state.Kind) } } // copyFileSnapshot 将源文件复制到目标路径并返回其尺寸与 SHA-256 身份信息。 // 复制采用先写临时文件、同步、再原子重命名的流程,并同步父目录,保证落盘一致。 func copyFileSnapshot(sourcePath string, destinationPath string, mode os.FileMode) (filestore.Identity, error) { if err := os.MkdirAll(filepath.Dir(destinationPath), 0o750); err != nil { return filestore.Identity{}, err } source, err := os.Open(sourcePath) if err != nil { return filestore.Identity{}, err } defer source.Close() temporary, err := os.CreateTemp(filepath.Dir(destinationPath), ".snapshot-*") if err != nil { return filestore.Identity{}, err } temporaryPath := temporary.Name() defer os.Remove(temporaryPath) if err := temporary.Chmod(mode); err != nil { _ = temporary.Close() return filestore.Identity{}, err } digest := sha256.New() size, copyErr := io.Copy(io.MultiWriter(temporary, digest), source) if copyErr != nil { _ = temporary.Close() return filestore.Identity{}, copyErr } if err := temporary.Sync(); err != nil { _ = temporary.Close() return filestore.Identity{}, err } if err := temporary.Close(); err != nil { return filestore.Identity{}, err } if err := os.Rename(temporaryPath, destinationPath); err != nil { return filestore.Identity{}, err } if err := syncDirectory(filepath.Dir(destinationPath)); err != nil { return filestore.Identity{}, err } return filestore.Identity{Size: size, SHA256: hex.EncodeToString(digest.Sum(nil))}, nil } // copyFileAtomic 原子地将源文件复制到目标路径:先写入同目录临时文件,同步并 // 关闭后再重命名,最后同步父目录。用于普通文件路径状态的恢复。 func copyFileAtomic(sourcePath string, destinationPath string, mode os.FileMode) error { source, err := os.Open(sourcePath) if err != nil { return err } defer source.Close() parent := filepath.Dir(destinationPath) temporary, err := os.CreateTemp(parent, ".restore-*") if err != nil { return err } temporaryPath := temporary.Name() defer os.Remove(temporaryPath) if err := temporary.Chmod(mode); err != nil { _ = temporary.Close() return err } if _, err := io.Copy(temporary, source); err != nil { _ = temporary.Close() return err } if err := temporary.Sync(); err != nil { _ = temporary.Close() return err } if err := temporary.Close(); err != nil { return err } if err := os.Rename(temporaryPath, destinationPath); err != nil { return err } return syncDirectory(parent) } // writeImmutableFile 以排他方式写入不可变文件:若路径已存在则要求其内容与待写 // 内容完全一致,否则报错;若不存在则以指定权限原子地创建并同步。写入失败时会 // 清理残留文件。 func writeImmutableFile(path string, content []byte, mode os.FileMode) error { if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { return err } file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, mode) if errors.Is(err, os.ErrExist) { existing, readErr := os.ReadFile(path) if readErr != nil { return readErr } if !bytes.Equal(existing, content) { return fmt.Errorf("immutable file already exists with different content: %s", path) } return nil } if err != nil { return err } complete := false defer func() { if !complete { _ = os.Remove(path) } }() if _, err := file.Write(content); err != nil { _ = file.Close() return err } if err := file.Sync(); err != nil { _ = file.Close() return err } if err := file.Close(); err != nil { return err } if err := syncDirectory(filepath.Dir(path)); err != nil { return err } complete = true return nil } // readGatewaySnapshot 读取持久化的 Nginx 配置快照文件,并校验其中活动端口与 // 期望端口一致,返回快照内容与端口。 func readGatewaySnapshot(path string, port int) (hostnginx.Snapshot, error) { content, err := os.ReadFile(path) if err != nil { return hostnginx.Snapshot{}, err } actual, err := hostnginx.ActiveBackendPort(content) if err != nil { return hostnginx.Snapshot{}, err } if actual != port { return hostnginx.Snapshot{}, fmt.Errorf("persisted host Nginx snapshot port mismatch: got %d, want %d", actual, port) } return hostnginx.Snapshot{Content: content, ActivePort: port}, nil } // verifyFileIdentity 校验文件的实际尺寸与 SHA-256 摘要是否与给定身份一致,任何 // 不一致都返回错误。 func verifyFileIdentity(path string, identity filestore.Identity) error { if err := identity.Validate(); err != nil { return err } file, err := os.Open(path) if err != nil { return err } digest := sha256.New() size, copyErr := io.Copy(digest, file) closeErr := file.Close() if err := errors.Join(copyErr, closeErr); err != nil { return err } if size != identity.Size || !strings.EqualFold(hex.EncodeToString(digest.Sum(nil)), identity.SHA256) { return errors.New("file identity mismatch") } return nil } // snapshotDigest 计算 Nginx 快照内容的 SHA-256 摘要(十六进制字符串),用于 // 回执与意图中的一致性标识。 func snapshotDigest(snapshot hostnginx.Snapshot) string { digest := sha256.Sum256(snapshot.Content) return hex.EncodeToString(digest[:]) } // gatewayResult 将 Nginx 快照的活动端口与内容摘要序列化为事务检查结果的 JSON。 func gatewayResult(snapshot hostnginx.Snapshot) json.RawMessage { result, _ := json.Marshal(struct { ActivePort int `json:"activePort"` SHA256 string `json:"sha256"` }{snapshot.ActivePort, snapshotDigest(snapshot)}) return result } // pathResult 将路径及其状态序列化为事务检查结果的 JSON。 func pathResult(path string, state pathState) json.RawMessage { result, _ := json.Marshal(struct { Path string `json:"path"` State pathState `json:"state"` }{path, state}) return result } // syncDirectory 打开目录并执行 fsync,将目录项变更持久化到磁盘。 func syncDirectory(directory string) error { file, err := os.Open(directory) if err != nil { return err } syncErr := file.Sync() closeErr := file.Close() return errors.Join(syncErr, closeErr) }