New restore method

Author: percona-csalguero

cli/pbm-coordinator/main.go

@@ -58,7 +58,7 @@ const (
 	defaultAPIPort         = 10001
 	defaultShutdownTimeout = 5 // Seconds
 	defaultClientsLogging  = true
-	defaultDebugMode       = true
+	defaultDebugMode       = false
 	defaultWorkDir         = "~/percona-backup-mongodb"
 )
 

grpc/api/api.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"os"
 	"path/filepath"
 	"time"
 
@@ -28,7 +29,9 @@ var (
 )
 
 func init() {
-	logger.SetLevel(logrus.DebugLevel)
+	if os.Getenv("DEBUG") == "1" {
+		logger.SetLevel(logrus.DebugLevel)
+	}
 }
 
 func (a *ApiServer) GetClients(m *pbapi.Empty, stream pbapi.Api_GetClientsServer) error {

grpc/client/client.go

@@ -17,7 +17,6 @@ import (
 	"github.com/globalsign/mgo"
 	"github.com/globalsign/mgo/bson"
 	"github.com/golang/snappy"
-	"github.com/kr/pretty"
 	"github.com/percona/percona-backup-mongodb/bsonfile"
 	"github.com/percona/percona-backup-mongodb/internal/awsutils"
 	"github.com/percona/percona-backup-mongodb/internal/backup/dumper"
@@ -364,11 +363,10 @@ func (c *Client) processIncommingServerMessages() {
 		case *pb.ServerMessage_ListReplicasets:
 			c.processListReplicasets()
 		case *pb.ServerMessage_PingMsg:
-			c.processPing()
-			// msg := c.processPing()
-			//if err := c.streamSend(msg); err != nil {
-			//	c.logger.Errorf("Cannot stream ping response to the server: %s. Out message: %+v. In message type: %T", err, *msg, msg.Payload)
-			//}
+			msg := c.processPing()
+			if err := c.streamSend(msg); err != nil {
+				c.logger.Errorf("Cannot stream ping response to the server: %s. Out message: %+v. In message type: %T", err, *msg, msg.Payload)
+			}
 			continue
 		case *pb.ServerMessage_CanRestoreBackupMsg:
 			msg, err := c.processCanRestoreBackup(msg.GetCanRestoreBackupMsg())
@@ -626,7 +624,7 @@ func (c *Client) processListReplicasets() error {
 
 func (c *Client) processPing() *pb.ClientMessage {
 	c.logger.Debug("Received Ping command")
-	//c.updateClientInfo()
+	c.updateClientInfo()
 
 	pongMsg := &pb.Pong{
 		Timestamp:           time.Now().Unix(),
@@ -658,7 +656,6 @@ func (c *Client) processRestore(msg *pb.RestoreBackup) error {
 	}()
 
 	c.sendACK()
-	pretty.Print(msg)
 
 	if err := c.restoreDBDump(msg); err != nil {
 		err := errors.Wrap(err, "cannot restore DB backup")
@@ -911,7 +908,6 @@ func (c *Client) runDBBackup(msg *pb.StartBackup) {
 		fw, err := os.Create(path.Join(c.backupDir, msg.GetDbBackupName()))
 		if err != nil {
 			log.Errorf("Cannot create backup file: %s", err)
-			// TODO Stream error msg to the server
 		}
 		writers = append(writers, fw)
 	}

grpc/server/clients.go

@@ -267,25 +267,24 @@ func (c *Client) listReplicasets() ([]string, error) {
 }
 
 func (c *Client) ping() error {
-	//c.logger.Debug("sending ping")
+	c.logger.Debug("sending ping")
 	err := c.streamSend(&pb.ServerMessage{Payload: &pb.ServerMessage_PingMsg{PingMsg: &pb.Ping{}}})
 	if err != nil {
 		return errors.Wrap(err, "clients.go -> ping()")
 	}
 
-	//msg, err := c.streamRecv()
-	//if err != nil {
-	//	return errors.Wrapf(err, "ping client %s (%s)", c.ID, c.NodeName)
-	//}
-	//pretty.Println(msg)
-	//pongMsg := msg.GetPongMsg()
-	//c.statusLock.Lock()
-	//c.NodeType = pongMsg.GetNodeType()
-	//c.ReplicasetUUID = pongMsg.GetReplicaSetUuid()
-	//c.ReplicasetVersion = pongMsg.GetReplicaSetVersion()
-	//c.isTailing = pongMsg.GetIsTailing()
-	//c.lastTailedTimestamp = pongMsg.GetLastTailedTimestamp()
-	//c.statusLock.Unlock()
+	msg, err := c.streamRecv()
+	if err != nil {
+		return errors.Wrapf(err, "ping client %s (%s)", c.ID, c.NodeName)
+	}
+	pongMsg := msg.GetPongMsg()
+	c.statusLock.Lock()
+	c.NodeType = pongMsg.GetNodeType()
+	c.ReplicasetUUID = pongMsg.GetReplicaSetUuid()
+	c.ReplicasetVersion = pongMsg.GetReplicaSetVersion()
+	c.isTailing = pongMsg.GetIsTailing()
+	c.lastTailedTimestamp = pongMsg.GetLastTailedTimestamp()
+	c.statusLock.Unlock()
 
 	return nil
 }

grpc/server/server.go

@@ -137,12 +137,6 @@ func (s *MessagesServer) RestoreSourcesByReplicaset(bm *pb.BackupMetadata) (map[
 			sources[resp.Replicaset] = s
 		}
 	}
-	for rs, source := range sources {
-		fmt.Printf("RS: %v\n", rs)
-		fmt.Printf("Client: %v\n", source.Client.NodeName)
-		fmt.Printf("Host: %v\n", source.Host)
-		fmt.Printf("Port: %v\n", source.Port)
-	}
 	return sources, nil
 }
 
@@ -330,9 +324,6 @@ func (s *MessagesServer) RestoreBackUp(bm *pb.BackupMetadata, skipUsersAndRoles
 	// Ping will also update the status and if it is primary or secondary
 	for _, source := range clients {
 		source.Client.ping()
-		fmt.Printf("Client ID  : %v\n", source.Client.ID)
-		fmt.Printf("Node Name  : %v\n", source.Client.NodeName)
-		fmt.Printf("Is primary : %v\n", source.Client.isPrimary)
 	}
 
 	for replName, source := range clients {

internal/backup/dumper/dumper.go

@@ -8,6 +8,7 @@ import (
 	"github.com/mongodb/mongo-tools/common/options"
 	"github.com/mongodb/mongo-tools/mongodump"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 type MongodumpInput struct {
@@ -32,6 +33,21 @@ type Mongodump struct {
 	//
 	lock    *sync.Mutex
 	running bool
+
+	mdumpLogger []byte
+}
+
+type log2LogrusWriter struct {
+	entry *logrus.Entry
+}
+
+func (w *log2LogrusWriter) Write(b []byte) (int, error) {
+	n := len(b)
+	if n > 0 && b[n-1] == '\n' {
+		b = b[:n-1]
+	}
+	w.entry.Info(string(b))
+	return n, nil
 }
 
 func NewMongodump(i *MongodumpInput) (*Mongodump, error) {
@@ -82,12 +98,15 @@ func NewMongodump(i *MongodumpInput) (*Mongodump, error) {
 
 	dump.OutputWriter = i.Writer
 
-	return &Mongodump{
+	mongoDump := &Mongodump{
 		MongodumpInput: i,
 		mongodump:      dump,
 		lock:           &sync.Mutex{},
 		running:        false,
-	}, nil
+	}
+	//mdumpLogger.SetWriter(logrus.StandardLogger().Writer())
+
+	return mongoDump, nil
 }
 
 func (md *Mongodump) LastError() error {

tests/general_test.go

@@ -356,7 +356,7 @@ func TestValidateReplicasetAgents(t *testing.T) {
 	time.Sleep(10 * time.Second)
 	for _, client := range d.Clients() {
 		if client.ReplicasetName() == "rs1" {
-			fmt.Printf("Stopping client: %s, rs: %s\n", client.NodeName(), client.ReplicasetName())
+			log.Infof("Stopping client: %s, rs: %s\n", client.NodeName(), client.ReplicasetName())
 			client.Stop()
 			time.Sleep(1 * time.Second)
 		}
@@ -466,7 +466,7 @@ func TestBackup1(t *testing.T) {
 		t.Fatalf("Cannot start backup: %s", err)
 	}
 
-	fmt.Printf("starting backup: %s\n", backupNamePrefix)
+	log.Infof("starting backup: %s\n", backupNamePrefix)
 
 	d.MessagesServer.WaitBackupFinish()
 	err = d.MessagesServer.StopOplogTail()