tian
6a995db6e2
- agent: procctl.Status 增加 LastError（未检测到进程/进程已退出原因）， config/status 透传 media.last_error - control: ConfigStatusMediaServer 增加 LastError 字段 - control: loadConfigStatus 错误分类（agent 未响应 vs HTTP 错误，含 IP:端口） - UI 文案: '待确认/未确认' -> '状态未知'，媒体服务行显示具体错误 用户可区分: 服务故障(带原因) vs 配置未取到 vs 视频源问题
441 lines
12 KiB
Go
441 lines
12 KiB
Go
package procctl
|
||
|
||
import (
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"safesight-agent/internal/config"
|
||
"safesight-agent/internal/files"
|
||
)
|
||
|
||
var ErrNotSupported = errors.New("not supported")
|
||
var ErrConflict = errors.New("conflict")
|
||
var ErrInvalidConfig = errors.New("invalid config")
|
||
var ErrConfigNotFound = errors.New("config not found")
|
||
|
||
// ProcessController 定义进程控制器接口
|
||
type ProcessController interface {
|
||
Enabled() bool
|
||
Status() (Status, error)
|
||
Version() (string, error)
|
||
Start(configName string) (Status, error)
|
||
Stop() (Status, error)
|
||
Restart(configName string) (Status, error)
|
||
BinaryInfo() (BinaryUpdateResult, error)
|
||
UpdateBinary(r io.Reader, contentLength int64, expectedSha256 string) (BinaryUpdateResult, error)
|
||
RollbackBinary(backupPath string) (BinaryUpdateResult, error)
|
||
}
|
||
|
||
type Status struct {
|
||
Running bool `json:"running"`
|
||
Pid int `json:"pid"`
|
||
ConfigPath string `json:"config_path"`
|
||
StartedAtMS int64 `json:"started_at_ms"`
|
||
LastError string `json:"last_error,omitempty"`
|
||
}
|
||
|
||
type BinaryUpdateResult struct {
|
||
Path string `json:"path"`
|
||
Sha256 string `json:"sha256"`
|
||
Size int64 `json:"size"`
|
||
MtimeMS int64 `json:"mtime_ms"`
|
||
BackupPath string `json:"backup_path"`
|
||
}
|
||
|
||
type pidFile struct {
|
||
Pid int `json:"pid"`
|
||
ConfigPath string `json:"config_path"`
|
||
StartedAtMS int64 `json:"started_at_ms"`
|
||
}
|
||
|
||
type Controller struct {
|
||
mu sync.Mutex
|
||
proc config.MediaServerProcessConfig
|
||
defCfg string
|
||
}
|
||
|
||
func New(agentCfg config.AgentConfig, baseDir string) *Controller {
|
||
p := agentCfg.MediaServerProcess
|
||
if baseDir != "" {
|
||
if p.ExecPath != "" && !filepath.IsAbs(p.ExecPath) {
|
||
p.ExecPath = filepath.Join(baseDir, p.ExecPath)
|
||
}
|
||
if p.WorkDir != "" && !filepath.IsAbs(p.WorkDir) {
|
||
p.WorkDir = filepath.Join(baseDir, p.WorkDir)
|
||
}
|
||
if p.ConfigsDir != "" && !filepath.IsAbs(p.ConfigsDir) {
|
||
p.ConfigsDir = filepath.Join(baseDir, p.ConfigsDir)
|
||
}
|
||
if p.PidFile != "" && !filepath.IsAbs(p.PidFile) {
|
||
p.PidFile = filepath.Join(baseDir, p.PidFile)
|
||
}
|
||
}
|
||
return &Controller{proc: p, defCfg: agentCfg.ConfigPath}
|
||
}
|
||
|
||
func (c *Controller) Enabled() bool { return c != nil && c.proc.Enable }
|
||
|
||
func (c *Controller) Status() (Status, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
pF, err := c.readPidFile()
|
||
if err != nil {
|
||
return Status{}, err
|
||
}
|
||
if pF == nil {
|
||
return Status{Running: false, LastError: "未检测到媒体进程（pid 文件不存在）"}, nil
|
||
}
|
||
|
||
alive, aerr := isAlive(pF.Pid)
|
||
if errors.Is(aerr, ErrNotSupported) {
|
||
return Status{}, ErrNotSupported
|
||
}
|
||
if !alive {
|
||
_ = os.Remove(c.proc.PidFile)
|
||
return Status{
|
||
Running: false,
|
||
Pid: pF.Pid,
|
||
ConfigPath: pF.ConfigPath,
|
||
StartedAtMS: pF.StartedAtMS,
|
||
LastError: fmt.Sprintf("媒体进程已退出 (pid %d)", pF.Pid),
|
||
}, nil
|
||
}
|
||
return Status{Running: true, Pid: pF.Pid, ConfigPath: pF.ConfigPath, StartedAtMS: pF.StartedAtMS}, nil
|
||
}
|
||
|
||
func (c *Controller) Version() (string, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
if strings.TrimSpace(c.proc.ExecPath) == "" {
|
||
return "", errors.New("exec_path is empty")
|
||
}
|
||
cmd := exec.Command(c.proc.ExecPath, "--version")
|
||
if strings.TrimSpace(c.proc.WorkDir) != "" {
|
||
cmd.Dir = c.proc.WorkDir
|
||
}
|
||
out, err := cmd.CombinedOutput()
|
||
if err != nil {
|
||
return "", fmt.Errorf("get version failed: %w", err)
|
||
}
|
||
ver := strings.TrimSpace(string(out))
|
||
if ver == "" {
|
||
return "", errors.New("version output empty")
|
||
}
|
||
return ver, nil
|
||
}
|
||
|
||
func (c *Controller) Start(configName string) (Status, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
resolved, err := c.resolveConfigPath(configName)
|
||
if err != nil {
|
||
return Status{}, err
|
||
}
|
||
|
||
pf, _ := c.readPidFile()
|
||
if pf != nil {
|
||
alive, _ := isAlive(pf.Pid)
|
||
if alive {
|
||
if filepath.Clean(pf.ConfigPath) == filepath.Clean(resolved) {
|
||
return Status{Running: true, Pid: pf.Pid, ConfigPath: pf.ConfigPath, StartedAtMS: pf.StartedAtMS}, nil
|
||
}
|
||
return Status{}, fmt.Errorf("%w: already running with config %s", ErrConflict, pf.ConfigPath)
|
||
}
|
||
_ = os.Remove(c.proc.PidFile)
|
||
}
|
||
|
||
pid, err := startProcess(c.proc.ExecPath, c.proc.WorkDir, resolved)
|
||
if err != nil {
|
||
return Status{}, err
|
||
}
|
||
|
||
pf2 := pidFile{Pid: pid, ConfigPath: resolved, StartedAtMS: time.Now().UnixMilli()}
|
||
b, _ := json.Marshal(pf2)
|
||
b = append(b, '\n')
|
||
if err := files.WriteFileAtomic(c.proc.PidFile, b, 0o644); err != nil {
|
||
_ = stopProcess(pid, 1*time.Second)
|
||
return Status{}, fmt.Errorf("write pid file: %w", err)
|
||
}
|
||
return Status{Running: true, Pid: pid, ConfigPath: resolved, StartedAtMS: pf2.StartedAtMS}, nil
|
||
}
|
||
|
||
func (c *Controller) Stop() (Status, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
pf, err := c.readPidFile()
|
||
if err != nil {
|
||
return Status{}, err
|
||
}
|
||
if pf == nil {
|
||
return Status{Running: false}, nil
|
||
}
|
||
|
||
alive, _ := isAlive(pf.Pid)
|
||
if !alive {
|
||
_ = os.Remove(c.proc.PidFile)
|
||
return Status{Running: false, Pid: pf.Pid, ConfigPath: pf.ConfigPath, StartedAtMS: pf.StartedAtMS}, nil
|
||
}
|
||
|
||
if err := stopProcess(pf.Pid, time.Duration(c.proc.GracefulTimeoutMS)*time.Millisecond); err != nil {
|
||
return Status{}, err
|
||
}
|
||
_ = os.Remove(c.proc.PidFile)
|
||
return Status{Running: false, Pid: pf.Pid, ConfigPath: pf.ConfigPath, StartedAtMS: pf.StartedAtMS}, nil
|
||
}
|
||
|
||
func (c *Controller) Restart(configName string) (Status, error) {
|
||
_, _ = c.Stop()
|
||
return c.Start(configName)
|
||
}
|
||
|
||
func (c *Controller) resolveConfigPath(name string) (string, error) {
|
||
n := strings.TrimSpace(name)
|
||
if n == "" {
|
||
if strings.TrimSpace(c.defCfg) == "" {
|
||
return "", fmt.Errorf("%w: default config_path is empty", ErrInvalidConfig)
|
||
}
|
||
return c.defCfg, nil
|
||
}
|
||
if strings.Contains(n, "..") || strings.ContainsAny(n, "/\\") {
|
||
return "", fmt.Errorf("%w: contains invalid characters", ErrInvalidConfig)
|
||
}
|
||
if !strings.HasSuffix(n, ".json") {
|
||
n += ".json"
|
||
}
|
||
base := strings.TrimSpace(c.proc.ConfigsDir)
|
||
if base == "" {
|
||
return "", fmt.Errorf("%w: configs_dir is empty", ErrInvalidConfig)
|
||
}
|
||
p := filepath.Join(base, n)
|
||
st, err := os.Stat(p)
|
||
if err != nil {
|
||
if os.IsNotExist(err) {
|
||
return "", fmt.Errorf("%w: %s", ErrConfigNotFound, p)
|
||
}
|
||
return "", fmt.Errorf("stat config: %w", err)
|
||
}
|
||
if st.IsDir() {
|
||
return "", fmt.Errorf("%w: is a directory", ErrInvalidConfig)
|
||
}
|
||
return p, nil
|
||
}
|
||
|
||
func (c *Controller) UpdateBinary(r io.Reader, contentLength int64, expectedSha256 string) (BinaryUpdateResult, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
if contentLength <= 0 {
|
||
return BinaryUpdateResult{}, errors.New("missing Content-Length")
|
||
}
|
||
if strings.TrimSpace(c.proc.ExecPath) == "" {
|
||
return BinaryUpdateResult{}, errors.New("exec_path is empty")
|
||
}
|
||
|
||
pf, _ := c.readPidFile()
|
||
if pf != nil {
|
||
alive, _ := isAlive(pf.Pid)
|
||
if alive {
|
||
return BinaryUpdateResult{}, fmt.Errorf("%w: media-server is running", ErrConflict)
|
||
}
|
||
_ = os.Remove(c.proc.PidFile)
|
||
}
|
||
|
||
dir := filepath.Dir(c.proc.ExecPath)
|
||
if err := files.EnsureDir(dir, 0o755); err != nil {
|
||
return BinaryUpdateResult{}, err
|
||
}
|
||
|
||
f, err := os.CreateTemp(dir, ".tmp-*")
|
||
if err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("create temp: %w", err)
|
||
}
|
||
tmp := f.Name()
|
||
ok := false
|
||
defer func() {
|
||
_ = f.Close()
|
||
if !ok {
|
||
_ = os.Remove(tmp)
|
||
}
|
||
}()
|
||
|
||
h := sha256.New()
|
||
mw := io.MultiWriter(f, h)
|
||
if _, err := io.CopyN(mw, r, contentLength); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("read body: %w", err)
|
||
}
|
||
if err := f.Chmod(0o755); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("chmod temp: %w", err)
|
||
}
|
||
if err := f.Sync(); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("fsync temp: %w", err)
|
||
}
|
||
if err := f.Close(); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("close temp: %w", err)
|
||
}
|
||
|
||
sha := hex.EncodeToString(h.Sum(nil))
|
||
if expectedSha256 != "" && !strings.EqualFold(expectedSha256, sha) {
|
||
return BinaryUpdateResult{}, errors.New("sha256 mismatch")
|
||
}
|
||
|
||
backup := ""
|
||
if st, err := os.Stat(c.proc.ExecPath); err == nil && !st.IsDir() {
|
||
backup = fmt.Sprintf("%s.bak.%s", c.proc.ExecPath, time.Now().Format("20060102-150405"))
|
||
if err := os.Rename(c.proc.ExecPath, backup); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("backup old binary: %w", err)
|
||
}
|
||
}
|
||
|
||
if err := files.ReplaceFile(tmp, c.proc.ExecPath); err != nil {
|
||
if backup != "" {
|
||
_ = os.Rename(backup, c.proc.ExecPath)
|
||
}
|
||
return BinaryUpdateResult{}, err
|
||
}
|
||
|
||
st, err := os.Stat(c.proc.ExecPath)
|
||
if err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("stat binary: %w", err)
|
||
}
|
||
|
||
ok = true
|
||
return BinaryUpdateResult{
|
||
Path: filepath.ToSlash(c.proc.ExecPath),
|
||
Sha256: sha,
|
||
Size: st.Size(),
|
||
MtimeMS: st.ModTime().UnixMilli(),
|
||
BackupPath: filepath.ToSlash(backup),
|
||
}, nil
|
||
}
|
||
|
||
func (c *Controller) RollbackBinary(backupPath string) (BinaryUpdateResult, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
if strings.TrimSpace(backupPath) == "" {
|
||
return BinaryUpdateResult{}, errors.New("backup_path is empty")
|
||
}
|
||
if strings.TrimSpace(c.proc.ExecPath) == "" {
|
||
return BinaryUpdateResult{}, errors.New("exec_path is empty")
|
||
}
|
||
|
||
pf, _ := c.readPidFile()
|
||
if pf != nil {
|
||
alive, _ := isAlive(pf.Pid)
|
||
if alive {
|
||
return BinaryUpdateResult{}, fmt.Errorf("%w: media-server is running", ErrConflict)
|
||
}
|
||
_ = os.Remove(c.proc.PidFile)
|
||
}
|
||
|
||
if st, err := os.Stat(backupPath); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("stat backup: %w", err)
|
||
} else if st.IsDir() {
|
||
return BinaryUpdateResult{}, errors.New("backup_path is a directory")
|
||
}
|
||
|
||
oldBackup := ""
|
||
if st, err := os.Stat(c.proc.ExecPath); err == nil && !st.IsDir() {
|
||
oldBackup = fmt.Sprintf("%s.bak.rollback.%s", c.proc.ExecPath, time.Now().Format("20060102-150405"))
|
||
if err := os.Rename(c.proc.ExecPath, oldBackup); err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("backup current binary: %w", err)
|
||
}
|
||
}
|
||
|
||
if err := files.ReplaceFile(backupPath, c.proc.ExecPath); err != nil {
|
||
if oldBackup != "" {
|
||
_ = os.Rename(oldBackup, c.proc.ExecPath)
|
||
}
|
||
return BinaryUpdateResult{}, err
|
||
}
|
||
|
||
st, err := os.Stat(c.proc.ExecPath)
|
||
if err != nil {
|
||
return BinaryUpdateResult{}, fmt.Errorf("stat binary: %w", err)
|
||
}
|
||
|
||
sha, err := sha256File(c.proc.ExecPath)
|
||
if err != nil {
|
||
return BinaryUpdateResult{}, err
|
||
}
|
||
|
||
return BinaryUpdateResult{
|
||
Path: filepath.ToSlash(c.proc.ExecPath),
|
||
Sha256: sha,
|
||
Size: st.Size(),
|
||
MtimeMS: st.ModTime().UnixMilli(),
|
||
BackupPath: filepath.ToSlash(oldBackup),
|
||
}, nil
|
||
}
|
||
|
||
func (c *Controller) BinaryInfo() (BinaryUpdateResult, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
|
||
if strings.TrimSpace(c.proc.ExecPath) == "" {
|
||
return BinaryUpdateResult{}, errors.New("exec_path is empty")
|
||
}
|
||
st, err := os.Stat(c.proc.ExecPath)
|
||
if err != nil {
|
||
return BinaryUpdateResult{}, err
|
||
}
|
||
if st.IsDir() {
|
||
return BinaryUpdateResult{}, errors.New("exec_path is a directory")
|
||
}
|
||
sha, err := sha256File(c.proc.ExecPath)
|
||
if err != nil {
|
||
return BinaryUpdateResult{}, err
|
||
}
|
||
return BinaryUpdateResult{
|
||
Path: filepath.ToSlash(c.proc.ExecPath),
|
||
Sha256: sha,
|
||
Size: st.Size(),
|
||
MtimeMS: st.ModTime().UnixMilli(),
|
||
}, nil
|
||
}
|
||
|
||
func (c *Controller) readPidFile() (*pidFile, error) {
|
||
b, err := os.ReadFile(c.proc.PidFile)
|
||
if err != nil {
|
||
if os.IsNotExist(err) {
|
||
return nil, nil
|
||
}
|
||
return nil, fmt.Errorf("read pid file: %w", err)
|
||
}
|
||
var pf pidFile
|
||
if err := json.Unmarshal(b, &pf); err != nil {
|
||
return nil, fmt.Errorf("parse pid file: %w", err)
|
||
}
|
||
if pf.Pid <= 0 {
|
||
return nil, fmt.Errorf("pid file invalid pid: %d", pf.Pid)
|
||
}
|
||
return &pf, nil
|
||
}
|
||
|
||
func sha256File(path string) (string, error) {
|
||
f, err := os.Open(path)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
defer f.Close()
|
||
H := sha256.New()
|
||
if _, err := io.Copy(H, f); err != nil {
|
||
return "", err
|
||
}
|
||
return hex.EncodeToString(H.Sum(nil)), nil
|
||
}
|