diff --git a/cmd/config.go b/cmd/config.go index f3b538b..04026a2 100644 --- a/cmd/config.go +++ b/cmd/config.go @@ -13,6 +13,8 @@ import ( yaml "gopkg.in/yaml.v2" "io" "io/ioutil" + "regexp" + "strconv" "strings" "time" ) @@ -58,12 +60,20 @@ type ClientMapping struct { Mapping zfs.DatasetMapping } +type Prune struct { + Name string + DatasetFilter zfs.DatasetMapping + SnapshotFilter zfs.FilesystemVersionFilter + RetentionPolicy *RetentionGrid // TODO abstract interface to support future policies? +} + type Config struct { Pools []Pool Pushs []Push Pulls []Pull Sinks []ClientMapping PullACLs []ClientMapping + Prunes []Prune } func ParseConfig(path string) (config Config, err error) { @@ -109,6 +119,9 @@ func parseMain(root map[string]interface{}) (c Config, err error) { if c.PullACLs, err = parseClientMappings(root["pull_acls"]); err != nil { return } + if c.Prunes, err = parsePrunes(root["prune"]); err != nil { + return + } return } @@ -432,3 +445,182 @@ func (t LocalTransport) Connect(rpcLog Logger) (r rpc.RPCRequester, err error) { func (t *LocalTransport) SetHandler(handler rpc.RPCHandler) { t.Handler = handler } + +func parsePrunes(m interface{}) (rets []Prune, err error) { + + asList := make([]map[string]interface{}, 0) + if err = mapstructure.Decode(m, &asList); err != nil { + return + } + + rets = make([]Prune, len(asList)) + + for i, e := range asList { + if rets[i], err = parsePrune(e); err != nil { + err = fmt.Errorf("cannot parse prune job #%d: %s", i+1, err) + return + } + } + + return +} + +func parsePrune(e map[string]interface{}) (prune Prune, err error) { + // Only support grid policy for now + policyName, ok := e["policy"] + if !ok || policyName != "grid" { + err = fmt.Errorf("prune job with unimplemented policy '%s'", policyName) + return + } + + var i struct { + Name string + Grid string + DatasetFilter map[string]string `mapstructure:"dataset_filter"` + SnapshotFilter map[string]string `mapstructure:"snapshot_filter"` + } + + if err = mapstructure.Decode(e, &i); err != nil { + return + } + + prune.Name = i.Name + + // Parse grid policy + intervals, err := parseRetentionGridIntervalsString(i.Grid) + if err != nil { + err = fmt.Errorf("cannot parse retention grid: %s", err) + return + } + // Assert intervals are of increasing length (not necessarily required, but indicates config mistake) + lastDuration := time.Duration(0) + for i := range intervals { + if intervals[i].Length < lastDuration { + err = fmt.Errorf("retention grid interval length must be monotonically increasing:"+ + "interval %d is shorter than %d", i+1, i) + return + } else { + lastDuration = intervals[i].Length + } + } + prune.RetentionPolicy = NewRetentionGrid(intervals) + + // Parse filters + if prune.DatasetFilter, err = parseComboMapping(i.DatasetFilter); err != nil { + err = fmt.Errorf("cannot parse dataset filter: %s", err) + return + } + if prune.SnapshotFilter, err = parseSnapshotFilter(i.SnapshotFilter); err != nil { + err = fmt.Errorf("cannot parse snapshot filter: %s", err) + return + } + + return +} + +var retentionStringIntervalRegex *regexp.Regexp = regexp.MustCompile(`^\s*(\d+)\s*x\s*(\d+)\s*(s|min|h|d|w|mon)\s*\s*(\((.*)\))?\s*$`) + +func parseRetentionGridIntervalString(e string) (intervals []RetentionInterval, err error) { + + comps := retentionStringIntervalRegex.FindStringSubmatch(e) + if comps == nil { + err = fmt.Errorf("retention string does not match expected format") + return + } + + times, err := strconv.Atoi(comps[1]) + if err != nil { + return nil, err + } else if times <= 0 { + return nil, fmt.Errorf("contains factor <= 0") + } + + durationFactor, err := strconv.ParseInt(comps[2], 10, 64) + if err != nil { + return nil, err + } + + var durationUnit time.Duration + switch comps[3] { + case "s": + durationUnit = time.Second + case "min": + durationUnit = time.Minute + case "h": + durationUnit = time.Hour + case "d": + durationUnit = 24 * time.Hour + case "w": + durationUnit = 24 * 7 * time.Hour + case "mon": + durationUnit = 32 * 24 * 7 * time.Hour + default: + err = fmt.Errorf("contains unknown time unit '%s'", comps[3]) + return nil, err + } + + keepCount := 1 + if comps[4] != "" { + // Decompose key=value, comma separated + // For now, only keep_count is supported + re := regexp.MustCompile(`^\s*keep=(.+)\s*$`) + res := re.FindStringSubmatch(comps[5]) + if res == nil || len(res) != 2 { + err = fmt.Errorf("interval parameter contains unknown parameters") + } + if res[1] == "all" { + keepCount = RetentionGridKeepCountAll + } else { + keepCount, err = strconv.Atoi(res[1]) + if err != nil { + err = fmt.Errorf("cannot parse keep_count value") + return + } + } + } + + intervals = make([]RetentionInterval, times) + for i := range intervals { + intervals[i] = RetentionInterval{ + Length: time.Duration(durationFactor) * durationUnit, // TODO is this conversion fututre-proof? + KeepCount: keepCount, + } + } + + return + +} + +func parseRetentionGridIntervalsString(s string) (intervals []RetentionInterval, err error) { + + ges := strings.Split(s, "|") + intervals = make([]RetentionInterval, 0, 7*len(ges)) + + for intervalIdx, e := range ges { + parsed, err := parseRetentionGridIntervalString(e) + if err != nil { + return nil, fmt.Errorf("cannot parse interval %d of %d: %s: %s", intervalIdx+1, len(ges), err, strings.TrimSpace(e)) + } + intervals = append(intervals, parsed...) + } + + return +} + +type prefixSnapshotFilter struct { + prefix string +} + +func (f prefixSnapshotFilter) Filter(fsv zfs.FilesystemVersion) (accept bool, err error) { + return fsv.Type == zfs.Snapshot && strings.HasPrefix(fsv.Name, f.prefix), nil +} + +func parseSnapshotFilter(fm map[string]string) (snapFilter zfs.FilesystemVersionFilter, err error) { + prefix, ok := fm["prefix"] + if !ok { + err = fmt.Errorf("unsupported snapshot filter") + return + } + snapFilter = prefixSnapshotFilter{prefix} + return +} diff --git a/cmd/config_test.go b/cmd/config_test.go index 8d0a303..d0e57a2 100644 --- a/cmd/config_test.go +++ b/cmd/config_test.go @@ -2,10 +2,46 @@ package main import ( "github.com/stretchr/testify/assert" + "github.com/zrepl/zrepl/util" "testing" + "time" ) func TestSampleConfigFileIsParsedWithoutErrors(t *testing.T) { _, err := ParseConfig("./sampleconf/zrepl.yml") assert.Nil(t, err) } + +func TestParseRetentionGridStringParsing(t *testing.T) { + + intervals, err := parseRetentionGridIntervalsString("2x10min(keep=2) | 1x1h | 3x1w") + + assert.Nil(t, err) + assert.Len(t, intervals, 6) + proto := util.RetentionInterval{ + KeepCount: 2, + Length: 10 * time.Minute, + } + assert.EqualValues(t, proto, intervals[0]) + assert.EqualValues(t, proto, intervals[1]) + + proto.KeepCount = 1 + proto.Length = 1 * time.Hour + assert.EqualValues(t, proto, intervals[2]) + + proto.Length = 7 * 24 * time.Hour + assert.EqualValues(t, proto, intervals[3]) + assert.EqualValues(t, proto, intervals[4]) + assert.EqualValues(t, proto, intervals[5]) + + intervals, err = parseRetentionGridIntervalsString("|") + assert.Error(t, err) + intervals, err = parseRetentionGridIntervalsString("2x10min") + assert.NoError(t, err) + + intervals, err = parseRetentionGridIntervalsString("1x10min(keep=all)") + assert.NoError(t, err) + assert.Len(t, intervals, 1) + assert.EqualValues(t, util.RetentionGridKeepCountAll, intervals[0].KeepCount) + +} diff --git a/cmd/main.go b/cmd/main.go index e8560e2..07585b8 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -111,6 +111,14 @@ func main() { }, }, }, + { + Name: "prune", + Action: cmdPrune, + Flags: []cli.Flag{ + cli.StringFlag{Name: "job"}, + cli.BoolFlag{Name: "n", Usage: "simulation (dry run)"}, + }, + }, } app.Run(os.Args) @@ -278,3 +286,33 @@ func jobPush(push Push, c *cli.Context, log jobrun.Logger) (err error) { return } + +func cmdPrune(c *cli.Context) error { + + log := defaultLog + + jobFailed := false + + log.Printf("retending...") + + for _, prune := range conf.Prunes { + + if !c.IsSet("job") || (c.IsSet("job") && c.String("job") == prune.Name) { + log.Printf("Beginning prune job:\n%s", prune) + ctx := PruneContext{prune, time.Now(), c.IsSet("n")} + err := doPrune(ctx, log) + if err != nil { + jobFailed = true + log.Printf("Prune job failed with error: %s", err) + } + log.Printf("\n") + } + + } + + if jobFailed { + return cli.NewExitError("At least one job failed with an error. Check log for details.", 1) + } + return nil + +} diff --git a/cmd/prune.go b/cmd/prune.go new file mode 100644 index 0000000..6d27a42 --- /dev/null +++ b/cmd/prune.go @@ -0,0 +1,97 @@ +package main + +import ( + "fmt" + "github.com/zrepl/zrepl/util" + "github.com/zrepl/zrepl/zfs" + "sort" + "time" +) + +type PruneContext struct { + Prune Prune + Now time.Time + DryRun bool +} + +type retentionGridAdaptor struct { + zfs.FilesystemVersion +} + +func (a retentionGridAdaptor) Date() time.Time { + return a.Creation +} + +func (a retentionGridAdaptor) LessThan(b util.RetentionGridEntry) bool { + return a.CreateTXG < b.(retentionGridAdaptor).CreateTXG +} + +func doPrune(ctx PruneContext, log Logger) error { + + if ctx.DryRun { + log.Printf("doing dry run") + } + + prune := ctx.Prune + + // ZFSListSnapsFiltered --> todo can extend fsfilter or need new? Have already something per fs + // Dedicated snapshot object? Adaptor object to FilesystemVersion? + + filesystems, err := zfs.ZFSListMapping(prune.DatasetFilter) + if err != nil { + return fmt.Errorf("error applying filesystem filter: %s", err) + } + + for _, fs := range filesystems { + + fsversions, err := zfs.ZFSListFilesystemVersions(fs, prune.SnapshotFilter) + if err != nil { + return fmt.Errorf("error listing filesytem versions of %s: %s", fs, err) + } + + if len(fsversions) == 0 { + continue + } + + adaptors := make([]util.RetentionGridEntry, len(fsversions)) + for fsv := range fsversions { + adaptors[fsv] = retentionGridAdaptor{fsversions[fsv]} + } + + sort.SliceStable(adaptors, func(i, j int) bool { + return adaptors[i].LessThan(adaptors[j]) + }) + ctx.Now = adaptors[len(adaptors)-1].Date() + + describe := func(a retentionGridAdaptor) string { + timeSince := a.Creation.Sub(ctx.Now) + const day time.Duration = 24 * time.Hour + days := timeSince / day + remainder := timeSince % day + return fmt.Sprintf("%s@%dd%s from now", a.ToAbsPath(fs), days, remainder) + } + + keep, remove := prune.RetentionPolicy.FitEntries(ctx.Now, adaptors) + for _, a := range remove { + r := a.(retentionGridAdaptor) + log.Printf("remove %s", describe(r)) + // do echo what we'll do and exec zfs destroy if not dry run + // special handling for EBUSY (zfs hold) + // error handling for clones? just echo to cli, skip over, and exit with non-zero status code (we're idempotent) + if !ctx.DryRun { + err := zfs.ZFSDestroy(r.ToAbsPath(fs)) + if err != nil { + // handle + log.Printf("error: %s", err) + } + } + } + for _, a := range keep { + r := a.(retentionGridAdaptor) + log.Printf("would keep %s", describe(r)) + } + } + + return nil + +} diff --git a/cmd/sampleconf/zrepl.yml b/cmd/sampleconf/zrepl.yml index b5a8980..81fd41f 100644 --- a/cmd/sampleconf/zrepl.yml +++ b/cmd/sampleconf/zrepl.yml @@ -84,3 +84,16 @@ pull_acls: mapping: { "tank/usr/home":"notnull" } + + +prune: + + - name: clean_backups + policy: grid + grid: 6x10min | 24x1h | 7x1d | 32 x 1d | 4 x 3mon + dataset_filter: { + "tank/backups/*": ok + } + snapshot_filter: { + prefix: zrepl_ + } diff --git a/zfs/versions.go b/zfs/versions.go index 331ba60..91e45e9 100644 --- a/zfs/versions.go +++ b/zfs/versions.go @@ -6,6 +6,7 @@ import ( "fmt" "strconv" "strings" + "time" ) type VersionType string @@ -39,6 +40,9 @@ type FilesystemVersion struct { // The TXG in which the snapshot was created. For bookmarks, // this is the GUID of the snapshot it was initially tied to. CreateTXG uint64 + + // The time the dataset was created + Creation time.Time } func (v FilesystemVersion) ToAbsPath(p DatasetPath) string { @@ -56,7 +60,7 @@ type FilesystemVersionFilter interface { func ZFSListFilesystemVersions(fs DatasetPath, filter FilesystemVersionFilter) (res []FilesystemVersion, err error) { var fieldLines [][]string fieldLines, err = ZFSList( - []string{"name", "guid", "createtxg"}, + []string{"name", "guid", "createtxg", "creation"}, "-r", "-d", "1", "-t", "bookmark,snapshot", "-s", "createtxg", fs.ToString()) @@ -97,6 +101,14 @@ func ZFSListFilesystemVersions(fs DatasetPath, filter FilesystemVersionFilter) ( return } + creationUnix, err := strconv.ParseInt(line[3], 10, 64) + if err != nil { + err = fmt.Errorf("cannot parse creation date '%s': %s", line[3], err) + return nil, err + } else { + v.Creation = time.Unix(creationUnix, 0) + } + accept := true if filter != nil { accept, err = filter.Filter(v)