Add internal cluster v5

internal/cluster/cluster.go

@@ -4,16 +4,38 @@ import (
 	"github.com/percona/mongodb-backup/mdbstructs"
 )
 
+type Config struct {
+	Username string
+	Password string
+	AuthDB   string
+}
+
 type Cluster struct {
 	shards map[string]*Shard
 }
 
-func New(shards []*mdbstructs.Shard) *Cluster {
+func New(config *Config, shards []*mdbstructs.Shard) (*Cluster, error) {
 	c := &Cluster{
 		shards: make(map[string]*Shard),
 	}
 	for _, shard := range shards {
-		c.shards[shard.Id] = NewShard(shard)
+		var err error
+		c.shards[shard.Id], err = NewShard(config, shard)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return c, nil
+}
+
+func (c *Cluster) GetBackupSources() ([]*mdbstructs.ReplsetConfigMember, error) {
+	sources := []*mdbstructs.ReplsetConfigMember{}
+	for _, shard := range c.shards {
+		source, err := shard.replset.GetBackupSource()
+		if err != nil {
+			return sources, err
+		}
+		sources = append(sources, source)
 	}
-	return c
+	return sources, nil
 }

internal/cluster/cluster_test.go

@@ -3,19 +3,93 @@ package cluster
 import (
 	"testing"
 
-	"github.com/percona/mongodb-backup/mdbstructs"
+	"github.com/globalsign/mgo"
+	"github.com/percona/mongodb-backup/internal/testutils"
+)
+
+const (
+	testListShardsFile = "testdata/listShards.bson"
 )
 
 func TestNew(t *testing.T) {
-	cluster := New([]*mdbstructs.Shard{{
-		Id:   "shard1",
-		Host: "rs/127.0.0.1:27017,127.0.0.1:27018,127.0.0.1:27019",
-	}})
+	session, err := mgo.DialWithInfo(testutils.MongosDialInfo())
+	if err != nil {
+		t.Fatalf("Got error getting test db session: %v", err.Error())
+	}
+	defer session.Close()
+
+	listShards, err := GetListShards(session)
+	if err != nil {
+		t.Fatalf("Got error running .GetListShards(): %v", err.Error())
+	}
+
+	cluster, err := New(testClusterConfig, listShards.Shards)
+	if err != nil {
+		t.Fatalf("Got error running .New(): %v", err.Error())
+	}
 	if len(cluster.shards) != 1 {
 		t.Fatal("Got unexpected number of shards")
 	}
-	shard := cluster.shards["shard1"]
-	if len(shard.replset.addrs) != 3 {
+	shard := cluster.shards[testutils.MongoDBReplsetName]
+	if len(shard.replset.addrs) != 2 {
 		t.Fatal("Got unexpected replset addresses for shard")
 	}
 }
+
+func TestGetBackupSourcesMongos(t *testing.T) {
+	session, err := mgo.DialWithInfo(testutils.MongosDialInfo())
+	if err != nil {
+		t.Fatalf("Got error getting test db session: %v", err.Error())
+	}
+	defer session.Close()
+
+	listShards, err := GetListShards(session)
+	if err != nil {
+		t.Fatalf("Got error running .GetListShards(): %v", err.Error())
+	}
+
+	cluster, err := New(testClusterConfig, listShards.Shards)
+	if err != nil {
+		t.Fatalf("Got error running .GetBackupSources(): %v", err.Error())
+	}
+
+	sources, err := cluster.GetBackupSources()
+	if err != nil {
+		t.Fatalf("Got error running .GetBackupSources(): %v", err.Error())
+	}
+	if len(sources) != 1 {
+		t.Fatal(".GetBackupSources() did not return 1 backup source")
+	}
+	if sources[0].Host != testSecondary2Host {
+		t.Fatalf(".GetBackupSources() did not return the winner: %v", testSecondary2Host)
+	}
+}
+
+func TestGetBackupSourcesConfigsvr(t *testing.T) {
+	session, err := mgo.DialWithInfo(testutils.ConfigsvrReplsetDialInfo())
+	if err != nil {
+		t.Fatalf("Got error getting test db session: %v", err.Error())
+	}
+	defer session.Close()
+
+	shards, err := GetConfigsvrShards(session)
+	if err != nil {
+		t.Fatalf("Got error running .GetConfigsvrShards(): %v", err.Error())
+	}
+
+	cluster, err := New(testClusterConfig, shards)
+	if err != nil {
+		t.Fatalf("Got error running .GetBackupSources(): %v", err.Error())
+	}
+
+	sources, err := cluster.GetBackupSources()
+	if err != nil {
+		t.Fatalf("Got error running .GetBackupSources(): %v", err.Error())
+	}
+	if len(sources) != 1 {
+		t.Fatal(".GetBackupSources() did not return 1 backup source")
+	}
+	if sources[0].Host != testSecondary2Host {
+		t.Fatalf(".GetBackupSources() did not return the winner: %v", testSecondary2Host)
+	}
+}

internal/cluster/replset.go

@@ -9,30 +9,27 @@ import (
 	"github.com/percona/mongodb-backup/mdbstructs"
 )
 
-const (
-	hiddenMemberWeight = 0.2
-	priorityZeroWeight = 0.1
-)
-
 var (
 	replsetReadPreference = mgo.PrimaryPreferred
 )
 
+//func HasReplsetTag(config *mdbstructs.ReplsetConfig, key, val string) bool {
+//}
+
 type Replset struct {
 	sync.Mutex
-	name     string
-	addrs    []string
-	username string
-	password string
-	session  *mgo.Session
+	name    string
+	addrs   []string
+	config  *Config
+	session *mgo.Session
+	scorer  *ReplsetScorer
 }
 
-func NewReplset(name string, addrs []string, username, password string) (*Replset, error) {
+func NewReplset(config *Config, name string, addrs []string) (*Replset, error) {
 	r := &Replset{
-		name:     name,
-		addrs:    addrs,
-		username: username,
-		password: password,
+		name:   name,
+		addrs:  addrs,
+		config: config,
 	}
 	return r, r.getSession()
 }
@@ -44,8 +41,8 @@ func (r *Replset) getSession() error {
 	var err error
 	r.session, err = mgo.DialWithInfo(&mgo.DialInfo{
 		Addrs:          r.addrs,
-		Username:       r.username,
-		Password:       r.password,
+		Username:       r.config.Username,
+		Password:       r.config.Password,
 		ReplicaSetName: r.name,
 		Timeout:        10 * time.Second,
 	})
@@ -71,12 +68,18 @@ func (r *Replset) GetConfig() (*mdbstructs.ReplsetConfig, error) {
 	return rsGetConfig.Config, err
 }
 
-func (r *Replset) GetStatus() (*mdbstructs.ReplsetStatus, error) {
-	status := mdbstructs.ReplsetStatus{}
-	err := r.session.Run(bson.D{{"replSetGetStatus", "1"}}, &status)
-	return &status, err
-}
-
-func getBackupNode(config *mdbstructs.ReplsetConfig, status *mdbstructs.ReplsetStatus) (*mdbstructs.ReplsetConfigMember, error) {
-	return config.Members[0], nil
+func (r *Replset) GetBackupSource() (*mdbstructs.ReplsetConfigMember, error) {
+	config, err := r.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+	status, err := r.GetStatus()
+	if err != nil {
+		return nil, err
+	}
+	scorer, err := ScoreReplset(config, status, nil)
+	if err != nil {
+		return nil, err
+	}
+	return scorer.Winner().config, nil
 }