Implement a 'restore' command

README.md

@@ -74,7 +74,15 @@ There is often a trade-off to be made when backing up multi-tenant
 services: do we invoke a backup handler once per tenant, or do we dump
 everything once and just say it's made of multiple atoms? You can pick
 the best approach on a case-by-case basis, by grouping atoms into
-*datasets*. A few examples might help:
+*datasets*. We'll look at examples later to clarify what this means.
+
+## Repository
+
+The first thing your backup needs is a destination *repository*, that
+is, a way to archive data long-term. The current implementation
+uses [restic](https://restic.net), an encrypted-deduplicating backup
+tool that supports a
+[large number of remote storage options](https://restic.readthedocs.io/en/stable/030_preparing_a_new_repo.html).
 
 ## The *file* handler
 
@@ -245,3 +253,15 @@ and this dataset source:
 ```
 
 
+# TODO
+
+Things still to do:
+
+* The agent can currently do both backups and restores, but there is
+  no way to trigger a restore. Some sort of authenticated API is
+  needed for this.
+
+Things not to do:
+
+* Global (cluster-wide) scheduling - that's the job of a global cron
+  scheduler, that could then easily trigger backups.

cmd/tabacco/restore.go

+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+
+	"github.com/google/subcommands"
+
+	"git.autistici.org/ale/tabacco"
+	"git.autistici.org/ale/tabacco/jobs"
+	mdbc "git.autistici.org/ale/tabacco/metadb/client"
+)
+
+type restoreCommand struct {
+	configPath string
+	httpAddr   string
+	targetDir  string
+}
+
+func (c *restoreCommand) Name() string     { return "restore" }
+func (c *restoreCommand) Synopsis() string { return "restore one or more datasets" }
+func (c *restoreCommand) Usage() string {
+	return `restore [<flags>] <dataset_filter(s)>...
+        Restore datasets to the local host, based on the provided
+        dataset matching criteria.
+
+`
+}
+
+func (c *restoreCommand) SetFlags(f *flag.FlagSet) {
+	f.StringVar(&c.configPath, "config", "/etc/tabacco/agent.yml", "configuration `file`")
+	f.StringVar(&c.httpAddr, "http-addr", ":5330", "listen `address` for the HTTP server exporting metrics and debugging")
+	f.StringVar(&c.targetDir, "target", "", "target `path` for restore")
+}
+
+func (c *restoreCommand) Execute(ctx context.Context, f *flag.FlagSet, args ...interface{}) subcommands.ExitStatus {
+	if f.NArg() == 0 {
+		log.Printf("error: not enough arguments")
+		return subcommands.ExitUsageError
+	}
+
+	configMgr, mgr, err := c.initManager(ctx)
+	if err != nil {
+		log.Printf("error: %v", err)
+		return subcommands.ExitFailure
+	}
+	defer configMgr.Close()
+	defer mgr.Close()
+
+	// Build the tree of restore jobs and execute it.
+	j, err := c.buildRestoreJob(ctx, mgr, f.Args())
+	if err != nil {
+		log.Printf("error: %v", err)
+		return subcommands.ExitFailure
+	}
+	if err := j.RunContext(ctx); err != nil {
+		log.Printf("error: %v", err)
+		return subcommands.ExitFailure
+	}
+
+	return subcommands.ExitSuccess
+}
+
+func (c *restoreCommand) buildRestoreJob(ctx context.Context, mgr tabacco.Manager, args []string) (jobs.Job, error) {
+	var restoreJobs []jobs.Job
+
+	for _, arg := range args {
+		j, err := mgr.RestoreJob(ctx, c.newFindRequest(arg), c.targetDir)
+		if err != nil {
+			return nil, err
+		}
+		log.Printf("preparing restore of %s", arg)
+		restoreJobs = append(restoreJobs, j)
+	}
+
+	return jobs.AsyncGroup(restoreJobs), nil
+}
+
+func (c *restoreCommand) newFindRequest(s string) tabacco.FindRequest {
+	return tabacco.FindRequest{
+		Pattern: s,
+	}
+}
+
+func (c *restoreCommand) initManager(ctx context.Context) (*tabacco.ConfigManager, tabacco.Manager, error) {
+	// Build a ConfigManager.
+	config, err := tabacco.ReadConfig(c.configPath)
+	if err != nil {
+		return nil, nil, err
+	}
+	configMgr, err := tabacco.NewConfigManager(config)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Connect to the metadata store.
+	store, err := mdbc.New(config.MetadataStoreBackend)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Directly create a Manager and tell it to invoke a restore.
+	mgr, err := tabacco.NewManager(ctx, configMgr, store)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return configMgr, mgr, nil
+}
+
+func init() {
+	subcommands.Register(&withSignalHandlers{&restoreCommand{}}, "")
+}

metadb/server/service.go

@@ -63,16 +63,19 @@ func (a *dbAtom) getAtom() tabacco.Atom {
 	}
 }
 
-func normalizeAtoms(dbAtoms []dbAtom) [][]tabacco.Version {
+func normalizeAtoms(dbAtoms []dbAtom, numVersions int) [][]tabacco.Version {
 	// Accumulate versions keyed by backup ID first, dataset name
-	// next.
+	// next. Preserve the ordering of backups in dbAtoms, which we
+	// are going to use later to apply a per-dataset limit.
 	backupMap := make(map[string]*tabacco.Backup)
 	dsMap := make(map[string]map[string]*tabacco.Dataset)
+	var backupsInOrder []string
 
 	for _, atom := range dbAtoms {
 		// Create the Backup object if it does not exist.
 		if _, ok := backupMap[atom.BackupID]; !ok {
 			backupMap[atom.BackupID] = atom.getBackup()
+			backupsInOrder = append(backupsInOrder, atom.BackupID)
 		}
 
 		// Create the Dataset object for this Backup in the
@@ -95,13 +98,21 @@ func normalizeAtoms(dbAtoms []dbAtom) [][]tabacco.Version {
 
 	// Now dump the maps to a Version array.
 	var out [][]tabacco.Version
-	for backupID, tmp := range dsMap {
+	dsCount := make(map[string]int)
+	for _, backupID := range backupsInOrder {
+		tmp := dsMap[backupID]
 		backup := backupMap[backupID]
 		var tmpv []tabacco.Version
 		for _, ds := range tmp {
+			if dsCount[ds.Name] >= numVersions {
+				continue
+			}
 			tmpv = append(tmpv, tabacco.Version{Backup: *backup, Dataset: *ds})
+			dsCount[ds.Name]++
+		}
+		if len(tmpv) > 0 {
+			out = append(out, tmpv)
 		}
-		out = append(out, tmpv)
 	}
 	return out
 }
@@ -221,5 +232,5 @@ func (s *Service) FindAtoms(ctx context.Context, req tabacco.FindRequest) ([][]t
 		return nil, err
 	}
 
-	return normalizeAtoms(atoms), nil
+	return normalizeAtoms(atoms, req.NumVersions), nil
 }

testdata/metadb.yml

+db_driver: sqlite3
+db_uri: meta.db
+http_server: {}
+