1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
|
package cli
import (
"context"
"errors"
"fmt"
"io"
"sync/atomic"
"time"
"github.com/spf13/pflag"
"tsne.dev/hopper/internal/log"
"tsne.dev/hopper/internal/source"
)
var _ command = (*pruneCommand)(nil)
type pruneCommand struct {
flags *pflag.FlagSet
main *mainCommand
concurrency uint
prefix string
root string
olderThan duration
}
func newPruneCommand(main *mainCommand) *pruneCommand {
cmd := &pruneCommand{
flags: newFlagset("prune"),
main: main,
}
cmd.flags.UintVar(&cmd.concurrency, "concurrency", 0, "number of concurrent transfers (0 = auto-detect based on CPUs)")
registerPathPrefixFlag(cmd.flags, &cmd.prefix, "prefix", "", "path prefix within the zone to target")
registerPathPrefixFlag(cmd.flags, &cmd.root, "root", "", "subtree of the source to consider live")
registerDurationFlag(cmd.flags, &cmd.olderThan, "older-than", "30d", "grace window, only orphans older than this are deleted")
registerDryRunFlag(cmd.flags)
return cmd
}
func (cmd *pruneCommand) Usage() string {
return "Usage:" +
"\n hopper [global flags] prune [flags] ⟨source⟩" +
"\n" +
"\nFlags:" +
"\n" + cmd.flags.FlagUsages()
}
func (cmd *pruneCommand) Run(ctx context.Context, args []string) error {
parseFlags(cmd, cmd.flags, args)
if cmd.olderThan.isZero() {
return usageError{"grace window must not be zero"}
}
positionals := cmd.flags.Args()
if len(positionals) == 0 || positionals[0] == "" {
return usageError{"missing source"}
}
if len(positionals) > 1 {
return usageError{"only a single source is supported"}
}
if err := cmd.main.lookupAccessKey(); err != nil {
return err
}
api := newStorageAPI(cmd.main.endpoint, cmd.main.zone, cmd.main.accessKey)
return cmd.prune(ctx, api, positionals[0], time.Now())
}
func (cmd *pruneCommand) prune(ctx context.Context, api storageAPI, sourceArg string, now time.Time) error {
start := time.Now()
live, err := cmd.liveSet(sourceArg)
if err != nil {
return err
}
if len(live) == 0 {
return errors.New("empty source (refusing to prune). check --root, --prefix, or the archive contents")
}
entries, err := listRemoteEntries(ctx, api, cmd.prefix)
if err != nil {
return err
}
cutoff := cmd.olderThan.cutoff(now)
candidates := make([]*remoteEntry, 0, len(entries)/4)
for i := range entries {
e := &entries[i]
if !e.isDir && !live[e.path] {
if e.lastChanged.IsZero() {
log.Warn("could not determine last modified date for %s (skipping)", e.path)
} else if e.lastChanged.Before(cutoff) {
candidates = append(candidates, e)
}
}
}
// Each delete is independent, so the pool never fails fast. A successful
// delete tombstones its entry.
pool := newWorkerPool(ctx, cmd.concurrency, false)
freed := int64(0)
for _, e := range candidates {
action := fmt.Sprintf("deleting %s (%s)", e.path, humanSize(e.size))
pool.submit(action, func(ctx context.Context) error {
if err := api.Delete(ctx, e.path); err != nil {
return fileError{path: e.path, err: err}
}
e.deleted = true
atomic.AddInt64(&freed, int64(e.size))
return nil
})
}
deleted, errs := pool.wait()
var cmdErr error
if reportFailures(ctx, "failed to delete files", errs) {
cmdErr = exitError{1}
}
cmd.removeEmptyDirs(ctx, api, entries)
failed := len(errs)
logSummary("deleted: %d"+
"\nfailed: %d"+
"\nfreed: %s"+
"\nelapsed: %s",
deleted, failed, humanSize64(freed), time.Since(start).Round(time.Millisecond),
)
return cmdErr
}
// removeEmptyDirs deletes directories left with no surviving children after the
// file deletes and returns the number successfully removed. It runs as a
// best-effort second pass: failures are reported but never propagated to the
// exit code.
func (cmd *pruneCommand) removeEmptyDirs(ctx context.Context, api storageAPI, entries []remoteEntry) {
dirs := emptyDirs(entries)
if len(dirs) == 0 {
return
}
pool := newWorkerPool(ctx, cmd.concurrency, false)
for _, dir := range dirs {
action := fmt.Sprintf("removing empty directory %s", dir.path)
pool.submit(action, func(ctx context.Context) error {
err := api.DeleteDir(ctx, dir.path)
if err != nil {
return fileError{path: dir.path, err: err}
}
return nil
})
}
_, errs := pool.wait()
if len(errs) > 0 {
_ = reportFailures(ctx, "failed to remove empty directories", errs)
}
}
// liveSet walks the source once and returns the set of prefixed storage keys in
// the current release. It iterates entries only and never reads content.
func (cmd *pruneCommand) liveSet(sourceArg string) (map[string]bool, error) {
src, err := source.NewTarGzSource(sourceArg, cmd.root)
if err != nil {
return nil, err
}
defer src.Close()
live := map[string]bool{}
for {
entry, err := src.Next()
if err != nil {
if errors.Is(err, io.EOF) {
return live, nil
}
return nil, err
}
live[prefixedPath(cmd.prefix, entry.Path)] = true
}
}
|
