PBM-799: Make WriteConcern option configurable (#925)

* Add WriteConcern option
* Add ConnectTimeout & ServerSelectionTimeout option
* Reuse MongoConnect within physical restore
* Use single node write concern for standalone mode

Author: boris-ilijic

pbm/connect/connect.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/url"
 	"strings"
+	"time"
 
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
@@ -34,6 +35,9 @@ func Direct(direct bool) MongoOption {
 	}
 }
 
+// ReadConcern option sets availability guarantees for read operation.
+// For PBM typically use: [readconcern.Local] or [readconcern.Majority].
+// If the option is not specified the default is: [readconcern.Majority].
 func ReadConcern(readConcern *readconcern.ReadConcern) MongoOption {
 	return func(opts *options.ClientOptions) error {
 		if readConcern == nil {
@@ -44,13 +48,41 @@ func ReadConcern(readConcern *readconcern.ReadConcern) MongoOption {
 	}
 }
 
+// WriteConcern option sets level of acknowledgment for write operation.
+// For PBM typically use: [writeconcern.W1] or [writeconcern.Majority].
+// If the option is not specified the default is: [writeconcern.Majority].
+func WriteConcern(writeConcern *writeconcern.WriteConcern) MongoOption {
+	return func(opts *options.ClientOptions) error {
+		if writeConcern == nil {
+			return errors.New("WriteConcern not specified")
+		}
+		opts.SetWriteConcern(writeConcern)
+		return nil
+	}
+}
+
+// NoRS option removes replica set name setting
 func NoRS() MongoOption {
 	return func(opts *options.ClientOptions) error {
 		opts.SetReplicaSet("")
 		return nil
 	}
 }
 
+func ConnectTimeout(d time.Duration) MongoOption {
+	return func(opts *options.ClientOptions) error {
+		opts.SetConnectTimeout(d)
+		return nil
+	}
+}
+
+func ServerSelectionTimeout(d time.Duration) MongoOption {
+	return func(opts *options.ClientOptions) error {
+		opts.SetServerSelectionTimeout(d)
+		return nil
+	}
+}
+
 func MongoConnect(ctx context.Context, uri string, mongoOptions ...MongoOption) (*mongo.Client, error) {
 	if !strings.HasPrefix(uri, "mongodb://") {
 		uri = "mongodb://" + uri

pbm/restore/physical.go

@@ -23,6 +23,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 	"golang.org/x/mod/semver"
 	"gopkg.in/yaml.v2"
 
@@ -1669,8 +1670,15 @@ func tryConn(port int, logpath string) (*mongo.Client, error) {
 
 	var cn *mongo.Client
 	var err error
+	host := fmt.Sprintf("mongodb://localhost:%d", port)
 	for i := 0; i < tryConnCount; i++ {
-		cn, err = conn(port, tryConnTimeout)
+		cn, err = connect.MongoConnect(context.Background(), host,
+			connect.AppName("pbm-physical-restore"),
+			connect.Direct(true),
+			connect.WriteConcern(writeconcern.W1()),
+			connect.ConnectTimeout(time.Second*120),
+			connect.ServerSelectionTimeout(tryConnTimeout),
+		)
 		if err == nil {
 			return cn, nil
 		}
@@ -1698,29 +1706,6 @@ func tryConn(port int, logpath string) (*mongo.Client, error) {
 	return nil, errors.Errorf("failed to  connect after %d tries: %v", tryConnCount, err)
 }
 
-func conn(port int, tout time.Duration) (*mongo.Client, error) {
-	ctx := context.Background()
-
-	opts := options.Client().
-		SetHosts([]string{"localhost:" + strconv.Itoa(port)}).
-		SetAppName("pbm-physical-restore").
-		SetDirect(true).
-		SetConnectTimeout(time.Second * 120).
-		SetServerSelectionTimeout(tout)
-
-	conn, err := mongo.Connect(ctx, opts)
-	if err != nil {
-		return nil, errors.Wrap(err, "connect")
-	}
-
-	err = conn.Ping(ctx, nil)
-	if err != nil {
-		return nil, errors.Wrap(err, "ping")
-	}
-
-	return conn, nil
-}
-
 const internalMongodLog = "pbm.restore.log"
 
 func (r *PhysRestore) startMongo(opts ...string) error {