[PBM-1223] backup oplog from the oldest active transaction timestamp (#900)

Author: defbin

pbm/backup/backup.go

@@ -16,6 +16,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/lock"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/oplog"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -174,7 +175,7 @@ func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l
 		return errors.Wrap(err, "get cluster info")
 	}
 
-	oplogTS, err := topo.OpTimeFromNodeInfo(inf, true)
+	oplogTS, err := oplog.GetOplogStartTime(ctx, b.nodeConn)
 	if err != nil {
 		return errors.Wrap(err, "define oplog start position")
 	}
@@ -656,7 +657,7 @@ func (b *Backup) setClusterFirstWrite(ctx context.Context, bcpName string) error
 			return errors.New("no replset metadata")
 		}
 
-		if condAll(bcp.Replsets, func(br *BackupReplset) bool { return !br.FirstWriteTS.IsZero() }) {
+		if condAll(bcp.Replsets, func(br *BackupReplset) bool { return br.FirstWriteTS.T > 1 }) {
 			break
 		}
 

pbm/oplog/oplog.go

@@ -0,0 +1,70 @@
+package oplog
+
+import (
+	"context"
+
+	"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/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
+
+	"github.com/percona/percona-backup-mongodb/pbm/errors"
+)
+
+var errNoTransaction = errors.New("no transaction found")
+
+func GetOplogStartTime(ctx context.Context, m *mongo.Client) (primitive.Timestamp, error) {
+	ts, err := findTransactionStartTime(ctx, m)
+	if errors.Is(err, errNoTransaction) {
+		ts, err = findLastOplogTS(ctx, m)
+	}
+
+	return ts, err
+}
+
+func findTransactionStartTime(ctx context.Context, m *mongo.Client) (primitive.Timestamp, error) {
+	coll := m.Database("config").Collection("transactions", options.Collection().SetReadConcern(readconcern.Local()))
+	f := bson.D{{"state", bson.D{{"$in", bson.A{"prepared", "inProgress"}}}}}
+	o := options.FindOne().SetSort(bson.D{{"startOpTime", 1}})
+	doc, err := coll.FindOne(ctx, f, o).Raw()
+	if err != nil {
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			return primitive.Timestamp{}, errNoTransaction
+		}
+		return primitive.Timestamp{}, errors.Wrap(err, "query transactions")
+	}
+
+	rawTS, err := doc.LookupErr("startOpTime", "ts")
+	if err != nil {
+		return primitive.Timestamp{}, errors.Wrap(err, "lookup timestamp")
+	}
+
+	t, i, ok := rawTS.TimestampOK()
+	if !ok {
+		return primitive.Timestamp{}, errors.Wrap(err, "parse timestamp")
+	}
+
+	return primitive.Timestamp{T: t, I: i}, nil
+}
+
+func findLastOplogTS(ctx context.Context, m *mongo.Client) (primitive.Timestamp, error) {
+	coll := m.Database("local").Collection("oplog.rs")
+	o := options.FindOne().SetSort(bson.M{"$natural": -1})
+	doc, err := coll.FindOne(ctx, bson.D{}, o).Raw()
+	if err != nil {
+		return primitive.Timestamp{}, errors.Wrap(err, "query oplog")
+	}
+
+	rawTS, err := doc.LookupErr("ts")
+	if err != nil {
+		return primitive.Timestamp{}, errors.Wrap(err, "lookup oplog ts")
+	}
+
+	t, i, ok := rawTS.TimestampOK()
+	if !ok {
+		return primitive.Timestamp{}, errors.Wrap(err, "parse oplog ts")
+	}
+
+	return primitive.Timestamp{T: t, I: i}, nil
+}

pbm/oplog/restore.go

@@ -574,7 +574,7 @@ func bsonDocToOplog(doc bson.D) (*db.Oplog, error) {
 func (o *OplogRestore) applyTxn(id string) error {
 	t, ok := o.txnData[id]
 	if !ok {
-		return errors.Errorf("unknown transaction id %s", id)
+		return nil
 	}
 
 	for _, op := range t.applyOps {