PBM-1159: Make ReadConcern Majority optional (#911)

boris-ilijic · web-flow · commit d6713f00a49d · 2024-03-07T09:35:14.000+01:00
* Introduce MongoOption functional options
* Apply new interface for pkg connect
* Add doc for connect.Connect func
diff --git a/cmd/pbm-agent/agent.go b/cmd/pbm-agent/agent.go
@@ -44,27 +44,27 @@ type Agent struct {
 }
 
 func newAgent(ctx context.Context, leadConn connect.Client, uri string, dumpConns int) (*Agent, error) {
-	m, err := connect.MongoConnect(ctx, uri, &connect.MongoConnectOptions{Direct: true})
+	nodeConn, err := connect.MongoConnect(ctx, uri, connect.Direct(true))
 	if err != nil {
 		return nil, err
 	}
 
-	info, err := topo.GetNodeInfo(ctx, m)
+	info, err := topo.GetNodeInfo(ctx, nodeConn)
 	if err != nil {
 		return nil, err
 	}
 
 	a := &Agent{
 		leadConn: leadConn,
 		closeCMD: make(chan struct{}),
+		nodeConn: nodeConn,
+		brief: topo.NodeBrief{
+			URI:     uri,
+			SetName: info.SetName,
+			Me:      info.Me,
+		},
+		dumpConns: dumpConns,
 	}
-	a.nodeConn = m
-	a.brief = topo.NodeBrief{
-		URI:     uri,
-		SetName: info.SetName,
-		Me:      info.Me,
-	}
-	a.dumpConns = dumpConns
 	return a, nil
 }
 
diff --git a/cmd/pbm-agent/main.go b/cmd/pbm-agent/main.go
@@ -86,7 +86,7 @@ func runAgent(mongoURI string, dumpConns int) error {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	leadConn, err := connect.Connect(ctx, mongoURI, &connect.ConnectOptions{AppName: "pbm-agent"})
+	leadConn, err := connect.Connect(ctx, mongoURI, "pbm-agent")
 	if err != nil {
 		return errors.Wrap(err, "connect to PBM")
 	}
diff --git a/cmd/pbm-speed-test/main.go b/cmd/pbm-speed-test/main.go
@@ -86,7 +86,7 @@ func testCompression(mURL string, compression compress.CompressionType, level *i
 	var cn *mongo.Client
 
 	if collection != "" {
-		cn, err := connect.MongoConnect(ctx, mURL, &connect.MongoConnectOptions{Direct: true})
+		cn, err := connect.MongoConnect(ctx, mURL, connect.Direct(true))
 		if err != nil {
 			stdlog.Fatalln("Error: connect to mongodb-node:", err)
 		}
@@ -108,13 +108,13 @@ func testCompression(mURL string, compression compress.CompressionType, level *i
 }
 
 func testStorage(mURL string, compression compress.CompressionType, level *int, sizeGb float64, collection string) {
-	sess, err := connect.MongoConnect(context.Background(), mURL, &connect.MongoConnectOptions{Direct: true})
+	sess, err := connect.MongoConnect(context.Background(), mURL, connect.Direct(true))
 	if err != nil {
 		stdlog.Fatalln("Error: connect to mongodb-node:", err)
 	}
 	defer sess.Disconnect(context.Background()) //nolint:errcheck
 
-	client, err := connect.Connect(context.Background(), mURL, &connect.ConnectOptions{AppName: "pbm-speed-test"})
+	client, err := connect.Connect(context.Background(), mURL, "pbm-speed-test")
 	if err != nil {
 		stdlog.Fatalln("Error: connect to mongodb-pbm:", err)
 	}
diff --git a/cmd/pbm/main.go b/cmd/pbm/main.go
@@ -382,7 +382,7 @@ func main() {
 	// we don't need pbm connection if it is `pbm describe-restore -c ...`
 	// or `pbm restore-finish `
 	if describeRestoreOpts.cfg == "" && finishRestore.cfg == "" {
-		conn, err = connect.Connect(ctx, *mURL, &connect.ConnectOptions{AppName: "pbm-ctl"})
+		conn, err = connect.Connect(ctx, *mURL, "pbm-ctl")
 		if err != nil {
 			exitErr(errors.Wrap(err, "connect to mongodb"), pbmOutF)
 		}
diff --git a/cmd/pbm/status.go b/cmd/pbm/status.go
@@ -13,7 +13,6 @@ import (
 
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/percona/percona-backup-mongodb/pbm/backup"
@@ -329,7 +328,7 @@ func directConnect(ctx context.Context, uri, hosts string) (*mongo.Client, error
 	curi.RawQuery = query.Encode()
 	curi.Host = host
 
-	conn, err := mongo.Connect(ctx, options.Client().ApplyURI(curi.String()).SetAppName("pbm-status"))
+	conn, err := connect.MongoConnect(ctx, curi.String(), connect.AppName("pbm-status"))
 	if err != nil {
 		return nil, errors.Wrap(err, "connect")
 	}
diff --git a/e2e-tests/cmd/ensure-oplog/main.go b/e2e-tests/cmd/ensure-oplog/main.go
@@ -78,7 +78,7 @@ const (
 )
 
 func connTopo(ctx context.Context, uri string) (topo, error) {
-	m, err := connect.MongoConnect(ctx, uri, nil)
+	m, err := connect.MongoConnect(ctx, uri)
 	if err != nil {
 		return topoUnknown, errors.Wrap(err, "connect")
 	}
@@ -162,7 +162,7 @@ func ensureClusterOplog(ctx context.Context, uri string, from, till primitive.Ti
 	logger.Printf("[%s] ensuring cluster oplog: %s - %s",
 		uri, formatTimestamp(from), formatTimestamp(from))
 
-	m, err := connect.MongoConnect(ctx, uri, nil)
+	m, err := connect.MongoConnect(ctx, uri)
 	if err != nil {
 		return errors.Wrap(err, "connect")
 	}
@@ -202,7 +202,7 @@ func ensureReplsetOplog(ctx context.Context, uri string, from, till primitive.Ti
 	logger.Printf("[%s] ensure replset oplog: %s - %s",
 		uri, formatTimestamp(from), formatTimestamp(from))
 
-	m, err := connect.MongoConnect(ctx, uri, nil)
+	m, err := connect.MongoConnect(ctx, uri)
 	if err != nil {
 		return errors.Wrap(err, "connect")
 	}
@@ -228,7 +228,7 @@ func ensureReplsetOplog(ctx context.Context, uri string, from, till primitive.Ti
 	logger.Printf("[%s] ensuring replset oplog (actual): %s - %s",
 		uri, formatTimestamp(firstOpT), formatTimestamp(lastOpT))
 
-	conn, err := connect.Connect(ctx, uri, &connect.ConnectOptions{AppName: "ensure-oplog"})
+	conn, err := connect.Connect(ctx, uri, "ensure-oplog")
 	if err != nil {
 		return errors.Wrap(err, "connect to PBM")
 	}
diff --git a/e2e-tests/pkg/pbm/mongo_pbm.go b/e2e-tests/pkg/pbm/mongo_pbm.go
@@ -25,7 +25,7 @@ type MongoPBM struct {
 }
 
 func NewMongoPBM(ctx context.Context, connectionURI string) (*MongoPBM, error) {
-	conn, err := connect.Connect(ctx, connectionURI, &connect.ConnectOptions{AppName: "e2e-tests-pbm"})
+	conn, err := connect.Connect(ctx, connectionURI, "e2e-tests-pbm")
 	if err != nil {
 		return nil, errors.Wrap(err, "connect")
 	}
diff --git a/e2e-tests/pkg/pbm/mongod.go b/e2e-tests/pkg/pbm/mongod.go
@@ -22,7 +22,7 @@ type Mongo struct {
 }
 
 func NewMongo(ctx context.Context, connectionURI string) (*Mongo, error) {
-	cn, err := connect.MongoConnect(ctx, connectionURI, &connect.MongoConnectOptions{AppName: "e2e-tests"})
+	cn, err := connect.MongoConnect(ctx, connectionURI, connect.AppName("e2e-tests"))
 	if err != nil {
 		return nil, errors.Wrap(err, "connect")
 	}
diff --git a/e2e-tests/pkg/tests/state.go b/e2e-tests/pkg/tests/state.go
@@ -162,7 +162,7 @@ func ClusterState(ctx context.Context, mongos *mongo.Client, creds *Credentials)
 		uri = "mongodb://" + uri
 
 		eg.Go(func() error {
-			m, err := connect.Connect(ctx, uri, nil)
+			m, err := connect.Connect(ctx, uri, "pbm")
 			if err != nil {
 				return errors.Wrapf(err, "connect: %q", uri)
 			}
diff --git a/pbm/backup/logical.go b/pbm/backup/logical.go
@@ -216,7 +216,7 @@ func (b *Backup) doLogical(
 }
 
 func dropTMPcoll(ctx context.Context, uri string) error {
-	m, err := connect.MongoConnect(ctx, uri, nil)
+	m, err := connect.MongoConnect(ctx, uri)
 	if err != nil {
 		return errors.Wrap(err, "connect to primary")
 	}
@@ -251,7 +251,7 @@ func waitForWrite(ctx context.Context, m *mongo.Client, ts primitive.Timestamp)
 
 //nolint:nonamedreturns
 func copyUsersNRolles(ctx context.Context, uri string) (lastWrite primitive.Timestamp, err error) {
-	cn, err := connect.MongoConnect(ctx, uri, nil)
+	cn, err := connect.MongoConnect(ctx, uri)
 	if err != nil {
 		return lastWrite, errors.Wrap(err, "connect to primary")
 	}
diff --git a/pbm/connect/connect.go b/pbm/connect/connect.go
@@ -15,30 +15,67 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 )
 
-type ConnectOptions struct {
-	AppName string
+type MongoOption func(*options.ClientOptions) error
+
+func AppName(name string) MongoOption {
+	return func(opts *options.ClientOptions) error {
+		if len(name) == 0 {
+			return errors.New("AppName is not specified")
+		}
+		opts.SetAppName(name)
+		return nil
+	}
 }
 
-type MongoConnectOptions struct {
-	AppName string
-	Direct  bool
+func Direct(direct bool) MongoOption {
+	return func(opts *options.ClientOptions) error {
+		opts.SetDirect(direct)
+		return nil
+	}
 }
 
-func MongoConnect(ctx context.Context, uri string, opts *MongoConnectOptions) (*mongo.Client, error) {
-	if !strings.HasPrefix(uri, "mongodb://") {
-		uri = "mongodb://" + uri
+func ReadConcern(readConcern *readconcern.ReadConcern) MongoOption {
+	return func(opts *options.ClientOptions) error {
+		if readConcern == nil {
+			return errors.New("ReadConcern not specified")
+		}
+		opts.SetReadConcern(readConcern)
+		return nil
 	}
+}
 
-	if opts == nil {
-		opts = &MongoConnectOptions{}
+func NoRS() MongoOption {
+	return func(opts *options.ClientOptions) error {
+		opts.SetReplicaSet("")
+		return nil
 	}
+}
 
-	mopts := options.Client().ApplyURI(uri).
-		SetAppName(opts.AppName).
+func MongoConnect(ctx context.Context, uri string, mongoOptions ...MongoOption) (*mongo.Client, error) {
+	if !strings.HasPrefix(uri, "mongodb://") {
+		uri = "mongodb://" + uri
+	}
+
+	// default options
+	mopts := options.Client().
+		SetAppName("pbm").
 		SetReadPreference(readpref.Primary()).
 		SetReadConcern(readconcern.Majority()).
 		SetWriteConcern(writeconcern.Majority()).
-		SetDirect(opts.Direct)
+		SetDirect(false)
+
+	// apply and override using end-user options from conn string
+	mopts.ApplyURI(uri)
+
+	// override with explicit options from the code
+	for _, opt := range mongoOptions {
+		if opt != nil {
+			if err := opt(mopts); err != nil {
+				return nil, errors.Wrap(err, "invalid mongo option")
+			}
+		}
+	}
+
 	conn, err := mongo.Connect(ctx, mopts)
 	if err != nil {
 		return nil, errors.Wrap(err, "connect")
@@ -60,13 +97,11 @@ func UnsafeClient(m *mongo.Client) Client {
 	return &clientImpl{m}
 }
 
-func Connect(ctx context.Context, uri string, opts *ConnectOptions) (Client, error) {
-	if opts == nil {
-		opts = &ConnectOptions{}
-	}
-
-	mopts := &MongoConnectOptions{AppName: opts.AppName}
-	client, err := MongoConnect(ctx, uri, mopts)
+// Connect resolves MongoDB connection to Primary member and wraps it within Client object.
+// In case of replica set it returns connection to Primary member,
+// while in case of sharded cluster it returns connection to Config RS Primary member.
+func Connect(ctx context.Context, uri, appName string) (Client, error) {
+	client, err := MongoConnect(ctx, uri, AppName(appName))
 	if err != nil {
 		return nil, errors.Wrap(err, "create mongo connection")
 	}
@@ -110,11 +145,8 @@ func Connect(ctx context.Context, uri string, opts *ConnectOptions) (Client, err
 
 	// Preserving the `replicaSet` parameter will cause an error
 	// while connecting to the ConfigServer (mismatched replicaset names)
-	query := curi.Query()
-	query.Del("replicaSet")
-	curi.RawQuery = query.Encode()
 	curi.Host = chost[1]
-	client, err = MongoConnect(ctx, curi.String(), mopts)
+	client, err = MongoConnect(ctx, curi.String(), AppName(appName), NoRS())
 	if err != nil {
 		return nil, errors.Wrapf(err, "create mongo connection to configsvr with connection string '%s'", curi)
 	}
diff --git a/sdk/sdk.go b/sdk/sdk.go
@@ -129,7 +129,7 @@ type Client interface {
 }
 
 func NewClient(ctx context.Context, uri string) (Client, error) {
-	conn, err := connect.Connect(ctx, uri, nil)
+	conn, err := connect.Connect(ctx, uri, "sdk")
 	if err != nil {
 		return nil, err
 	}

Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ func runAgent(mongoURI string, dumpConns int) error {`
`86`	`86`	`ctx, cancel := context.WithCancel(context.Background())`
`87`	`87`	`defer cancel()`
`88`	`88`
`89`		`- leadConn, err := connect.Connect(ctx, mongoURI, &connect.ConnectOptions{AppName: "pbm-agent"})`
	`89`	`+ leadConn, err := connect.Connect(ctx, mongoURI, "pbm-agent")`
`90`	`90`	`if err != nil {`
`91`	`91`	`return errors.Wrap(err, "connect to PBM")`
`92`	`92`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ func testCompression(mURL string, compression compress.CompressionType, level *i`
`86`	`86`	`var cn *mongo.Client`
`87`	`87`
`88`	`88`	`if collection != "" {`
`89`		`- cn, err := connect.MongoConnect(ctx, mURL, &connect.MongoConnectOptions{Direct: true})`
	`89`	`+ cn, err := connect.MongoConnect(ctx, mURL, connect.Direct(true))`
`90`	`90`	`if err != nil {`
`91`	`91`	`stdlog.Fatalln("Error: connect to mongodb-node:", err)`
`92`	`92`	`}`
`@@ -108,13 +108,13 @@ func testCompression(mURL string, compression compress.CompressionType, level *i`
`108`	`108`	`}`
`109`	`109`
`110`	`110`	`func testStorage(mURL string, compression compress.CompressionType, level *int, sizeGb float64, collection string) {`
`111`		`- sess, err := connect.MongoConnect(context.Background(), mURL, &connect.MongoConnectOptions{Direct: true})`
	`111`	`+ sess, err := connect.MongoConnect(context.Background(), mURL, connect.Direct(true))`
`112`	`112`	`if err != nil {`
`113`	`113`	`stdlog.Fatalln("Error: connect to mongodb-node:", err)`
`114`	`114`	`}`
`115`	`115`	`defer sess.Disconnect(context.Background()) //nolint:errcheck`
`116`	`116`
`117`		`- client, err := connect.Connect(context.Background(), mURL, &connect.ConnectOptions{AppName: "pbm-speed-test"})`
	`117`	`+ client, err := connect.Connect(context.Background(), mURL, "pbm-speed-test")`
`118`	`118`	`if err != nil {`
`119`	`119`	`stdlog.Fatalln("Error: connect to mongodb-pbm:", err)`
`120`	`120`	`}`
Original file line number	Diff line number	Diff line change
`@@ -382,7 +382,7 @@ func main() {`
`382`	`382`	// we don't need pbm connection if it is `pbm describe-restore -c ...`
`383`	`383`	// or `pbm restore-finish `
`384`	`384`	`if describeRestoreOpts.cfg == "" && finishRestore.cfg == "" {`
`385`		`- conn, err = connect.Connect(ctx, *mURL, &connect.ConnectOptions{AppName: "pbm-ctl"})`
	`385`	`+ conn, err = connect.Connect(ctx, *mURL, "pbm-ctl")`
`386`	`386`	`if err != nil {`
`387`	`387`	`exitErr(errors.Wrap(err, "connect to mongodb"), pbmOutF)`
`388`	`388`	`}`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,7 @@ const (`
`78`	`78`	`)`
`79`	`79`
`80`	`80`	`func connTopo(ctx context.Context, uri string) (topo, error) {`
`81`		`- m, err := connect.MongoConnect(ctx, uri, nil)`
	`81`	`+ m, err := connect.MongoConnect(ctx, uri)`
`82`	`82`	`if err != nil {`
`83`	`83`	`return topoUnknown, errors.Wrap(err, "connect")`
`84`	`84`	`}`
`@@ -162,7 +162,7 @@ func ensureClusterOplog(ctx context.Context, uri string, from, till primitive.Ti`
`162`	`162`	`logger.Printf("[%s] ensuring cluster oplog: %s - %s",`
`163`	`163`	`uri, formatTimestamp(from), formatTimestamp(from))`
`164`	`164`
`165`		`- m, err := connect.MongoConnect(ctx, uri, nil)`
	`165`	`+ m, err := connect.MongoConnect(ctx, uri)`
`166`	`166`	`if err != nil {`
`167`	`167`	`return errors.Wrap(err, "connect")`
`168`	`168`	`}`
`@@ -202,7 +202,7 @@ func ensureReplsetOplog(ctx context.Context, uri string, from, till primitive.Ti`
`202`	`202`	`logger.Printf("[%s] ensure replset oplog: %s - %s",`
`203`	`203`	`uri, formatTimestamp(from), formatTimestamp(from))`
`204`	`204`
`205`		`- m, err := connect.MongoConnect(ctx, uri, nil)`
	`205`	`+ m, err := connect.MongoConnect(ctx, uri)`
`206`	`206`	`if err != nil {`
`207`	`207`	`return errors.Wrap(err, "connect")`
`208`	`208`	`}`
`@@ -228,7 +228,7 @@ func ensureReplsetOplog(ctx context.Context, uri string, from, till primitive.Ti`
`228`	`228`	`logger.Printf("[%s] ensuring replset oplog (actual): %s - %s",`
`229`	`229`	`uri, formatTimestamp(firstOpT), formatTimestamp(lastOpT))`
`230`	`230`
`231`		`- conn, err := connect.Connect(ctx, uri, &connect.ConnectOptions{AppName: "ensure-oplog"})`
	`231`	`+ conn, err := connect.Connect(ctx, uri, "ensure-oplog")`
`232`	`232`	`if err != nil {`
`233`	`233`	`return errors.Wrap(err, "connect to PBM")`
`234`	`234`	`}`
Original file line number	Diff line number	Diff line change
`@@ -25,7 +25,7 @@ type MongoPBM struct {`
`25`	`25`	`}`
`26`	`26`
`27`	`27`	`func NewMongoPBM(ctx context.Context, connectionURI string) (*MongoPBM, error) {`
`28`		`- conn, err := connect.Connect(ctx, connectionURI, &connect.ConnectOptions{AppName: "e2e-tests-pbm"})`
	`28`	`+ conn, err := connect.Connect(ctx, connectionURI, "e2e-tests-pbm")`
`29`	`29`	`if err != nil {`
`30`	`30`	`return nil, errors.Wrap(err, "connect")`
`31`	`31`	`}`
Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,7 @@ type Mongo struct {`
`22`	`22`	`}`
`23`	`23`
`24`	`24`	`func NewMongo(ctx context.Context, connectionURI string) (*Mongo, error) {`
`25`		`- cn, err := connect.MongoConnect(ctx, connectionURI, &connect.MongoConnectOptions{AppName: "e2e-tests"})`
	`25`	`+ cn, err := connect.MongoConnect(ctx, connectionURI, connect.AppName("e2e-tests"))`
`26`	`26`	`if err != nil {`
`27`	`27`	`return nil, errors.Wrap(err, "connect")`
`28`	`28`	`}`
Original file line number	Diff line number	Diff line change
`@@ -162,7 +162,7 @@ func ClusterState(ctx context.Context, mongos mongo.Client, creds Credentials)`
`162`	`162`	`uri = "mongodb://" + uri`
`163`	`163`
`164`	`164`	`eg.Go(func() error {`
`165`		`- m, err := connect.Connect(ctx, uri, nil)`
	`165`	`+ m, err := connect.Connect(ctx, uri, "pbm")`
`166`	`166`	`if err != nil {`
`167`	`167`	`return errors.Wrapf(err, "connect: %q", uri)`
`168`	`168`	`}`
Original file line number	Diff line number	Diff line change
`@@ -216,7 +216,7 @@ func (b *Backup) doLogical(`
`216`	`216`	`}`
`217`	`217`
`218`	`218`	`func dropTMPcoll(ctx context.Context, uri string) error {`
`219`		`- m, err := connect.MongoConnect(ctx, uri, nil)`
	`219`	`+ m, err := connect.MongoConnect(ctx, uri)`
`220`	`220`	`if err != nil {`
`221`	`221`	`return errors.Wrap(err, "connect to primary")`
`222`	`222`	`}`
`@@ -251,7 +251,7 @@ func waitForWrite(ctx context.Context, m *mongo.Client, ts primitive.Timestamp)`
`251`	`251`
`252`	`252`	`//nolint:nonamedreturns`
`253`	`253`	`func copyUsersNRolles(ctx context.Context, uri string) (lastWrite primitive.Timestamp, err error) {`
`254`		`- cn, err := connect.MongoConnect(ctx, uri, nil)`
	`254`	`+ cn, err := connect.MongoConnect(ctx, uri)`
`255`	`255`	`if err != nil {`
`256`	`256`	`return lastWrite, errors.Wrap(err, "connect to primary")`
`257`	`257`	`}`