steve
b054e7b987
version.go: add a comment block explaining why Phase 5's wire changes (CommandRunPayload, ConfigUpdatePayload, RepoStatsPayload reshapes) did not bump CurrentProtocolVersion — lockstep deploy, no rolling-upgrade path, smoke env restage enforces it. Notes where a version bump to 2 would be required if a multi-version path is ever introduced. cmd/agent/main.go: document why the JobForget handler hard-errors on empty ForgetGroups rather than falling back to a single-policy form. The maintenance ticker is the only writer and always populates the field; the fallback was specced but skipped given lockstep deploy.
448 lines
14 KiB
Go
448 lines
14 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"syscall"
|
|
"time"
|
|
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/agent/config"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/agent/runner"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/agent/scheduler"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/agent/secrets"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/agent/sysinfo"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/agent/wsclient"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
|
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/restic"
|
|
)
|
|
|
|
var version = "dev"
|
|
|
|
func main() {
|
|
if err := run(); err != nil {
|
|
slog.Error("agent fatal", "err", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func run() error {
|
|
configPath := flag.String("config", config.DefaultPath(), "path to agent.yaml")
|
|
enrollServer := flag.String("enroll-server", "", "server URL (used with -enroll-token to perform first-run enrollment)")
|
|
enrollToken := flag.String("enroll-token", "", "one-time enrollment token (operator copies this from the UI)")
|
|
showVersion := flag.Bool("version", false, "print version and exit")
|
|
flag.Parse()
|
|
|
|
if *showVersion {
|
|
fmt.Println("restic-manager-agent", version)
|
|
return nil
|
|
}
|
|
|
|
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
|
slog.SetDefault(logger)
|
|
|
|
cfg, err := config.Load(*configPath)
|
|
if err != nil {
|
|
return fmt.Errorf("config: %w", err)
|
|
}
|
|
|
|
if *enrollToken != "" {
|
|
if *enrollServer == "" {
|
|
return errors.New("enrollment: -enroll-server is required with -enroll-token")
|
|
}
|
|
return doEnroll(*enrollServer, *enrollToken, cfg, version)
|
|
}
|
|
|
|
if !cfg.Enrolled() {
|
|
return fmt.Errorf("agent is not enrolled; run with -enroll-server and -enroll-token first (config %q)", *configPath)
|
|
}
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
snap, err := sysinfo.Collect(ctx, cfg.ResticPath)
|
|
if err != nil {
|
|
return fmt.Errorf("sysinfo: %w", err)
|
|
}
|
|
slog.Info("agent starting",
|
|
"version", version,
|
|
"host_id", cfg.HostID,
|
|
"server", cfg.ServerURL,
|
|
"restic_version", snap.ResticVersion,
|
|
"protocol_version", snap.ProtocolVersion,
|
|
)
|
|
|
|
resticBin, _ := restic.Locate(cfg.ResticPath) // empty is fine; commands fail with a clear error later
|
|
|
|
// Open the secrets store. If the agent is enrolled but has no
|
|
// secrets key yet (legacy YAML), mint one and migrate any
|
|
// plaintext repo fields into the encrypted blob.
|
|
sec, err := openSecretsStore(cfg)
|
|
if err != nil {
|
|
return fmt.Errorf("secrets: %w", err)
|
|
}
|
|
|
|
wsCfg := wsclient.Config{
|
|
ServerURL: cfg.ServerURL,
|
|
AgentToken: cfg.AgentToken,
|
|
HostID: cfg.HostID,
|
|
CertPinSHA256: cfg.CertPinSHA256,
|
|
HelloPayload: api.HelloPayload{
|
|
ProtocolVersion: snap.ProtocolVersion,
|
|
AgentVersion: version,
|
|
ResticVersion: snap.ResticVersion,
|
|
Hostname: snap.Hostname,
|
|
OS: snap.OS,
|
|
Arch: snap.Arch,
|
|
},
|
|
}
|
|
|
|
d := &dispatcher{
|
|
resticBin: resticBin,
|
|
secrets: sec,
|
|
scheduler: scheduler.New(),
|
|
}
|
|
if err := wsclient.Run(ctx, wsCfg, d.handle); err != nil {
|
|
return fmt.Errorf("ws run: %w", err)
|
|
}
|
|
slog.Info("agent shutting down")
|
|
return nil
|
|
}
|
|
|
|
// openSecretsStore opens (or one-time migrates) the agent's encrypted
|
|
// secrets file. Side effects:
|
|
// - mints SecretsKey if absent and persists agent.yaml.
|
|
// - if legacy plaintext repo_url/repo_password sit in agent.yaml,
|
|
// copies them into secrets.enc and clears the YAML fields on
|
|
// the next save.
|
|
func openSecretsStore(cfg *config.Config) (*secrets.Store, error) {
|
|
if err := cfg.EnsureSecretsKey(); err != nil {
|
|
return nil, err
|
|
}
|
|
keyBytes, err := cfg.SecretsKeyBytes()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
st, err := secrets.New(cfg.ResolvedSecretsPath(), keyBytes)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
migrated := false
|
|
if cfg.LegacyRepoURL != "" || cfg.LegacyRepoPassword != "" {
|
|
cur, _ := st.Load() // empty Repo on first run is fine
|
|
if cur.URL == "" {
|
|
cur.URL = cfg.LegacyRepoURL
|
|
}
|
|
if cur.Password == "" {
|
|
cur.Password = cfg.LegacyRepoPassword
|
|
}
|
|
if err := st.Save(cur); err != nil {
|
|
return nil, fmt.Errorf("migrate legacy creds into secrets.enc: %w", err)
|
|
}
|
|
cfg.LegacyRepoURL = ""
|
|
cfg.LegacyRepoPassword = ""
|
|
migrated = true
|
|
slog.Info("agent: migrated legacy plaintext repo creds into secrets.enc")
|
|
}
|
|
|
|
// Persist key (and the cleared legacy fields) regardless of
|
|
// whether we migrated, in case we just minted SecretsKey.
|
|
if migrated || cfg.SecretsKey != "" {
|
|
if err := cfg.Save(); err != nil {
|
|
return nil, fmt.Errorf("persist agent config: %w", err)
|
|
}
|
|
}
|
|
return st, nil
|
|
}
|
|
|
|
// dispatcher closes over the long-lived agent settings (restic path
|
|
// + secrets handle) so handle() can spawn the runner without
|
|
// re-loading config every time. Repo creds are read fresh from the
|
|
// secrets store on each job — config.update writes through to disk,
|
|
// so a job dispatched in the same session sees the latest values.
|
|
type dispatcher struct {
|
|
resticBin string
|
|
secrets *secrets.Store
|
|
scheduler *scheduler.Scheduler
|
|
}
|
|
|
|
func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.Sender) error {
|
|
switch env.Type {
|
|
case api.MsgCommandRun:
|
|
var p api.CommandRunPayload
|
|
if err := env.UnmarshalPayload(&p); err != nil {
|
|
return fmt.Errorf("command.run: %w", err)
|
|
}
|
|
return d.runJob(ctx, p, tx)
|
|
|
|
case api.MsgCommandCancel:
|
|
// TODO(P2): cancellation requires keeping a job→cancelFunc map.
|
|
slog.Info("ws agent: command.cancel received (cancellation lands in P2)", "id", env.ID)
|
|
|
|
case api.MsgScheduleSet:
|
|
var p api.ScheduleSetPayload
|
|
if err := env.UnmarshalPayload(&p); err != nil {
|
|
return fmt.Errorf("schedule.set: %w", err)
|
|
}
|
|
// scheduler.Apply rebuilds the local cron from scratch and
|
|
// emits schedule.ack via tx. Async-safe: tx may have to wait
|
|
// briefly on the connection's writeMu, but the read loop
|
|
// keeps draining other messages.
|
|
go d.scheduler.Apply(p, tx)
|
|
|
|
case api.MsgConfigUpdate:
|
|
var p api.ConfigUpdatePayload
|
|
_ = env.UnmarshalPayload(&p)
|
|
slot := p.Slot
|
|
if slot == "" {
|
|
slot = "repo"
|
|
}
|
|
switch slot {
|
|
case "repo":
|
|
// Merge with whatever's already in secrets.enc — empty fields
|
|
// in the push mean "leave alone." Atomic write underneath.
|
|
cur, err := d.secrets.Load()
|
|
if err != nil {
|
|
slog.Error("ws agent: load secrets for merge", "err", err)
|
|
return nil
|
|
}
|
|
changed := false
|
|
if p.RepoURL != "" && p.RepoURL != cur.URL {
|
|
cur.URL = p.RepoURL
|
|
changed = true
|
|
}
|
|
if p.RepoUsername != "" && p.RepoUsername != cur.Username {
|
|
cur.Username = p.RepoUsername
|
|
changed = true
|
|
}
|
|
if p.RepoPassword != "" && p.RepoPassword != cur.Password {
|
|
cur.Password = p.RepoPassword
|
|
changed = true
|
|
}
|
|
if changed {
|
|
if err := d.secrets.Save(cur); err != nil {
|
|
slog.Error("ws agent: persist secrets", "err", err)
|
|
return nil
|
|
}
|
|
slog.Info("ws agent: repo credentials updated via config.update")
|
|
}
|
|
case "admin":
|
|
cur, err := d.secrets.LoadAdmin()
|
|
if err != nil && !errors.Is(err, secrets.ErrNoAdmin) {
|
|
slog.Error("ws agent: load admin secrets", "err", err)
|
|
return nil
|
|
}
|
|
// ErrNoAdmin is not an error here — we are creating the slot.
|
|
changed := false
|
|
if p.RepoURL != "" && p.RepoURL != cur.URL {
|
|
cur.URL = p.RepoURL
|
|
changed = true
|
|
}
|
|
if p.RepoUsername != "" && p.RepoUsername != cur.Username {
|
|
cur.Username = p.RepoUsername
|
|
changed = true
|
|
}
|
|
if p.RepoPassword != "" && p.RepoPassword != cur.Password {
|
|
cur.Password = p.RepoPassword
|
|
changed = true
|
|
}
|
|
if changed {
|
|
if err := d.secrets.SaveAdmin(cur); err != nil {
|
|
slog.Error("ws agent: persist admin secrets", "err", err)
|
|
return nil
|
|
}
|
|
slog.Info("ws agent: admin credentials updated via config.update")
|
|
}
|
|
default:
|
|
slog.Warn("ws agent: unknown config.update slot, ignoring", "slot", p.Slot)
|
|
}
|
|
|
|
case api.MsgAgentUpdateAvail:
|
|
var p api.AgentUpdateAvailablePayload
|
|
_ = env.UnmarshalPayload(&p)
|
|
slog.Info("ws agent: update available", "version", p.LatestVersion, "url", p.PackageURL)
|
|
|
|
default:
|
|
slog.Debug("ws agent: ignored message", "type", env.Type)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// runJob spawns a runner for one job. We launch a goroutine so the
|
|
// WS read loop keeps draining messages while restic chugs along.
|
|
func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsclient.Sender) error {
|
|
if d.resticBin == "" {
|
|
return fmt.Errorf("restic binary not located on this agent")
|
|
}
|
|
creds, err := d.secrets.Load()
|
|
if err != nil {
|
|
return fmt.Errorf("load repo credentials: %w", err)
|
|
}
|
|
if creds.Empty() {
|
|
return fmt.Errorf("repo credentials not configured (waiting for server config.update push)")
|
|
}
|
|
// r is the everyday runner — bound to the host's repo
|
|
// (append-only) credentials. Reused by every kind except
|
|
// JobPrune, which builds its own runner against the
|
|
// admin-credentials slot when p.RequiresAdminCreds is set
|
|
// (admin creds are not loaded for any other kind, so they're
|
|
// not on r). If you find yourself adding a new JobKind that
|
|
// needs delete authority, mirror the JobPrune pattern below
|
|
// — don't try to overload r.
|
|
r := runner.New(runner.Config{
|
|
ResticBin: d.resticBin,
|
|
RepoURL: creds.URL,
|
|
RepoUsername: creds.Username,
|
|
RepoPassword: creds.Password,
|
|
}, tx, time.Second)
|
|
|
|
switch p.Kind {
|
|
case api.JobBackup:
|
|
// Includes/Excludes/Tag come from the source group resolved
|
|
// server-side. Args is preserved for backwards compatibility:
|
|
// if the server sends only Args (older shape) we fall back to
|
|
// treating it as the paths list with no tag.
|
|
paths := p.Includes
|
|
if len(paths) == 0 {
|
|
paths = p.Args
|
|
}
|
|
var tags []string
|
|
if p.Tag != "" {
|
|
tags = []string{p.Tag}
|
|
}
|
|
slog.Info("agent: accepting backup job",
|
|
"job_id", p.JobID, "paths", paths, "excludes", p.Excludes, "tag", p.Tag)
|
|
go func() {
|
|
if err := r.RunBackup(ctx, p.JobID, paths, p.Excludes, tags); err != nil {
|
|
slog.Warn("agent: backup job failed", "job_id", p.JobID, "err", err)
|
|
return
|
|
}
|
|
slog.Info("agent: backup job complete", "job_id", p.JobID)
|
|
}()
|
|
case api.JobInit:
|
|
slog.Info("agent: accepting init job", "job_id", p.JobID)
|
|
go func() {
|
|
if err := r.RunInit(ctx, p.JobID); err != nil {
|
|
slog.Warn("agent: init job failed", "job_id", p.JobID, "err", err)
|
|
return
|
|
}
|
|
slog.Info("agent: init job complete", "job_id", p.JobID)
|
|
}()
|
|
case api.JobForget:
|
|
if len(p.ForgetGroups) == 0 {
|
|
// Hard-error rather than fall back to a single-policy form:
|
|
// the server-side dispatch path (maintenance ticker) is the
|
|
// only writer of forget command.run today, and it always
|
|
// populates ForgetGroups. A backwards-compatible single-
|
|
// policy fallback was specced but skipped — see the
|
|
// Phase 5 plan rationale and version.go's lockstep-deploy
|
|
// note for why.
|
|
return fmt.Errorf("forget: command.run carried no forget_groups (server didn't populate them)")
|
|
}
|
|
groups := make([]restic.ForgetGroup, 0, len(p.ForgetGroups))
|
|
for _, g := range p.ForgetGroups {
|
|
groups = append(groups, restic.ForgetGroup{
|
|
Tag: g.Tag,
|
|
Policy: restic.ForgetPolicy{
|
|
KeepLast: g.Policy.KeepLast,
|
|
KeepHourly: g.Policy.KeepHourly,
|
|
KeepDaily: g.Policy.KeepDaily,
|
|
KeepWeekly: g.Policy.KeepWeekly,
|
|
KeepMonthly: g.Policy.KeepMonthly,
|
|
KeepYearly: g.Policy.KeepYearly,
|
|
},
|
|
})
|
|
}
|
|
slog.Info("agent: accepting forget job", "job_id", p.JobID, "groups", len(groups))
|
|
go func() {
|
|
if err := r.RunForget(ctx, p.JobID, groups); err != nil {
|
|
slog.Warn("agent: forget job failed", "job_id", p.JobID, "err", err)
|
|
return
|
|
}
|
|
slog.Info("agent: forget job complete", "job_id", p.JobID)
|
|
}()
|
|
case api.JobPrune:
|
|
// Prune may require admin creds (delete authority on rest-server).
|
|
runCreds := creds
|
|
if p.RequiresAdminCreds {
|
|
ac, err := d.secrets.LoadAdmin()
|
|
if err != nil {
|
|
return fmt.Errorf("prune: admin creds not configured (server didn't push them): %w", err)
|
|
}
|
|
if ac.Empty() {
|
|
return fmt.Errorf("prune: admin creds incomplete")
|
|
}
|
|
runCreds = ac
|
|
}
|
|
prr := runner.New(runner.Config{
|
|
ResticBin: d.resticBin,
|
|
RepoURL: runCreds.URL,
|
|
RepoUsername: runCreds.Username,
|
|
RepoPassword: runCreds.Password,
|
|
}, tx, time.Second)
|
|
slog.Info("agent: accepting prune job", "job_id", p.JobID, "admin_creds", p.RequiresAdminCreds)
|
|
go func() {
|
|
if err := prr.RunPrune(ctx, p.JobID); err != nil {
|
|
slog.Warn("agent: prune job failed", "job_id", p.JobID, "err", err)
|
|
}
|
|
}()
|
|
case api.JobCheck:
|
|
subset := 0
|
|
if len(p.Args) > 0 {
|
|
subset, _ = strconv.Atoi(p.Args[0])
|
|
}
|
|
slog.Info("agent: accepting check job", "job_id", p.JobID, "subset_pct", subset)
|
|
go func() {
|
|
if err := r.RunCheck(ctx, p.JobID, subset); err != nil {
|
|
slog.Warn("agent: check job failed", "job_id", p.JobID, "err", err)
|
|
}
|
|
}()
|
|
case api.JobUnlock:
|
|
slog.Info("agent: accepting unlock job", "job_id", p.JobID)
|
|
go func() {
|
|
if err := r.RunUnlock(ctx, p.JobID); err != nil {
|
|
slog.Warn("agent: unlock job failed", "job_id", p.JobID, "err", err)
|
|
}
|
|
}()
|
|
default:
|
|
return fmt.Errorf("kind %q not implemented yet (Phase 2 lands the rest)", p.Kind)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func doEnroll(serverURL, token string, cfg *config.Config, agentVersion string) error {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
|
defer cancel()
|
|
|
|
snap, err := sysinfo.Collect(ctx, cfg.ResticPath)
|
|
if err != nil {
|
|
return fmt.Errorf("sysinfo: %w", err)
|
|
}
|
|
res, err := wsclient.Enroll(ctx, serverURL, wsclient.EnrollRequest{
|
|
Token: token,
|
|
HostName: snap.Hostname,
|
|
OS: snap.OS,
|
|
Arch: snap.Arch,
|
|
AgentVersion: agentVersion,
|
|
ResticVersion: snap.ResticVersion,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("enroll: %w", err)
|
|
}
|
|
cfg.ServerURL = serverURL
|
|
cfg.HostID = res.HostID
|
|
cfg.AgentToken = res.AgentToken
|
|
cfg.CertPinSHA256 = res.CertPinSHA256
|
|
if err := cfg.Save(); err != nil {
|
|
return fmt.Errorf("save config: %w", err)
|
|
}
|
|
fmt.Fprintf(os.Stderr, "enrolled as host %s on %s\n", res.HostID, serverURL)
|
|
return nil
|
|
}
|