Merge pull request #88 from percona/PMB-31

PMB-31 Support start backup from the cli

Author: percona-csalguero

.env

@@ -1,7 +1,7 @@
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
 GOCACHE=off
-GOLANG_DOCKERHUB_TAG=1.11-stretch
+GOLANG_DOCKERHUB_TAG=1.10-stretch
 TEST_MONGODB_ADMIN_USERNAME=admin
 TEST_MONGODB_ADMIN_PASSWORD=admin123456
 TEST_MONGODB_USERNAME=test

cli/mongodb-backup-admin/main.go

@@ -4,27 +4,32 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"log"
 	"os"
-	"strings"
+	"sort"
 	"time"
 
 	"github.com/alecthomas/kingpin"
 	pbapi "github.com/percona/mongodb-backup/proto/api"
 	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/testdata"
 )
 
 type cliOptions struct {
 	app        *kingpin.Application
-	clientID   *string
 	tls        *bool
 	caFile     *string
 	serverAddr *string
 
 	listClients *kingpin.CmdClause
+
+	startBackup          *kingpin.CmdClause
+	backupType           *string
+	destinationType      *string
+	compressionAlgorithm *string
+	encryptionAlgorithm  *string
 }
 
 func main() {
@@ -56,18 +61,30 @@ func main() {
 	}
 	defer conn.Close()
 
+	apiClient := pbapi.NewApiClient(conn)
+	if err != nil {
+		log.Fatalf("Cannot connect to the API: %s", err)
+	}
+
 	switch cmd {
 	case "list-agents":
-		clients, err := getConnectedAgents(conn)
+		clients, err := connectedAgents(conn)
 		if err != nil {
-			log.Fatal(err)
+			log.Debugf("Cannot connect to the gRPC server: %s", err)
+			log.Fatal("Cannot connect to the gRPC server")
 		}
 		printConnectedAgents(clients)
+	case "start-backup":
+		err := startBackup(apiClient, opts)
+		if err != nil {
+			log.Fatal(err)
+			log.Fatalf("Cannot connect to the gRPC server: %s", err)
+		}
 	}
 
 }
 
-func getConnectedAgents(conn *grpc.ClientConn) ([]*pbapi.Client, error) {
+func connectedAgents(conn *grpc.ClientConn) ([]*pbapi.Client, error) {
 	apiClient := pbapi.NewApiClient(conn)
 	stream, err := apiClient.GetClients(context.Background(), &pbapi.Empty{})
 	if err != nil {
@@ -84,6 +101,7 @@ func getConnectedAgents(conn *grpc.ClientConn) ([]*pbapi.Client, error) {
 		}
 		clients = append(clients, msg)
 	}
+	sort.Slice(clients, func(i, j int) bool { return clients[i].NodeName < clients[j].NodeName })
 	return clients, nil
 }
 
@@ -92,21 +110,69 @@ func printConnectedAgents(clients []*pbapi.Client) {
 		fmt.Println("There are no agents connected to the backup master")
 		return
 	}
-	fmt.Println(strings.Repeat("-", 100))
+	fmt.Println("      Node Name       -    Node Type         -     Replicaset     -  Backup Running  -            Last Seen")
 	for _, client := range clients {
-		fmt.Printf("%s  -  %s  -  %v\n", client.ClientID, client.Status, time.Unix(client.LastSeen, 0))
+		runningBackup := "false"
+		if client.Status.GetRunningDBBackup() {
+			runningBackup = "true"
+		}
+		fmt.Printf("%20s  - %20s - %18s -       %5s      -  %v\n", client.ID, client.NodeType, client.ReplicasetName, runningBackup, time.Unix(client.LastSeen, 0))
 	}
 }
 
