fix(forget): populate retention groups for manual runs
CI / Test (store) (pull_request) Successful in 5s
CI / Lint (pull_request) Successful in 10s
CI / Build (windows/amd64) (pull_request) Successful in 7s
CI / Build (linux/amd64) (pull_request) Successful in 8s
CI / Build (linux/arm64) (pull_request) Successful in 7s
CI / Test (rest) (pull_request) Successful in 39s
CI / Test (server-http) (pull_request) Successful in 1m37s
e2e / Playwright vs docker-compose (pull_request) Successful in 1m16s
CI / Test (store) (pull_request) Successful in 5s
CI / Lint (pull_request) Successful in 10s
CI / Build (windows/amd64) (pull_request) Successful in 7s
CI / Build (linux/amd64) (pull_request) Successful in 8s
CI / Build (linux/arm64) (pull_request) Successful in 7s
CI / Test (rest) (pull_request) Successful in 39s
CI / Test (server-http) (pull_request) Successful in 1m37s
e2e / Playwright vs docker-compose (pull_request) Successful in 1m16s
This commit is contained in:
+7
-2
@@ -601,6 +601,11 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc
|
|||||||
failJob(p, tx, "forget: command.run carried no forget_groups (server didn't populate them)")
|
failJob(p, tx, "forget: command.run carried no forget_groups (server didn't populate them)")
|
||||||
return fmt.Errorf("forget: command.run carried no forget_groups (server didn't populate them)")
|
return fmt.Errorf("forget: command.run carried no forget_groups (server didn't populate them)")
|
||||||
}
|
}
|
||||||
|
if len(p.Args) > 1 || (len(p.Args) == 1 && p.Args[0] != "--dry-run") {
|
||||||
|
failJob(p, tx, "forget: command.run carried unsupported arguments")
|
||||||
|
return fmt.Errorf("forget: command.run carried unsupported arguments")
|
||||||
|
}
|
||||||
|
dryRun := len(p.Args) == 1
|
||||||
groups := make([]restic.ForgetGroup, 0, len(p.ForgetGroups))
|
groups := make([]restic.ForgetGroup, 0, len(p.ForgetGroups))
|
||||||
for _, g := range p.ForgetGroups {
|
for _, g := range p.ForgetGroups {
|
||||||
groups = append(groups, restic.ForgetGroup{
|
groups = append(groups, restic.ForgetGroup{
|
||||||
@@ -615,9 +620,9 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc
|
|||||||
},
|
},
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
slog.Info("agent: accepting forget job", "job_id", p.JobID, "groups", len(groups))
|
slog.Info("agent: accepting forget job", "job_id", p.JobID, "groups", len(groups), "dry_run", dryRun)
|
||||||
spawn("forget", func(jobCtx context.Context) error {
|
spawn("forget", func(jobCtx context.Context) error {
|
||||||
return r.RunForget(jobCtx, p.JobID, groups)
|
return r.RunForget(jobCtx, p.JobID, groups, dryRun)
|
||||||
})
|
})
|
||||||
case api.JobPrune:
|
case api.JobPrune:
|
||||||
// Prune may require admin creds (delete authority on rest-server).
|
// Prune may require admin creds (delete authority on rest-server).
|
||||||
|
|||||||
@@ -274,13 +274,13 @@ func (r *Runner) RunInit(ctx context.Context, jobID string) error {
|
|||||||
// snapshot projection (forget rewrites the snapshot index — the
|
// snapshot projection (forget rewrites the snapshot index — the
|
||||||
// host's snapshot list shrinks). Snapshot refresh runs once after
|
// host's snapshot list shrinks). Snapshot refresh runs once after
|
||||||
// every group completes, not per-group.
|
// every group completes, not per-group.
|
||||||
func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.ForgetGroup) error {
|
func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.ForgetGroup, dryRun bool) error {
|
||||||
startedAt := time.Now().UTC()
|
startedAt := time.Now().UTC()
|
||||||
r.sendStarted(jobID, api.JobForget, startedAt)
|
r.sendStarted(jobID, api.JobForget, startedAt)
|
||||||
|
|
||||||
env := r.resticEnv()
|
env := r.resticEnv()
|
||||||
var seq atomic.Int64
|
var seq atomic.Int64
|
||||||
err := env.RunForget(ctx, groups, r.streamHandler(jobID, &seq))
|
err := env.RunForget(ctx, groups, dryRun, r.streamHandler(jobID, &seq))
|
||||||
finishedAt := time.Now().UTC()
|
finishedAt := time.Now().UTC()
|
||||||
r.sendFinished(ctx, jobID, finishedAt, err, nil)
|
r.sendFinished(ctx, jobID, finishedAt, err, nil)
|
||||||
|
|
||||||
|
|||||||
@@ -398,7 +398,7 @@ esac
|
|||||||
Tag: "documents",
|
Tag: "documents",
|
||||||
Policy: restic.ForgetPolicy{KeepLast: &keepLast},
|
Policy: restic.ForgetPolicy{KeepLast: &keepLast},
|
||||||
}}
|
}}
|
||||||
if err := r.RunForget(context.Background(), "job-forget", groups); err != nil {
|
if err := r.RunForget(context.Background(), "job-forget", groups, false); err != nil {
|
||||||
t.Fatalf("RunForget: %v", err)
|
t.Fatalf("RunForget: %v", err)
|
||||||
}
|
}
|
||||||
_ = firstEnvOfType(t, tx.envs, api.MsgJobStarted)
|
_ = firstEnvOfType(t, tx.envs, api.MsgJobStarted)
|
||||||
|
|||||||
@@ -322,7 +322,7 @@ type ForgetGroup struct {
|
|||||||
// any keep-* would delete every snapshot in the tagged set).
|
// any keep-* would delete every snapshot in the tagged set).
|
||||||
// Returns the first error encountered, or nil when every group runs
|
// Returns the first error encountered, or nil when every group runs
|
||||||
// to a clean exit.
|
// to a clean exit.
|
||||||
func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, handle LineHandler) error {
|
func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, dryRun bool, handle LineHandler) error {
|
||||||
if len(groups) == 0 {
|
if len(groups) == 0 {
|
||||||
return fmt.Errorf("restic forget: refusing to run with no groups (would be a no-op)")
|
return fmt.Errorf("restic forget: refusing to run with no groups (would be a no-op)")
|
||||||
}
|
}
|
||||||
@@ -332,6 +332,9 @@ func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, handle LineHan
|
|||||||
}
|
}
|
||||||
args := []string{"forget", "--json", "--tag", g.Tag}
|
args := []string{"forget", "--json", "--tag", g.Tag}
|
||||||
args = append(args, g.Policy.args()...)
|
args = append(args, g.Policy.args()...)
|
||||||
|
if dryRun {
|
||||||
|
args = append(args, "--dry-run")
|
||||||
|
}
|
||||||
cmd := e.resticCmd(ctx, args...)
|
cmd := e.resticCmd(ctx, args...)
|
||||||
if err := runWithPump(cmd, handle); err != nil {
|
if err := runWithPump(cmd, handle); err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
@@ -60,6 +60,26 @@ func TestRunPruneInvokesPrune(t *testing.T) {
|
|||||||
t.Fatalf("expected 'prune' in captured output; got: %v", *lines)
|
t.Fatalf("expected 'prune' in captured output; got: %v", *lines)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRunForgetDryRunArgument(t *testing.T) {
|
||||||
|
bin := setupScriptBin(t, `echo "$@"`)
|
||||||
|
env := Env{Bin: bin}
|
||||||
|
lines, h := captureLines()
|
||||||
|
keepLast := 1
|
||||||
|
groups := []ForgetGroup{{
|
||||||
|
Tag: "documents",
|
||||||
|
Policy: ForgetPolicy{KeepLast: &keepLast},
|
||||||
|
}}
|
||||||
|
if err := env.RunForget(context.Background(), groups, true, h); err != nil {
|
||||||
|
t.Fatalf("RunForget: %v", err)
|
||||||
|
}
|
||||||
|
for _, line := range *lines {
|
||||||
|
if strings.Contains(line, "forget --json --tag documents --keep-last 1 --dry-run") {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t.Fatalf("expected forget invocation with --dry-run; got: %v", *lines)
|
||||||
|
}
|
||||||
|
|
||||||
// --- B2: RunCheck ---
|
// --- B2: RunCheck ---
|
||||||
|
|
||||||
func TestRunCheckLockSniff(t *testing.T) {
|
func TestRunCheckLockSniff(t *testing.T) {
|
||||||
|
|||||||
@@ -65,10 +65,32 @@ func (s *Server) handleRunNow(w stdhttp.ResponseWriter, r *stdhttp.Request) {
|
|||||||
func (s *Server) dispatchJob(ctx context.Context, user *store.User,
|
func (s *Server) dispatchJob(ctx context.Context, user *store.User,
|
||||||
hostID string, kind api.JobKind, args []string,
|
hostID string, kind api.JobKind, args []string,
|
||||||
) (res runNowResponse, status int, code, msg string) {
|
) (res runNowResponse, status int, code, msg string) {
|
||||||
return s.dispatchJobWithPayload(ctx, user, hostID, kind, nil, api.CommandRunPayload{
|
payload := api.CommandRunPayload{
|
||||||
Kind: kind,
|
Kind: kind,
|
||||||
Args: args,
|
Args: args,
|
||||||
})
|
}
|
||||||
|
if kind == api.JobForget {
|
||||||
|
if !validForgetArgs(args) {
|
||||||
|
return res, stdhttp.StatusBadRequest, "invalid_args",
|
||||||
|
"forget accepts no arguments other than --dry-run"
|
||||||
|
}
|
||||||
|
var ok bool
|
||||||
|
var err error
|
||||||
|
payload, ok, err = s.buildForgetPayloadForHost(ctx, hostID)
|
||||||
|
if err != nil {
|
||||||
|
return res, stdhttp.StatusInternalServerError, "internal", ""
|
||||||
|
}
|
||||||
|
if !ok {
|
||||||
|
return res, stdhttp.StatusUnprocessableEntity, "no_retention_policy",
|
||||||
|
"host has no source groups with a retention policy"
|
||||||
|
}
|
||||||
|
payload.Args = args
|
||||||
|
}
|
||||||
|
return s.dispatchJobWithPayload(ctx, user, hostID, kind, nil, payload)
|
||||||
|
}
|
||||||
|
|
||||||
|
func validForgetArgs(args []string) bool {
|
||||||
|
return len(args) == 0 || (len(args) == 1 && args[0] == "--dry-run")
|
||||||
}
|
}
|
||||||
|
|
||||||
// dispatchJobWithPayload is dispatchJob's variant that lets callers
|
// dispatchJobWithPayload is dispatchJob's variant that lets callers
|
||||||
|
|||||||
@@ -0,0 +1,106 @@
|
|||||||
|
package http
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
stdhttp "net/http"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/oklog/ulid/v2"
|
||||||
|
|
||||||
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
|
||||||
|
"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestRunNowForgetShipsRetentionGroupsAndDryRun(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
srv, ts, st := rawTestServer(t)
|
||||||
|
hostID, token := enrolHostForWS(t, srv, st, "manual-forget-host")
|
||||||
|
seedInitJob(t, st, hostID)
|
||||||
|
|
||||||
|
keepDaily := 7
|
||||||
|
if err := st.CreateSourceGroup(context.Background(), &store.SourceGroup{
|
||||||
|
ID: ulid.Make().String(),
|
||||||
|
HostID: hostID,
|
||||||
|
Name: "documents",
|
||||||
|
Includes: []string{"/home/documents"},
|
||||||
|
RetentionPolicy: store.RetentionPolicy{KeepDaily: &keepDaily},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("create source group: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
c := agentDial(t, srv, ts, hostID, token)
|
||||||
|
sendHello(t, c, "manual-forget-host")
|
||||||
|
_ = drainUntil(t, c, api.MsgScheduleSet)
|
||||||
|
|
||||||
|
body, err := json.Marshal(runNowRequest{Kind: api.JobForget, Args: []string{"--dry-run"}})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal request: %v", err)
|
||||||
|
}
|
||||||
|
req, err := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/jobs", bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("new request: %v", err)
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
req.AddCookie(loginAsAdmin(t, st))
|
||||||
|
res, err := stdhttp.DefaultClient.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("post run-now: %v", err)
|
||||||
|
}
|
||||||
|
defer res.Body.Close()
|
||||||
|
if res.StatusCode != stdhttp.StatusAccepted {
|
||||||
|
t.Fatalf("status: got %d, want %d", res.StatusCode, stdhttp.StatusAccepted)
|
||||||
|
}
|
||||||
|
|
||||||
|
got := readNextCommandRun(t, c, time.Now().Add(2*time.Second))
|
||||||
|
if got == nil {
|
||||||
|
t.Fatal("no command.run received")
|
||||||
|
}
|
||||||
|
if len(got.Args) != 1 || got.Args[0] != "--dry-run" {
|
||||||
|
t.Fatalf("Args: got %q, want [--dry-run]", got.Args)
|
||||||
|
}
|
||||||
|
if len(got.ForgetGroups) != 1 {
|
||||||
|
t.Fatalf("ForgetGroups: got %d, want 1", len(got.ForgetGroups))
|
||||||
|
}
|
||||||
|
group := got.ForgetGroups[0]
|
||||||
|
if group.Tag != "documents" || group.Policy.KeepDaily == nil || *group.Policy.KeepDaily != 7 {
|
||||||
|
t.Fatalf("ForgetGroups[0]: got %+v", group)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunNowForgetRejectsHostWithoutRetention(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
srv, ts, st := rawTestServer(t)
|
||||||
|
hostID, token := enrolHostForWS(t, srv, st, "no-manual-retention-host")
|
||||||
|
seedInitJob(t, st, hostID)
|
||||||
|
|
||||||
|
c := agentDial(t, srv, ts, hostID, token)
|
||||||
|
sendHello(t, c, "no-manual-retention-host")
|
||||||
|
_ = drainUntil(t, c, api.MsgScheduleSet)
|
||||||
|
|
||||||
|
body := []byte(`{"kind":"forget","args":["--dry-run"]}`)
|
||||||
|
req, err := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/jobs", bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("new request: %v", err)
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
req.AddCookie(loginAsAdmin(t, st))
|
||||||
|
res, err := stdhttp.DefaultClient.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("post run-now: %v", err)
|
||||||
|
}
|
||||||
|
defer res.Body.Close()
|
||||||
|
if res.StatusCode != stdhttp.StatusUnprocessableEntity {
|
||||||
|
t.Fatalf("status: got %d, want %d", res.StatusCode, stdhttp.StatusUnprocessableEntity)
|
||||||
|
}
|
||||||
|
|
||||||
|
var jobs int
|
||||||
|
if err := st.DB().QueryRow(`SELECT COUNT(*) FROM jobs WHERE host_id = ? AND kind = 'forget'`, hostID).Scan(&jobs); err != nil {
|
||||||
|
t.Fatalf("count forget jobs: %v", err)
|
||||||
|
}
|
||||||
|
if jobs != 0 {
|
||||||
|
t.Fatalf("forget jobs: got %d, want 0", jobs)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -37,7 +37,12 @@ func (s *Server) DispatchMaintenance(ctx context.Context, decisions []maintenanc
|
|||||||
}
|
}
|
||||||
switch d.Kind {
|
switch d.Kind {
|
||||||
case "forget":
|
case "forget":
|
||||||
payload, ok := s.buildForgetPayloadForHost(ctx, d.HostID)
|
payload, ok, err := s.buildForgetPayloadForHost(ctx, d.HostID)
|
||||||
|
if err != nil {
|
||||||
|
slog.Warn("maintenance: list source groups failed",
|
||||||
|
"host_id", d.HostID, "err", err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
if !ok {
|
if !ok {
|
||||||
slog.Info("maintenance: forget skipped — no source groups with retention",
|
slog.Info("maintenance: forget skipped — no source groups with retention",
|
||||||
"host_id", d.HostID)
|
"host_id", d.HostID)
|
||||||
@@ -88,11 +93,10 @@ func (s *Server) DispatchMaintenance(ctx context.Context, decisions []maintenanc
|
|||||||
// that has a non-empty retention policy and builds a CommandRunPayload
|
// that has a non-empty retention policy and builds a CommandRunPayload
|
||||||
// with ForgetGroups populated. Returns ok=false if the host has no
|
// with ForgetGroups populated. Returns ok=false if the host has no
|
||||||
// such groups (the dispatcher then skips this kind).
|
// such groups (the dispatcher then skips this kind).
|
||||||
func (s *Server) buildForgetPayloadForHost(ctx context.Context, hostID string) (api.CommandRunPayload, bool) {
|
func (s *Server) buildForgetPayloadForHost(ctx context.Context, hostID string) (api.CommandRunPayload, bool, error) {
|
||||||
groups, err := s.deps.Store.ListSourceGroupsByHost(ctx, hostID)
|
groups, err := s.deps.Store.ListSourceGroupsByHost(ctx, hostID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
slog.Warn("maintenance: list source groups failed", "host_id", hostID, "err", err)
|
return api.CommandRunPayload{}, false, err
|
||||||
return api.CommandRunPayload{}, false
|
|
||||||
}
|
}
|
||||||
fg := make([]api.ForgetGroup, 0, len(groups))
|
fg := make([]api.ForgetGroup, 0, len(groups))
|
||||||
for _, g := range groups {
|
for _, g := range groups {
|
||||||
@@ -105,9 +109,9 @@ func (s *Server) buildForgetPayloadForHost(ctx context.Context, hostID string) (
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
if len(fg) == 0 {
|
if len(fg) == 0 {
|
||||||
return api.CommandRunPayload{}, false
|
return api.CommandRunPayload{}, false, nil
|
||||||
}
|
}
|
||||||
return api.CommandRunPayload{ForgetGroups: fg}, true
|
return api.CommandRunPayload{ForgetGroups: fg}, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func isEmptyRetention(p store.RetentionPolicy) bool {
|
func isEmptyRetention(p store.RetentionPolicy) bool {
|
||||||
|
|||||||
Reference in New Issue
Block a user