feat(backup): influx backup creates data backup

Author: Jacob Marble

authorizer/authorize.go

@@ -25,3 +25,23 @@ func IsAllowed(ctx context.Context, p influxdb.Permission) error {
 
 	return nil
 }
+
+// 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
+	}
+
+	for _, p := range permissions {
+		if !a.Allowed(p) {
+			return &influxdb.Error{
+				Code: influxdb.EUnauthorized,
+				Msg:  fmt.Sprintf("%s is unauthorized", p),
+			}
+		}
+	}
+
+	return nil
+}

authorizer/backup.go

@@ -0,0 +1,49 @@
+package authorizer
+
+import (
+	"context"
+	"io"
+
+	"github.com/influxdata/influxdb/kit/tracing"
+
+	"github.com/influxdata/influxdb"
+)
+
+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,16 @@
+package influxdb
+
+import (
+	"context"
+	"io"
+)
+
+type BackupService interface {
+	CreateBackup(context.Context) (int, []string, error)
+	FetchBackupFile(ctx context.Context, backupID int, backupFile string, w io.Writer) error
+	InternalBackupPath(backupID int) string
+}
+
+type KVBackupService interface {
+	Backup(ctx context.Context, w io.Writer) error
+}

bolt/kv.go

@@ -3,14 +3,16 @@ package bolt
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"time"
 
 	bolt "github.com/coreos/bbolt"
+	"go.uber.org/zap"
+
 	"github.com/influxdata/influxdb/kit/tracing"
 	"github.com/influxdata/influxdb/kv"
-	"go.uber.org/zap"
 )
 
 // KVStore is a kv.Store backed by boltdb.
@@ -125,6 +127,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, ctx := 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,90 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/influxdata/influxdb"
+	"github.com/influxdata/influxdb/http"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"go.uber.org/multierr"
+	"os"
+	"path/filepath"
+)
+
+var backupCmd = &cobra.Command{
+	Use:   "backup",
+	Short: "Backup the data in InfluxDB",
+	Long:  `Backs up data and meta data for the InfluxDB instance. OSS only.`,
+	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.Mkdir(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

@@ -33,6 +33,7 @@ const maxTCPConnections = 128
 func init() {
 	influxCmd.AddCommand(
 		authorizationCmd,
+		backupCmd,
 		bucketCmd,
 		deleteCmd,
 		organizationCmd,

cmd/influxd/launcher/launcher.go

@@ -578,6 +578,7 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 
 	var deleteService platform.DeleteService
 	var pointsWriter storage.PointsWriter
+	var backupService platform.BackupService
 	{
 		m.engine = storage.NewEngine(m.enginePath, m.StorageConfig, storage.WithRetentionEnforcer(bucketSvc))
 		m.engine.WithLogger(m.logger)
@@ -591,6 +592,7 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 
 		pointsWriter = m.engine
 		deleteService = m.engine
+		backupService = m.engine
 
 		// TODO(cwolff): Figure out a good default per-query memory limit:
 		//   https://github.com/influxdata/influxdb/issues/13642
@@ -801,6 +803,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),

http/api_handler.go

@@ -21,6 +21,7 @@ type APIHandler struct {
 	influxdb.HTTPErrorHandler
 	AssetHandler                *AssetHandler
 	AuthorizationHandler        *AuthorizationHandler
+	BackupHandler               *BackupHandler
 	BucketHandler               *BucketHandler
 	CheckHandler                *CheckHandler
 	ChronografHandler           *ChronografHandler
@@ -62,6 +63,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
@@ -213,6 +216,10 @@ func NewAPIHandler(b *APIBackend, opts ...APIHandlerOptFn) *APIHandler {
 	deleteBackend := NewDeleteBackend(b)
 	h.DeleteHandler = NewDeleteHandler(deleteBackend)
 
+	backupBackend := NewBackupBackend(b)
+	backupBackend.BackupService = authorizer.NewBackupService(backupBackend.BackupService)
+	h.BackupHandler = NewBackupHandler(backupBackend)
+
 	fluxBackend := NewFluxBackend(b)
 	h.QueryHandler = NewFluxHandler(fluxBackend)
 
@@ -227,40 +234,41 @@ 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",
+	"checks":         "/api/v2/checks",
 	"dashboards":     "/api/v2/dashboards",
+	"delete":         "/api/v2/delete",
 	"external": map[string]string{
 		"statusFeed": "https://www.influxdata.com/feed/json",
 	},
 	"labels":                "/api/v2/labels",
-	"variables":             "/api/v2/variables",
 	"me":                    "/api/v2/me",
-	"notificationRules":     "/api/v2/notificationRules",
 	"notificationEndpoints": "/api/v2/notificationEndpoints",
+	"notificationRules":     "/api/v2/notificationRules",
 	"orgs":                  "/api/v2/orgs",
 	"query": map[string]string{
 		"self":        "/api/v2/query",
 		"ast":         "/api/v2/query/ast",
 		"analyze":     "/api/v2/query/analyze",
 		"suggestions": "/api/v2/query/suggestions",
 	},
+	"scrapers": "/api/v2/scrapers",
 	"setup":    "/api/v2/setup",
 	"signin":   "/api/v2/signin",
 	"signout":  "/api/v2/signout",
 	"sources":  "/api/v2/sources",
-	"scrapers": "/api/v2/scrapers",
 	"swagger":  "/api/v2/swagger.json",
 	"system": map[string]string{
 		"metrics": "/metrics",
 		"debug":   "/debug/pprof",
 		"health":  "/health",
 	},
 	"tasks":     "/api/v2/tasks",
-	"checks":    "/api/v2/checks",
 	"telegrafs": "/api/v2/telegrafs",
 	"users":     "/api/v2/users",
+	"variables": "/api/v2/variables",
 	"write":     "/api/v2/write",
-	"delete":    "/api/v2/delete",
 }
 
 func (h *APIHandler) serveLinks(w http.ResponseWriter, r *http.Request) {
@@ -303,6 +311,11 @@ func (h *APIHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if strings.HasPrefix(r.URL.Path, "/api/v2/backup") {
+		h.BackupHandler.ServeHTTP(w, r)
+		return
+	}
+
 	if strings.HasPrefix(r.URL.Path, "/api/v2/query") {
 		h.QueryHandler.ServeHTTP(w, r)
 		return