feat(backup): influx backup creates data backup

Author: Jacob Marble

authorizer/authorize.go

@@ -11,15 +11,23 @@ import (
 // IsAllowed checks to see if an action is authorized by retrieving the authorizer
 // off of context and authorizing the action appropriately.
 func IsAllowed(ctx context.Context, p influxdb.Permission) error {
+	return IsAllowedAll(ctx, []influxdb.Permission{p})
+}
+
+// IsAllowedAll checks to see if an action is authorized by ALL permissions.
+// Also see IsAllowed.
+func IsAllowedAll(ctx context.Context, permissions []influxdb.Permission) error {
 	a, err := influxdbcontext.GetAuthorizer(ctx)
 	if err != nil {
 		return err
 	}
 
-	if !a.Allowed(p) {
-		return &influxdb.Error{
-			Code: influxdb.EUnauthorized,
-			Msg:  fmt.Sprintf("%s is unauthorized", p),
+	for _, p := range permissions {
+		if !a.Allowed(p) {
+			return &influxdb.Error{
+				Code: influxdb.EUnauthorized,
+				Msg:  fmt.Sprintf("%s is unauthorized", p),
+			}
 		}
 	}
 

authorizer/backup.go

@@ -0,0 +1,48 @@
+package authorizer
+
+import (
+	"context"
+	"io"
+
+	"github.com/influxdata/influxdb"
+	"github.com/influxdata/influxdb/kit/tracing"
+)
+
+var _ influxdb.BackupService = (*BackupService)(nil)
+
+// BackupService wraps a influxdb.BackupService and authorizes actions
+// against it appropriately.
+type BackupService struct {
+	s influxdb.BackupService
+}
+
+// NewBackupService constructs an instance of an authorizing backup service.
+func NewBackupService(s influxdb.BackupService) *BackupService {
+	return &BackupService{
+		s: s,
+	}
+}
+
+func (b BackupService) CreateBackup(ctx context.Context) (int, []string, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx)
+	defer span.Finish()
+
+	if err := IsAllowedAll(ctx, influxdb.ReadAllPermissions()); err != nil {
+		return 0, nil, err
+	}
+	return b.s.CreateBackup(ctx)
+}
+
+func (b BackupService) FetchBackupFile(ctx context.Context, backupID int, backupFile string, w io.Writer) error {
+	span, ctx := tracing.StartSpanFromContext(ctx)
+	defer span.Finish()
+
+	if err := IsAllowedAll(ctx, influxdb.ReadAllPermissions()); err != nil {
+		return nil
+	}
+	return b.s.FetchBackupFile(ctx, backupID, backupFile, w)
+}
+
+func (b BackupService) InternalBackupPath(backupID int) string {
+	return b.s.InternalBackupPath(backupID)
+}

authz.go

@@ -335,6 +335,16 @@ func OperPermissions() []Permission {
 	return ps
 }
 
+// ReadAllPermissions represents permission to read all data and metadata.
+// Like OperPermissions, but allows read-only users.
+func ReadAllPermissions() []Permission {
+	ps := make([]Permission, len(AllResourceTypes))
+	for i, t := range AllResourceTypes {
+		ps[i] = Permission{Action: ReadAction, Resource: Resource{Type: t}}
+	}
+	return ps
+}
+
 // OwnerPermissions are the default permissions for those who own a resource.
 func OwnerPermissions(orgID ID) []Permission {
 	ps := []Permission{}

backup.go

@@ -0,0 +1,23 @@
+package influxdb
+
+import (
+	"context"
+	"io"
+)
+
+// BackupService represents the data backup functions of InfluxDB.
+type BackupService interface {
+	// CreateBackup creates a local copy (hard links) of the TSM data for all orgs and buckets.
+	// The return values are used to download each backup file.
+	CreateBackup(context.Context) (backupID int, backupFiles []string, err error)
+	// FetchBackupFile downloads one backup file, data or metadata.
+	FetchBackupFile(ctx context.Context, backupID int, backupFile string, w io.Writer) error
+	// InternalBackupPath is a utility to determine the on-disk location of a backup fileset.
+	InternalBackupPath(backupID int) string
+}
+
+// KVBackupService represents the meta data backup functions of InfluxDB.
+type KVBackupService interface {
+	// Backup creates a live backup copy of the metadata database.
+	Backup(ctx context.Context, w io.Writer) error
+}

bolt/bbolt.go

@@ -14,8 +14,12 @@ import (
 	"go.uber.org/zap"
 )
 
-// OpPrefix is the prefix for bolt ops
-const OpPrefix = "bolt/"
+const (
+	// OpPrefix is the prefix for bolt ops
+	OpPrefix = "bolt/"
+
+	DefaultFilename = "influxd.bolt"
+)
 
 func getOp(op string) string {
 	return OpPrefix + op

bolt/kv.go

@@ -3,6 +3,7 @@ package bolt
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"time"
@@ -120,6 +121,17 @@ func (s *KVStore) Update(ctx context.Context, fn func(tx kv.Tx) error) error {
 	})
 }
 
+// Backup copies all K:Vs to a writer, in BoltDB format.
+func (s *KVStore) Backup(ctx context.Context, w io.Writer) error {
+	span, _ := tracing.StartSpanFromContext(ctx)
+	defer span.Finish()
+
+	return s.db.View(func(tx *bolt.Tx) error {
+		_, err := tx.WriteTo(w)
+		return err
+	})
+}
+
 // Tx is a light wrapper around a boltdb transaction. It implements kv.Tx.
 type Tx struct {
 	tx  *bolt.Tx

cmd/influx/backup.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/influxdata/influxdb"
+	"github.com/influxdata/influxdb/bolt"
+	"github.com/influxdata/influxdb/http"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.uber.org/multierr"
+)
+
+var backupCmd = &cobra.Command{
+	Use:   "backup",
+	Short: "Backup the data in InfluxDB",
+	Long: fmt.Sprintf(
+		`Backs up data and meta data for the running InfluxDB instance.
+Downloaded files are written to the directory indicated by --path.
+The target directory, and any parent directories, are created automatically.
+Data file have extension .tsm; meta data is written to %s in the same directory.`,
+		bolt.DefaultFilename),
+	RunE: backupF,
+}
+
+var backupFlags struct {
+	Path string
+}
+
+func init() {
+	backupCmd.PersistentFlags().StringVarP(&backupFlags.Path, "path", "p", "", "directory path to write backup files to")
+	err := viper.BindEnv("PATH")
+	if err != nil {
+		panic(err)
+	}
+	if h := viper.GetString("PATH"); h != "" {
+		backupFlags.Path = h
+	}
+}
+
+func newBackupService(flags Flags) (influxdb.BackupService, error) {
+	return &http.BackupService{
+		Addr:  flags.host,
+		Token: flags.token,
+	}, nil
+}
+
+func backupF(cmd *cobra.Command, args []string) error {
+	ctx := context.Background()
+
+	if flags.local {
+		return fmt.Errorf("local flag not supported for backup command")
+	}
+
+	if backupFlags.Path == "" {
+		return fmt.Errorf("must specify path")
+	}
+
+	err := os.MkdirAll(backupFlags.Path, 0770)
+	if err != nil && !os.IsExist(err) {
+		return err
+	}
+
+	backupService, err := newBackupService(flags)
+	if err != nil {
+		return err
+	}
+
+	id, backupFilenames, err := backupService.CreateBackup(ctx)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("Backup ID %d contains %d files\n", id, len(backupFilenames))
+
+	for _, backupFilename := range backupFilenames {
+		dest := filepath.Join(backupFlags.Path, backupFilename)
+		w, err := os.OpenFile(dest, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+		if err != nil {
+			return err
+		}
+		err = backupService.FetchBackupFile(ctx, id, backupFilename, w)
+		if err != nil {
+			return multierr.Append(err, w.Close())
+		}
+		if err = w.Close(); err != nil {
+			return err
+		}
+	}
+
+	fmt.Printf("Backup complete")
+
+	return nil
+}

cmd/influx/main.go

@@ -85,6 +85,7 @@ func influxCmd() *cobra.Command {
 	}
 	cmd.AddCommand(
 		authCmd(),
+		backupCmd,
 		bucketCmd,
 		deleteCmd,
 		organizationCmd(),

cmd/influxd/launcher/engine.go

@@ -2,6 +2,7 @@ package launcher
 
 import (
 	"context"
+	"io"
 	"io/ioutil"
 	"os"
 	"sync"
@@ -29,6 +30,7 @@ type Engine interface {
 	storage.PointsWriter
 	storage.BucketDeleter
 	prom.PrometheusCollector
+	influxdb.BackupService
 
 	SeriesCardinality() int64
 
@@ -165,3 +167,15 @@ func (t *TemporaryEngine) Flush(ctx context.Context) {
 		t.log.Fatal("unable to open engine", zap.Error(err))
 	}
 }
+
+func (t *TemporaryEngine) CreateBackup(ctx context.Context) (int, []string, error) {
+	return t.engine.CreateBackup(ctx)
+}
+
+func (t *TemporaryEngine) FetchBackupFile(ctx context.Context, backupID int, backupFile string, w io.Writer) error {
+	return t.engine.FetchBackupFile(ctx, backupID, backupFile, w)
+}
+
+func (t *TemporaryEngine) InternalBackupPath(backupID int) string {
+	return t.engine.InternalBackupPath(backupID)
+}

cmd/influxd/launcher/launcher.go

@@ -146,7 +146,7 @@ func buildLauncherCommand(l *Launcher, cmd *cobra.Command) {
 		{
 			DestP:   &l.boltPath,
 			Flag:    "bolt-path",
-			Default: filepath.Join(dir, "influxd.bolt"),
+			Default: filepath.Join(dir, bolt.DefaultFilename),
 			Desc:    "path to boltdb database",
 		},
 		{
@@ -590,6 +590,7 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 	var (
 		deleteService platform.DeleteService = m.engine
 		pointsWriter  storage.PointsWriter   = m.engine
+		backupService platform.BackupService = m.engine
 	)
 
 	// TODO(cwolff): Figure out a good default per-query memory limit:
@@ -796,6 +797,8 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		NewQueryService:      source.NewQueryService,
 		PointsWriter:         pointsWriter,
 		DeleteService:        deleteService,
+		BackupService:        backupService,
+		KVBackupService:      m.kvService,
 		AuthorizationService: authSvc,
 		// Wrap the BucketService in a storage backed one that will ensure deleted buckets are removed from the storage engine.
 		BucketService:                   storage.NewBucketService(bucketSvc, m.engine),

cmd/influxd/launcher/launcher_helpers.go

@@ -17,6 +17,7 @@ import (
 	"github.com/influxdata/flux"
 	"github.com/influxdata/flux/lang"
 	platform "github.com/influxdata/influxdb"
+	"github.com/influxdata/influxdb/bolt"
 	influxdbcontext "github.com/influxdata/influxdb/context"
 	"github.com/influxdata/influxdb/http"
 	"github.com/influxdata/influxdb/kv"
@@ -79,7 +80,7 @@ func RunTestLauncherOrFail(tb testing.TB, ctx context.Context, args ...string) *
 
 // Run executes the program with additional arguments to set paths and ports.
 func (tl *TestLauncher) Run(ctx context.Context, args ...string) error {
-	args = append(args, "--bolt-path", filepath.Join(tl.Path, "influxd.bolt"))
+	args = append(args, "--bolt-path", filepath.Join(tl.Path, bolt.DefaultFilename))
 	args = append(args, "--engine-path", filepath.Join(tl.Path, "engine"))
 	args = append(args, "--http-bind-address", "127.0.0.1:0")
 	args = append(args, "--log-level", "debug")

http/api_handler.go

@@ -36,6 +36,8 @@ type APIBackend struct {
 
 	PointsWriter                    storage.PointsWriter
 	DeleteService                   influxdb.DeleteService
+	BackupService                   influxdb.BackupService
+	KVBackupService                 influxdb.KVBackupService
 	AuthorizationService            influxdb.AuthorizationService
 	BucketService                   influxdb.BucketService
 	SessionService                  influxdb.SessionService
@@ -197,6 +199,10 @@ func NewAPIHandler(b *APIBackend, opts ...APIHandlerOptFn) *APIHandler {
 	variableBackend.VariableService = authorizer.NewVariableService(b.VariableService)
 	h.Mount(prefixVariables, NewVariableHandler(b.Logger, variableBackend))
 
+	backupBackend := NewBackupBackend(b)
+	backupBackend.BackupService = authorizer.NewBackupService(backupBackend.BackupService)
+	h.Mount(prefixBackup, NewBackupHandler(backupBackend))
+
 	writeBackend := NewWriteBackend(b.Logger.With(zap.String("handler", "write")), b)
 	h.Mount(prefixWrite, NewWriteHandler(b.Logger, writeBackend))
 
@@ -210,6 +216,7 @@ var apiLinks = map[string]interface{}{
 	// when adding new links, please take care to keep this list alphabetical
 	// as this makes it easier to verify values against the swagger document.
 	"authorizations": "/api/v2/authorizations",
+	"backup":         "/api/v2/backup",
 	"buckets":        "/api/v2/buckets",
 	"dashboards":     "/api/v2/dashboards",
 	"external": map[string]string{