Pmb 5

cli/agent/main.go

@@ -11,8 +11,8 @@ import (
 	"github.com/alecthomas/kingpin"
 	"github.com/globalsign/mgo"
 	"github.com/percona/mongodb-backup/grpc/client"
+	"github.com/percona/mongodb-backup/internal/cluster"
 	pb "github.com/percona/mongodb-backup/proto/messages"
-	"github.com/percona/percona-toolkit/src/go/mongolib/proto"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 )
@@ -72,11 +72,11 @@ func main() {
 	nodeType, err := getNodeType(mdbSession)
 
 	// Run the mongodb-backup agent
-	Run(conn, mdbSession, clientID, nodeType)
+	run(conn, mdbSession, clientID, nodeType)
 
 }
 
-func Run(conn *grpc.ClientConn, mdbSession *mgo.Session, clientID, nodeType string) {
+func run(conn *grpc.ClientConn, mdbSession *mgo.Session, clientID string, nodeType pb.NodeType) {
 	messagesClient := pb.NewMessagesClient(conn)
 	rpcClient, err := client.NewClient(clientID, nodeType, messagesClient)
 	if err != nil {
@@ -140,19 +140,22 @@ func getgRPCOptions(opts *cliOptios) []grpc.DialOption {
 	return grpcOpts
 }
 
-func getNodeType(session *mgo.Session) (string, error) {
-	md := proto.MasterDoc{}
-	err := session.Run("isMaster", &md)
+func getNodeType(session *mgo.Session) (pb.NodeType, error) {
+	isMaster, err := cluster.NewIsMaster(session)
 	if err != nil {
-		return "", err
+		return pb.NodeType_UNDEFINED, err
 	}
-
-	if md.SetName != nil || md.Hosts != nil {
-		return "replset", nil
-	} else if md.Msg == "isdbgrid" {
-		// isdbgrid is always the msg value when calling isMaster on a mongos
-		// see http://docs.mongodb.org/manual/core/sharded-cluster-query-router/
-		return "mongos", nil
+	if isMaster.IsShardServer() {
+		return pb.NodeType_MONGOD_SHARDSVR, nil
+	}
+	if isMaster.IsReplset() {
+		return pb.NodeType_MONGOD_REPLSET, nil
+	}
+	if isMaster.IsConfigServer() {
+		return pb.NodeType_MONGOD_CONFIGSVR, nil
+	}
+	if isMaster.IsMongos() {
+		return pb.NodeType_MONGOS, nil
 	}
-	return "mongod", nil
+	return pb.NodeType_MONGOD, nil
 }

cli/agent/main_test.go

@@ -0,0 +1,131 @@
+package main
+
+import (
+	"reflect"
+	"sync"
+	"testing"
+
+	"github.com/globalsign/mgo"
+	"github.com/percona/mongodb-backup/grpc/client"
+	pb "github.com/percona/mongodb-backup/proto/messages"
+	"google.golang.org/grpc"
+)
+
+func Test_main(t *testing.T) {
+	tests := []struct {
+		name string
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			main()
+		})
+	}
+}
+
+func Test_run(t *testing.T) {
+	type args struct {
+		conn       *grpc.ClientConn
+		mdbSession *mgo.Session
+		clientID   string
+		nodeType   pb.NodeType
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			run(tt.args.conn, tt.args.mdbSession, tt.args.clientID, tt.args.nodeType)
+		})
+	}
+}
+
+func Test_processMessages(t *testing.T) {
+	type args struct {
+		rpcClient *client.Client
+		wg        *sync.WaitGroup
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			processMessages(tt.args.rpcClient, tt.args.wg)
+		})
+	}
+}
+
+func Test_processCliArgs(t *testing.T) {
+	tests := []struct {
+		name    string
+		want    *cliOptios
+		wantErr bool
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := processCliArgs()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("processCliArgs() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("processCliArgs() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_getgRPCOptions(t *testing.T) {
+	type args struct {
+		opts *cliOptios
+	}
+	tests := []struct {
+		name string
+		args args
+		want []grpc.DialOption
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getgRPCOptions(tt.args.opts); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("getgRPCOptions() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_getNodeType(t *testing.T) {
+	type args struct {
+		session *mgo.Session
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    pb.NodeType
+		wantErr bool
+	}{
+		// TODO: Add test cases.
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := getNodeType(tt.args.session)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("getNodeType() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("getNodeType() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

docker-compose.yml

@@ -7,6 +7,10 @@ services:
     user: root
     entrypoint: /entrypoint-mongod.sh
     command: --replSet=${TEST_PSMDB_RSNAME:-rs} --port=${TEST_MONGODB_PRIMARY_PORT:-17001} --shardsvr
+    expose: 
+        - ${TEST_MONGODB_PRIMARY_PORT:-17001}
+    ports:
+      - ${TEST_MONGODB_PRIMARY_PORT:-17001}:${TEST_MONGODB_PRIMARY_PORT:-17001} 
     volumes:
     - ./scripts/entrypoint-mongod.sh:/entrypoint-mongod.sh:ro
     - ./scripts/mongod.key:/mongod.key:ro
@@ -19,6 +23,10 @@ services:
     user: root
     entrypoint: /entrypoint-mongod.sh
     command: --replSet=${TEST_PSMDB_RSNAME:-rs} --port=${TEST_MONGODB_SECONDARY1_PORT:-17002} --shardsvr
+    expose: 
+        - ${TEST_MONGODB_SECONDARY1_PORT:-17002}
+    ports:
+        - ${TEST_MONGODB_SECONDARY1_PORT:-17002}:${TEST_MONGODB_SECONDARY1_PORT:-17002}
     volumes:
     - ./scripts/entrypoint-mongod.sh:/entrypoint-mongod.sh:ro
     - ./scripts/mongod.key:/mongod.key:ro
@@ -31,6 +39,10 @@ services:
     user: root
     entrypoint: /entrypoint-mongod.sh
     command: --replSet=${TEST_PSMDB_RSNAME:-rs} --port=${TEST_MONGODB_SECONDARY2_PORT:-17003} --shardsvr
+    expose: 
+        - ${TEST_MONGODB_SECONDARY2_PORT:-17003}
+    ports:
+        - ${TEST_MONGODB_SECONDARY2_PORT:-17003}:${TEST_MONGODB_SECONDARY2_PORT:-17003}
     volumes:
     - ./scripts/entrypoint-mongod.sh:/entrypoint-mongod.sh:ro
     - ./scripts/mongod.key:/mongod.key:ro
@@ -43,6 +55,10 @@ services:
     user: root
     entrypoint: /entrypoint-mongod.sh
     command: --replSet=${TEST_MONGODB_CONFIGSVR_RS:-csReplSet} --port=${TEST_MONGODB_CONFIGSVR1_PORT:-17004} --configsvr
+    expose: 
+        - ${TEST_MONGODB_CONFIGSVR1_PORT:-17004}
+    ports:
+        - ${TEST_MONGODB_CONFIGSVR1_PORT:-17004}:${TEST_MONGODB_CONFIGSVR1_PORT:-17004}
     volumes:
     - ./scripts/entrypoint-mongod.sh:/entrypoint-mongod.sh:ro
     - ./scripts/mongod.key:/mongod.key:ro
@@ -54,6 +70,12 @@ services:
     image: percona/percona-server-mongodb:${TEST_PSMDB_VERSION:-latest}
     entrypoint: /entrypoint-mongos.sh
     command: --port=${TEST_MONGODB_MONGOS_PORT:-17005} --configdb=${TEST_MONGODB_CONFIGSVR_RS:-csReplSet}/127.0.0.1:${TEST_MONGODB_CONFIGSVR1_PORT:-17004} 
+    expose: 
+        - ${TEST_MONGODB_CONFIGSVR1_PORT:-17004}
+        - ${TEST_MONGODB_MONGOS_PORT:-17005}
+    ports:
+        - ${TEST_MONGODB_MONGOS_PORT:-17005}:${TEST_MONGODB_MONGOS_PORT:-17005}
+        - ${TEST_MONGODB_CONFIGSVR1_PORT:-17004}:${TEST_MONGODB_CONFIGSVR1_PORT:-17004}
     volumes:
     - ./scripts/entrypoint-mongos.sh:/entrypoint-mongos.sh:ro
     - ./scripts/mongod.key:/mongos.key:ro

grpc/api/api_test.go

@@ -53,6 +53,7 @@ func TestServerAndClients(t *testing.T) {
 		&pb.ClientMessage{
 			Type:     pb.ClientMessage_REGISTER,
 			ClientID: clientID,
+			Payload:  &pb.ClientMessage_RegisterMsg{RegisterMsg: &pb.RegisterPayload{NodeType: pb.NodeType_MONGOD}},
 		},
 		nil,
 	}
@@ -75,6 +76,7 @@ func TestServerAndClients(t *testing.T) {
 		&pb.ClientMessage{
 			Type:     pb.ClientMessage_PONG,
 			ClientID: clientID,
+			Payload:  &pb.ClientMessage_RegisterMsg{RegisterMsg: &pb.RegisterPayload{NodeType: pb.NodeType_MONGOD}},
 		},
 		nil,
 	}
@@ -91,7 +93,7 @@ func TestServerAndClients(t *testing.T) {
 	})
 
 	apiServer := NewApiServer(messagesServer)
-	apiServer.GetClients(pbapi.Empty{}, apiStream)
+	apiServer.GetClients(&pbapi.Empty{}, apiStream)
 	msg := <-apiOutChan
 	if msg.(*pbapi.Client).ClientID != clientID {
 		t.Errorf("Received invalid clientID")
@@ -113,7 +115,7 @@ func TestServerAndClients(t *testing.T) {
 	}
 
 	// Check there are no messages in the stream after unregistring the client
-	apiServer.GetClients(pbapi.Empty{}, apiStream)
+	apiServer.GetClients(&pbapi.Empty{}, apiStream)
 	select {
 	case <-apiOutChan:
 		t.Error("Received a client but the clients list should be empty")