PBM-1311: PBM fails to do PITR slicing when RS doesn't have a majority (#940)

* Expand RS status with WriteMajorityCount field
* Add topo.IsWriteMajorityRequested to determine if majority is requested in current setup
* Fix PITR slicing intervals for the case when cluster doesn't have majority

Author: boris-ilijic

pbm/backup/logical.go

@@ -89,6 +89,7 @@ func (b *Backup) doLogical(
 
 	stopOplogSlicer := startOplogSlicer(ctx,
 		b.nodeConn,
+		b.leadConn.MongoOptions().WriteConcern,
 		b.SlicerInterval(),
 		rsMeta.FirstWriteTS,
 		func(ctx context.Context, w io.WriterTo, from, till primitive.Timestamp) (int64, error) {

pbm/backup/slicer.go

@@ -9,6 +9,7 @@ import (
 
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
@@ -22,6 +23,7 @@ type stopSlicerFunc func() (primitive.Timestamp, int64, error)
 func startOplogSlicer(
 	ctx context.Context,
 	m *mongo.Client,
+	writeConcern *writeconcern.WriteConcern,
 	interval time.Duration,
 	startOpTime primitive.Timestamp,
 	upload uploadChunkFunc,
@@ -53,7 +55,11 @@ func startOplogSlicer(
 				return
 			}
 
-			currOpTime, err := topo.GetLastWrite(ctx, m, true)
+			majority, err := topo.IsWriteMajorityRequested(ctx, m, writeConcern)
+			if err != nil {
+				l.Error("failed to inspect requested majority: %v", err)
+			}
+			currOpTime, err := topo.GetLastWrite(ctx, m, majority)
 			if err != nil {
 				if errors.Is(err, context.Canceled) {
 					return

pbm/slicer/slicer.go

@@ -392,7 +392,11 @@ func (s *Slicer) Stream(
 			return OpMovedError{ld.Node}
 		}
 		if sliceTo.IsZero() {
-			sliceTo, err = topo.GetLastWrite(ctx, s.node, true)
+			majority, err := topo.IsWriteMajorityRequested(ctx, s.node, s.leadClient.MongoOptions().WriteConcern)
+			if err != nil {
+				return errors.Wrap(err, "define requested majority")
+			}
+			sliceTo, err = topo.GetLastWrite(ctx, s.node, majority)
 			if err != nil {
 				return errors.Wrap(err, "define last write timestamp")
 			}

pbm/topo/cluster.go

@@ -7,6 +7,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 
 	"github.com/percona/percona-backup-mongodb/pbm/connect"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
@@ -66,6 +67,32 @@ func OpTimeFromNodeInfo(inf *NodeInfo, majority bool) (primitive.Timestamp, erro
 	return lw, nil
 }
 
+// IsWriteMajorityRequested compares cluster wide majority (replSetGetStatus.writeMajorityCount)
+// with WriteConcern requested in connection string and determinates if majority is requested or not
+func IsWriteMajorityRequested(
+	ctx context.Context,
+	m *mongo.Client,
+	writeConcern *writeconcern.WriteConcern,
+) (bool, error) {
+	if writeConcern == nil ||
+		!writeConcern.IsValid() ||
+		writeConcern == writeconcern.Majority() {
+		return true, nil
+	}
+
+	w, ok := writeConcern.W.(int)
+	if !ok {
+		return true, nil
+	}
+
+	s, err := GetReplsetStatus(ctx, m)
+	if err != nil {
+		return true, errors.Wrap(err, "get replset status")
+	}
+
+	return w >= s.WriteMajorityCount, nil
+}
+
 // ClusterMembers returns list of replicasets current cluster consists of
 // (shards + configserver). The list would consist of on rs if cluster is
 // a non-sharded rs.

pbm/topo/status.go

@@ -54,6 +54,7 @@ type ReplsetStatus struct {
 	ClusterTime             *ClusterTime         `bson:"$clusterTime,omitempty" json:"$clusterTime,omitempty"`
 	ConfigServerState       *ConfigServerState   `bson:"$configServerState,omitempty" json:"$configServerState,omitempty"`
 	OperationTime           *primitive.Timestamp `bson:"operationTime,omitempty" json:"operationTime,omitempty"`
+	WriteMajorityCount      int                  `bson:"writeMajorityCount,omitempty" json:"writeMajorityCount,omitempty"`
 }
 
 func CurrentUser(ctx context.Context, m *mongo.Client) (*AuthInfo, error) {