PMB-54

Author: percona-csalguero

cli/mongodb-backupd/main.go

@@ -23,6 +23,7 @@ type cliOptions struct {
 	app             *kingpin.Application
 	cmd             string
 	tls             *bool
+	workDir         *string
 	certFile        *string
 	keyFile         *string
 	grpcPort        *int
@@ -46,6 +47,7 @@ func processCliParams() (*cliOptions, error) {
 	opts := &cliOptions{
 		app:             app,
 		tls:             app.Flag("tls", "Enable TLS").Bool(),
+		workDir:         app.Flag("work-dir", "Working directory for backup metadata").String(),
 		certFile:        app.Flag("cert-file", "Cert file for gRPC client connections").String(),
 		keyFile:         app.Flag("key-file", "Key file for gRPC client connections").String(),
 		grpcPort:        app.Flag("grpc-port", "Listening port for client connections").Default(defaultGrpcPort).Int(),
@@ -97,7 +99,7 @@ func main() {
 	wg := &sync.WaitGroup{}
 
 	grpcServer := grpc.NewServer(grpcOpts...)
-	messagesServer := server.NewMessagesServer(log)
+	messagesServer := server.NewMessagesServer(*opts.workDir, log)
 	pb.RegisterMessagesServer(grpcServer, messagesServer)
 
 	wg.Add(1)

grpc/api/api.go

@@ -80,6 +80,11 @@ func (a *ApiServer) RunBackup(ctx context.Context, opts *pbapi.RunBackupParams)
 		Cypher:          pb.Cypher(opts.Cypher),
 	}
 
+	logger.Debug("Stopping the balancer")
+	if err := a.messagesServer.StopBalancer(); err != nil {
+		return &pbapi.Error{Message: err.Error()}, err
+	}
+
 	if err := a.messagesServer.StartBackup(msg); err != nil {
 		return &pbapi.Error{Message: err.Error()}, err
 	}
@@ -97,5 +102,9 @@ func (a *ApiServer) RunBackup(ctx context.Context, opts *pbapi.RunBackupParams)
 	a.messagesServer.WaitOplogBackupFinish()
 	logger.Debug("Oplog finished")
 
+	logger.Debug("Starting the balancer")
+	if err := a.messagesServer.StartBalancer(); err != nil {
+		return &pbapi.Error{Message: err.Error()}, err
+	}
 	return &pbapi.Error{}, nil
 }

grpc/server/clients.go

@@ -134,10 +134,32 @@ func (c *Client) Status() (pb.Status, error) {
 	return *statusMsg, nil
 }
 
+func (c *Client) startBalancer() error {
+	err := c.streamSend(&pb.ServerMessage{
+		Payload: &pb.ServerMessage_StartBalancerMsg{},
+	})
+	if err != nil {
+		return err
+	}
+	msg, err := c.streamRecv()
+	if err != nil {
+		return err
+	}
+
+	switch msg.Payload.(type) {
+	case *pb.ClientMessage_AckMsg:
+		return nil
+	case *pb.ClientMessage_ErrorMsg:
+		errMsg := msg.GetErrorMsg()
+		return fmt.Errorf("%s", errMsg.Message)
+	}
+	return fmt.Errorf("unknown respose type %T", msg)
+}
+
 func (c *Client) stopBalancer() error {
 	err := c.streamSend(&pb.ServerMessage{
 		Type:    pb.ServerMessage_STOP_BALANCER,
-		Payload: &pb.ServerMessage_EmptyMsg{},
+		Payload: &pb.ServerMessage_StopBalancerMsg{},
 	})
 	if err != nil {
 		return err

grpc/server/metadata.go

@@ -0,0 +1,98 @@
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"sync"
+	"time"
+
+	pb "github.com/percona/mongodb-backup/proto/messages"
+	"github.com/pkg/errors"
+)
+
+type ReplicasetMetadata struct {
+	ReplicasetUUID  string `json:"replicaset_uuid"`
+	ReplicasetName  string `json:"replicaset_name"`
+	DestinationName string `json:"destination_name"`
+}
+
+type BackupMetadata struct {
+	StartTs         time.Time          `json:"start_ts"`
+	EndTs           time.Time          `json:"end_ts"`
+	BackupType      pb.BackupType      `json:"backup_type"`
+	OplogStartTime  int64              `json:"oplog_start_time"`
+	LastOplogTs     int64              `json:"last_oplog_ts"`
+	DestinationType pb.DestinationType `json:"destination_type"`
+	DestinationDir  string             `json:"destination_dir"`
+	Cypher          pb.Cypher          `json:"cypher"`
+	CompressionType pb.CompressionType `json:"compression_type"`
+
+	lock        *sync.Mutex                   `json:"-"`
+	Replicasets map[string]ReplicasetMetadata `json:"replicas"` // key is replicaset name
+}
+
+func NewBackupMetadata() *BackupMetadata {
+	return &BackupMetadata{
+		Replicasets: make(map[string]ReplicasetMetadata),
+		lock:        &sync.Mutex{},
+	}
+}
+
+func LoadMetadataFromFile(name string) (*BackupMetadata, error) {
+	buf, err := ioutil.ReadFile(name)
+	if err != nil {
+		return nil, err
+	}
+	metadata := &BackupMetadata{
+		Replicasets: make(map[string]ReplicasetMetadata),
+		lock:        &sync.Mutex{},
+	}
+	err = json.Unmarshal(buf, metadata)
+	return metadata, nil
+}
+
+// AddReplicaset adds backup info for a replicaset using the replicaset name as the key
+func (b *BackupMetadata) AddReplicaset(replName, replUUID, destinationName string) error {
+	b.lock.Lock()
+
+	if _, ok := b.Replicasets[replName]; ok {
+		return fmt.Errorf("Info for replicaset %s already exists", replName)
+	}
+
+	b.Replicasets[replName] = ReplicasetMetadata{
+		ReplicasetUUID:  replUUID,
+		ReplicasetName:  replName,
+		DestinationName: destinationName,
+	}
+
+	b.lock.Unlock()
+	return nil
+}
+
+func (b *BackupMetadata) RemoveReplicaset(replName string) error {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	if _, ok := b.Replicasets[replName]; !ok {
+		return fmt.Errorf("Info for replicaset %s doesn't exists", replName)
+	}
+	delete(b.Replicasets, replName)
+	return nil
+}
+
+// WriteMetadataToFile writes the backup metadata to a file as JSON
+func (b *BackupMetadata) WriteMetadataToFile(name string) error {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	buf, err := json.Marshal(b)
+	if err != nil {
+		return errors.Wrap(err, "cannot encode backup metadata")
+	}
+	if err = ioutil.WriteFile(name, buf, os.ModePerm); err != nil {
+		return err
+	}
+	return nil
+}

grpc/server/server.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"path"
 	"sync"
 	"time"
 
@@ -29,13 +30,15 @@ type MessagesServer struct {
 	oplogBackupRunning    bool
 	err                   error
 	//
+	workDir               string
+	lastBackupMetadata    *BackupMetadata
 	clientDisconnetedChan chan string
 	dbBackupFinishChan    chan interface{}
 	oplogBackupFinishChan chan interface{}
 	logger                *logrus.Logger
 }
 
-func NewMessagesServer(logger *logrus.Logger) *MessagesServer {
+func NewMessagesServer(workDir string, logger *logrus.Logger) *MessagesServer {
 	if logger == nil {
 		logger = logrus.New()
 		logger.SetLevel(logrus.StandardLogger().Level)
@@ -53,6 +56,7 @@ func NewMessagesServer(logger *logrus.Logger) *MessagesServer {
 		dbBackupFinishChan:    bfc,
 		oplogBackupFinishChan: ofc,
 		replicasRunningBackup: make(map[string]bool),
+		workDir:               workDir,
 		logger:                logger,
 	}
 
@@ -186,6 +190,17 @@ func (s *MessagesServer) StartBackup(opts *pb.StartBackup) error {
 	if s.isBackupRunning() {
 		return fmt.Errorf("Backup is already running")
 	}
+
+	s.lastBackupMetadata = NewBackupMetadata()
+
+	s.lastBackupMetadata.StartTs = time.Now()
+	s.lastBackupMetadata.Replicasets = make(map[string]ReplicasetMetadata)
+	s.lastBackupMetadata.BackupType = opts.GetBackupType()
+	s.lastBackupMetadata.DestinationType = opts.GetDestinationType()
+	s.lastBackupMetadata.DestinationDir = opts.GetDestinationDir()
+	s.lastBackupMetadata.CompressionType = opts.GetCompressionType()
+	s.lastBackupMetadata.Cypher = opts.GetCypher()
+
 	clients, err := s.BackupSourceByReplicaset()
 	if err != nil {
 		return errors.Wrapf(err, "Cannot start backup. Cannot find backup source for replicas")
@@ -198,7 +213,10 @@ func (s *MessagesServer) StartBackup(opts *pb.StartBackup) error {
 	for replName, client := range clients {
 		s.logger.Printf("Starting backup for replicaset %q on client %s %s %s", replName, client.ID, client.NodeName, client.NodeType)
 		s.replicasRunningBackup[replName] = true
-		destinationName := fmt.Sprintf("%s_%s_%s", time.Now().Format("2006-01-02_15.04.05"), client.ReplicasetName, opts.GetDestinationName())
+		destinationName := fmt.Sprintf("%s_%s_%s", s.lastBackupMetadata.StartTs.Format("2006-01-02_15.04.05"), client.ReplicasetName, opts.GetDestinationName())
+
+		s.lastBackupMetadata.AddReplicaset(client.ReplicasetName, client.ReplicasetUUID, destinationName)
+
 		client.startBackup(&pb.StartBackup{
 			BackupType:      opts.GetBackupType(),
 			DestinationType: opts.GetDestinationType(),
@@ -210,9 +228,28 @@ func (s *MessagesServer) StartBackup(opts *pb.StartBackup) error {
 		})
 	}
 
+	metadataFilename := path.Join(s.workDir, fmt.Sprintf("%s.json", s.lastBackupMetadata.StartTs.Format("2006-01-02_15.04.05")))
+	err = s.lastBackupMetadata.WriteMetadataToFile(metadataFilename)
+	if err != nil {
+		log.Warn("Cannot write metadata file %s: %s", metadataFilename, err)
+	}
 	return nil
 }
 
+// StartBalancer restarts the balancer if this is a sharded system
+func (s *MessagesServer) StartBalancer() error {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	for _, client := range s.clients {
+		if client.NodeType == pb.NodeType_MONGOS {
+			return client.startBalancer()
+		}
+	}
+	// This is not a sharded system. There is nothing to do.
+	return nil
+}
+
+// StopBalancer stops the balancer if this is a sharded system
 func (s *MessagesServer) StopBalancer() error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -221,7 +258,8 @@ func (s *MessagesServer) StopBalancer() error {
 			return client.stopBalancer()
 		}
 	}
-	return fmt.Errorf("No MongoS server found to stop the balancer")
+	// This is not a sharded system. There is nothing to do.
+	return nil
 }
 
 func (s *MessagesServer) cancelBackup() error {

internal/testutils/grpc-daeamon.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"os"
-	"path"
 	"sync"
 	"testing"
 	"time"
@@ -43,7 +41,7 @@ type GrpcDaemon struct {
 	clients            []*client.Client
 }
 
-func NewGrpcDaemon(ctx context.Context, t *testing.T, logger *logrus.Logger) (*GrpcDaemon, error) {
+func NewGrpcDaemon(ctx context.Context, workDir string, t *testing.T, logger *logrus.Logger) (*GrpcDaemon, error) {
 	if logger == nil {
 		logger = logrus.New()
 		logger.SetLevel(logrus.StandardLogger().Level)
@@ -56,13 +54,7 @@ func NewGrpcDaemon(ctx context.Context, t *testing.T, logger *logrus.Logger) (*G
 		logger:  logger,
 		lock:    &sync.Mutex{},
 	}
-
-	tmpDir := path.Join(os.TempDir(), "dump_test")
-	os.RemoveAll(tmpDir) // Don't check for errors. The path might not exist
-	err := os.MkdirAll(tmpDir, os.ModePerm)
-	if err != nil {
-		return nil, fmt.Errorf("Cannot create temp dir %q, %s", tmpDir, err)
-	}
+	var err error
 
 	// Start the grpc server
 	d.msgListener, err = net.Listen("tcp", fmt.Sprintf("localhost:%s", TEST_GRPC_MESSAGES_PORT))
@@ -73,7 +65,7 @@ func NewGrpcDaemon(ctx context.Context, t *testing.T, logger *logrus.Logger) (*G
 	d.ctx, d.cancelFunc = context.WithCancel(ctx)
 	// This is the sever/agents gRPC server
 	d.grpcServer4Clients = grpc.NewServer(opts...)
-	d.MessagesServer = server.NewMessagesServer(logger)
+	d.MessagesServer = server.NewMessagesServer(workDir, logger)
 	pb.RegisterMessagesServer(d.grpcServer4Clients, d.MessagesServer)
 
 	d.wg.Add(1)
@@ -128,7 +120,7 @@ func NewGrpcDaemon(ctx context.Context, t *testing.T, logger *logrus.Logger) (*G
 			ReplicasetName: di.ReplicaSetName,
 		}
 
-		client, err := client.NewClient(d.ctx, tmpDir, dbConnOpts, client.SSLOptions{}, clientConn, logger)
+		client, err := client.NewClient(d.ctx, workDir, dbConnOpts, client.SSLOptions{}, clientConn, logger)
 		if err != nil {
 			return nil, fmt.Errorf("Cannot create an agent instance %s: %s", agentID, err)
 		}