PBM-1426: Sporadic failures on physical restore with PITR (#1061)

Ensure that oplog is applied on Primary node during Physical Restore

When applying oplog chunks on the the Primary node, it might happen that
Primary node is still not reached Primary state, simply because after its
restart, it is in transition towards Primary:
STARTUP -> STARTUP2 -> RECOVERING -> SECONDARY -> PRIMARY

To fix such a situation, PBM checks if the Primary state is reached. After
the Primay state is reached, oplog is applied on that Primary.

Author: boris-ilijic

pbm/restore/physical.go

@@ -324,6 +324,39 @@ func waitMgoShutdown(dbpath string) error {
 	return nil
 }
 
+// waitToBecomePrimary pause execution until RS member becomes primary node.
+// Error is returned in case of timeout.
+// Unexpected error while getting node info is just logged.
+func (r *PhysRestore) waitToBecomePrimary(ctx context.Context, m *mongo.Client) error {
+	tk := time.NewTicker(time.Second)
+	defer tk.Stop()
+
+	tout := time.NewTimer(2 * time.Minute)
+	defer tout.Stop()
+
+	for {
+		select {
+		case <-tk.C:
+			inf, err := topo.GetNodeInfo(ctx, m)
+			if err != nil {
+				r.log.Debug("get node info error while waiting to become primary: %v", err)
+				continue
+			}
+
+			if inf.IsPrimary {
+				return nil
+			}
+			r.log.Debug("node: %s is still not primary, waiting for another cycle", inf.Me)
+
+		case <-tout.C:
+			return errors.New("timeout while waiting the node to become primary")
+
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
 // toState moves cluster to the given restore state.
 // All communication happens via files in the restore dir on storage.
 //
@@ -1347,6 +1380,11 @@ func (r *PhysRestore) replayOplog(
 		return errors.Wrap(err, "define mongo version")
 	}
 
+	err = r.waitToBecomePrimary(ctx, nodeConn)
+	if err != nil {
+		return errors.Wrap(err, "wait to become primary before applying oplog")
+	}
+
 	oplogOption := applyOplogOption{
 		start:  &from,
 		end:    &to,