-func processCliArgs(args []string) (string, *cliOptions, error) {
+func startBackup(apiClient pbapi.ApiClient, opts *cliOptions) error {
+	msg := &pbapi.RunBackupParams{
+		CompressionType: pbapi.CompressionType_NO_COMPRESSION,
+		Cypher:          pbapi.Cypher_NO_CYPHER,
+	}
 
+	switch *opts.backupType {
+	case "logical":
+		msg.BackupType = pbapi.BackupType_LOGICAL
+	case "hot":
+		msg.BackupType = pbapi.BackupType_LOGICAL
+	}
+
+	switch *opts.destinationType {
+	case "logical":
+		msg.DestinationType = pbapi.DestinationType_FILE
+	case "aws":
+		msg.DestinationType = pbapi.DestinationType_AWS
+	}
+
+	switch *opts.compressionAlgorithm {
+	case "gzip":
+		msg.CompressionType = pbapi.CompressionType_GZIP
+	}
+
+	switch *opts.encryptionAlgorithm {
+	}
+
+	_, err := apiClient.RunBackup(context.Background(), msg)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func processCliArgs(args []string) (string, *cliOptions, error) {
 	app := kingpin.New("mongodb-backup-admin", "MongoDB backup admin")
+	listClientsCmd := app.Command("list-agents", "List all agents connected to the server")
+	startBackupCmd := app.Command("start-backup", "Start a backup")
+
 	opts := &cliOptions{
-		clientID:    kingpin.Flag("client-id", "Client ID").Required().String(),
-		tls:         kingpin.Flag("tls", "Connection uses TLS if true, else plain TCP").Default("false").Bool(),
-		caFile:      kingpin.Flag("ca-file", "The file containning the CA root cert file").String(),
-		serverAddr:  kingpin.Flag("server-addr", "The server address in the format of host:port").Default("127.0.0.1:10001").String(),
-		listClients: kingpin.Command("list-agents", "List all agents connected to the server"),
+		tls:        app.Flag("tls", "Connection uses TLS if true, else plain TCP").Default("false").Bool(),
+		caFile:     app.Flag("ca-file", "The file containning the CA root cert file").String(),
+		serverAddr: app.Flag("server-addr", "The server address in the format of host:port").Default("127.0.0.1:10001").String(),
+
+		listClients: listClientsCmd,
+
+		startBackup:          startBackupCmd,
+		backupType:           startBackupCmd.Flag("backup-type", "Backup type").Enum("logical", "hot"),
+		destinationType:      startBackupCmd.Flag("destination-type", "Backup destination type").Enum("file", "aws"),
+		compressionAlgorithm: startBackupCmd.Flag("compression-algorithm", "Compression algorithm used for the backup").String(),
+		encryptionAlgorithm:  startBackupCmd.Flag("encryption-algorithm", "Encryption algorithm used for the backup").String(),
 	}
 
 	cmd, err := app.Parse(args)

cli/mongodb-backup-agent/main.go

@@ -3,25 +3,34 @@ package main
 import (
 	"context"
 	"fmt"
+	"io/ioutil"
 	"log"
 	"math/rand"
 	"os"
 	"os/signal"
+	"strconv"
 	"strings"
+	"syscall"
 
 	"github.com/alecthomas/kingpin"
 	"github.com/globalsign/mgo"
 	"github.com/percona/mongodb-backup/grpc/client"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 )
 
 type cliOptios struct {
 	app           *kingpin.Application
 	dsn           *string
+	pidFile       *string
 	serverAddress *string
+	backupDir     *string
 	tls           *bool
 	caFile        *string
+	debug         *bool
+	quiet         *bool
 
 	// MongoDB connection options
 	mongodbConnOptions struct {
@@ -48,43 +57,64 @@ type cliOptios struct {
 }
 
 func main() {
+	log := logrus.New()
 	opts, err := processCliArgs()
 	if err != nil {
 		log.Fatalf("Cannot parse command line arguments: %s", err)
 	}
 
+	log.SetLevel(logrus.InfoLevel)
+	if *opts.quiet {
+		log.SetLevel(logrus.ErrorLevel)
+	}
+	if *opts.debug {
+		log.SetLevel(logrus.DebugLevel)
+	}
+	log.SetLevel(logrus.DebugLevel)
+
 	grpcOpts := getgRPCOptions(opts)
 	clientID := fmt.Sprintf("ABC%04d", rand.Int63n(10000))
-	log.Printf("Using Client ID: %s", clientID)
+	log.Infof("Using Client ID: %s", clientID)
 
 	// Connect to the mongodb-backup gRPC server
 	conn, err := grpc.Dial(*opts.serverAddress, grpcOpts...)
 	if err != nil {
-		log.Fatalf("fail to dial: %v", err)
+		log.Fatalf("Fail to connect to the gRPC server at %q: %v", *opts.serverAddress, err)
 	}
 	defer conn.Close()
-	log.Printf("Connected to the gRPC server at %s", *opts.serverAddress)
+	log.Infof("Connected to the gRPC server at %s", *opts.serverAddress)
 
 	// Connect to the MongoDB instance
 	var di *mgo.DialInfo
 	if *opts.dsn == "" {
 		di = &mgo.DialInfo{
-			Addrs: []string{opts.mongodbConnOptions.Host + ":" + opts.mongodbConnOptions.Port},
+			Addrs:          []string{opts.mongodbConnOptions.Host + ":" + opts.mongodbConnOptions.Port},
+			Username:       opts.mongodbConnOptions.User,
+			Password:       opts.mongodbConnOptions.Password,
+			ReplicaSetName: opts.mongodbConnOptions.ReplicasetName,
+			FailFast:       true,
+			Source:         "admin",
 		}
 	} else {
 		di, err = mgo.ParseURL(*opts.dsn)
+		di.FailFast = true
 		if err != nil {
-			log.Fatalf("Cannot parse MongoDB dsn %q, %s", *opts.dsn, err)
+			log.Fatalf("Cannot parse MongoDB DSN %q, %s", *opts.dsn, err)
 		}
 	}
+
 	mdbSession, err := mgo.DialWithInfo(di)
 	if err != nil {
-		log.Fatalf("Cannot connect to MongoDB on %s: %s", di.Addrs[0], err)
+		log.Fatalf("Cannot connect to MongoDB at %s: %s", di.Addrs[0], err)
 	}
 	defer mdbSession.Close()
-	log.Printf("Connected to MongoDB at %s", di.Addrs[0])
+	log.Infof("Connected to MongoDB at %s", di.Addrs[0])
+
+	client, err := client.NewClient(context.Background(), *opts.backupDir, opts.mongodbConnOptions, opts.mongodbSslOptions, conn, log)
+	if err != nil {
+		log.Fatal(err)
+	}
 
-	client, err := client.NewClient(context.Background(), opts.mongodbConnOptions, opts.mongodbSslOptions, conn)
 	c := make(chan os.Signal, 1)
 	signal.Notify(c, os.Interrupt)
 
@@ -97,22 +127,34 @@ func processCliArgs() (*cliOptios, error) {
 	opts := &cliOptios{
 		app:           app,
 		dsn:           app.Flag("dsn", "MongoDB connection string").String(),
-		serverAddress: app.Flag("server-address", "MongoDB backup server address").String(),
+		serverAddress: app.Flag("server-address", "MongoDB backup server address").Default("127.0.0.1:10000").String(),
+		backupDir:     app.Flag("backup-dir", "Directory where to store the backups").Default("/tmp").String(),
 		tls:           app.Flag("tls", "Use TLS").Bool(),
+		pidFile:       app.Flag("pid-file", "pid file").String(),
 		caFile:        app.Flag("ca-file", "CA file").String(),
+		debug:         app.Flag("debug", "Enable debug log level").Bool(),
+		quiet:         app.Flag("quiet", "Quiet mode. Log only errors").Bool(),
 	}
 
-	app.Flag("mongodb-host", "MongoDB host").StringVar(&opts.mongodbConnOptions.Host)
-	app.Flag("mongodb-port", "MongoDB port").StringVar(&opts.mongodbConnOptions.Port)
-	app.Flag("mongodb-user", "MongoDB username").StringVar(&opts.mongodbConnOptions.User)
-	app.Flag("mongodb-password", "MongoDB password").StringVar(&opts.mongodbConnOptions.Password)
-	app.Flag("replicaset", "Replicaset name").StringVar(&opts.mongodbConnOptions.ReplicasetName)
+	//TODO: Remove defaults. These values are only here to test during development
+	// These params should be Required()
+	app.Flag("mongodb-host", "MongoDB host").Default("127.0.0.1").StringVar(&opts.mongodbConnOptions.Host)
+	app.Flag("mongodb-port", "MongoDB port").Default(os.Getenv("TEST_MONGODB_S1_PRIMARY_PORT")).StringVar(&opts.mongodbConnOptions.Port)
+	app.Flag("mongodb-user", "MongoDB username").Default(os.Getenv("TEST_MONGODB_USERNAME")).StringVar(&opts.mongodbConnOptions.User)
+	app.Flag("mongodb-password", "MongoDB password").Default(os.Getenv("TEST_MONGODB_PASSWORD")).StringVar(&opts.mongodbConnOptions.Password)
+	app.Flag("replicaset", "Replicaset name").Default(os.Getenv("TEST_MONGODB_S1_RS")).StringVar(&opts.mongodbConnOptions.ReplicasetName)
 
 	_, err := app.Parse(os.Args[1:])
 	if err != nil {
 		return nil, err
 	}
 
+	if *opts.pidFile != "" {
+		if err := writePidFile(*opts.pidFile); err != nil {
+			return nil, errors.Wrapf(err, "cannot write pid file %q", *opts.pidFile)
+		}
+	}
+
 	if *opts.dsn != "" {
 		di, err := mgo.ParseURL(*opts.dsn)
 		if err != nil {
@@ -128,6 +170,15 @@ func processCliArgs() (*cliOptios, error) {
 		opts.mongodbConnOptions.Password = di.Password
 		opts.mongodbConnOptions.ReplicasetName = di.ReplicaSetName
 	}
+
+	fi, err := os.Stat(*opts.backupDir)
+	if err != nil {
+		return nil, errors.Wrap(err, "invalid backup destination dir")
+	}
+	if !fi.IsDir() {
+		return nil, fmt.Errorf("%q is not a directory", *opts.backupDir)
+	}
+
 	return opts, nil
 }
 
@@ -144,3 +195,24 @@ func getgRPCOptions(opts *cliOptios) []grpc.DialOption {
 	}
 	return grpcOpts
 }
+
+// Write a pid file, but first make sure it doesn't exist with a running pid.
+func writePidFile(pidFile string) error {
+	// Read in the pid file as a slice of bytes.
+	if piddata, err := ioutil.ReadFile(pidFile); err == nil {
+		// Convert the file contents to an integer.
+		if pid, err := strconv.Atoi(string(piddata)); err == nil {
+			// Look for the pid in the process list.
+			if process, err := os.FindProcess(pid); err == nil {
+				// Send the process a signal zero kill.
+				if err := process.Signal(syscall.Signal(0)); err == nil {
+					// We only get an error if the pid isn't running, or it's not ours.
+					return fmt.Errorf("pid already running: %d", pid)
+				}
+			}
+		}
+	}
+	// If we get here, then the pidfile didn't exist,
+	// or the pid in it doesn't belong to the user running this app.
+	return ioutil.WriteFile(pidFile, []byte(fmt.Sprintf("%d", os.Getpid())), 0664)
+}