[flakiness] improve flakiness and perf of multi_cluster restore test (#37)

# Summary

this improves our third most flaky test (not 100% confident) -
[HC](REDACTED)

- some changes:
  - fail early if primary is not electable (no primary -> no backup)
- use proper pit restore time instead of randomly choose a date, this
sometimes led to errors like these - which won't happen anymore - it
took around 40s now its immediately done;
```
[2025/03/03 13:24:01.103] Restoring back to the moment 15 seconds ago (millis): 1741004626102.5642
[2025/03/03 13:24:01.233] Encountered exception: Error sending request to Ops Manager API. 409 ({"detail":"Restore process initiation failed: The requested restore could not be started: Invalid restore point: Are you sure your backups were running at the time you selected?.","error":409,"errorCode":"RESTORE_INITIATION_FAILED","parameters":["The requested restore could not be started: Invalid restore point: Are you sure your backups were running at the time you selected?"],"reason":"Conflict"}).

[2025/03/03 13:24:01.233]  Request details: post https://om-backup.a-1741002799-ftz7steaq7z.interconnected:8443/api/public/v1.0/groups/67c59c69bb720856d1c0c3e1/clusters/67c59cb0bb720856d1c0cdce/restoreJobs (data: {'delivery': {'methodName': 'AUTOMATED_RESTORE', 'targetGroupId': '67c59c69bb720856d1c0c3e1', 'targetClusterId': '67c59cb0bb720856d1c0cdce'}, 'pointInTimeUTCMillis': 1741004626102.5642}) on retry number 0
```

## Proof of Work


https://spruce.mongodb.com/task/ops_manager_kubernetes_e2e_multi_cluster_kind_e2e_multi_cluster_backup_restore_patch_b17acb360b9b2d9a598fe564ee2ced8f99dd7634_67c582286041a800070f3a6a_25_03_03_10_19_28/logs?execution=1&page=0&sorts=STATUS%3AASC%3BDURATION%3ADESC

https://spruce.mongodb.com/task/ops_manager_kubernetes_e2e_multi_cluster_kind_e2e_multi_cluster_backup_restore_patch_b17acb360b9b2d9a598fe564ee2ced8f99dd7634_67c85db031d4ae0007515f56_25_03_05_14_20_34/logs?execution=0&sortBy=STATUS&sortDir=ASC


## Checklist
- [ ] Have you linked a jira ticket and/or is the ticket in the title?
- [x] Have you checked whether your jira ticket required DOCSP changes?
- [x] Have you checked for release_note changes?

## Reminder (Please remove this when merging)
- Please try to Approve or Reject Changes the PR, keep PRs in review as
short as possible
- Our Short Guide for PRs:
[Link](https://docs.google.com/document/d/1T93KUtdvONq43vfTfUt8l92uo4e4SEEvFbIEKOxGr44/edit?tab=t.0)
- Remember the following Communication Standards - use comment prefixes
for clarity:
  * **blocking**: Must be addressed before approval.
  * **follow-up**: Can be addressed in a later PR or ticket.
  * **q**: Clarifying question.
  * **nit**: Non-blocking suggestions.
  * **note**: Side-note, non-actionable. Example: Praise 
  * --> no prefix is considered a question

---------

Co-authored-by: Nam Nguyen <[email protected]>

Author: lucian-tosa

docker/mongodb-enterprise-tests/kubetester/omtester.py

@@ -106,6 +106,7 @@ def __init__(self, om_context: OMContext):
             os.environ["OM_USER"] = self.context.user
         if self.context.base_url:
             os.environ["OM_HOST"] = self.context.base_url
+        self.latest_backup_completion_time = None
 
     def ensure_group_id(self):
         if self.context.project_id is None:
@@ -124,6 +125,12 @@ def create_restore_job_snapshot(self, snapshot_id: Optional[str] = None) -> str:
 
         return self.api_create_restore_job_from_snapshot(cluster_id, snapshot_id)["id"]
 
+    def set_latest_backup_completion_time(self, timestamp):
+        self.latest_backup_completion_time = timestamp
+
+    def get_latest_backup_completion_time(self):
+        return self.latest_backup_completion_time or 0
+
     def create_restore_job_pit(self, pit_milliseconds: int, retry: int = 120):
         """creates a restore job to restore the mongodb cluster to some version specified by the parameter."""
         cluster_id = self.get_backup_cluster_id()
@@ -165,6 +172,13 @@ def wait_until_backup_snapshots_are_ready(
                 print(f"Snapshots are ready, project: {self.context.group_name}, time: {time.time() - start_time} sec")
                 span = trace.get_current_span()
                 span.set_attribute(key="mck.snapshot_time", value=time.time() - start_time)
+                completed_snapshots = [s for s in snapshots if s.get("complete", False)]
+                latest_snapshot = max(completed_snapshots, key=lambda s: s["created"]["date"])
+                snapshot_timestamp = latest_snapshot["created"]["date"]
+                print(f"Current Backup Snapshots: {snapshots}")
+                self.set_latest_backup_completion_time(
+                    time_to_millis(datetime.fromisoformat(snapshot_timestamp.replace("Z", "")))
+                )
                 return
             time.sleep(3)
             timeout -= 3
@@ -804,3 +818,10 @@ def should_include_tag(version: Optional[Dict[str, str]]) -> bool:
         return semver.compare(version_string, feature_controls_enabled_version) < 0
 
     return True
+
+
+def time_to_millis(date_time) -> int:
+    """https://stackoverflow.com/a/11111177/614239"""
+    epoch = datetime.utcfromtimestamp(0)
+    pit_millis = (date_time - epoch).total_seconds() * 1000
+    return pit_millis

docker/mongodb-enterprise-tests/tests/conftest.py

@@ -1662,6 +1662,25 @@ def install_multi_cluster_operator_cluster_scoped(
     )
 
 
+def wait_for_primary(mdb_client, timeout=500):
+    """
+    Waits until a primary node is elected before proceeding with database operations.
+    This prevents unnecessary timeouts during restore actions (they will just fail otherwise).
+    """
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        try:
+            status = mdb_client.admin.command("replSetGetStatus")
+            primary = [m for m in status["members"] if m["stateStr"] == "PRIMARY"]
+            if primary:
+                print(f"✅ Primary node detected: {primary[0]['name']}")
+                return True
+        except Exception as e:
+            print(f"⏳ Waiting for primary election... Error: {e}")
+        time.sleep(5)
+    raise Exception("❌ No primary found within timeout")
+
+
 def assert_data_got_restored(test_data, collection1, collection2=None, timeout=300):
     """The data in the db has been restored to the initial state. Note, that this happens eventually - so
     we need to loop for some time (usually takes 60 seconds max). This is different from restoring from a

docker/mongodb-enterprise-tests/tests/multicluster/multi_cluster_backup_restore.py

@@ -23,9 +23,12 @@
 from kubetester.omtester import OMTester
 from kubetester.operator import Operator
 from kubetester.opsmanager import MongoDBOpsManager
-from pymongo.errors import ServerSelectionTimeoutError
 from pytest import fixture, mark
-from tests.conftest import assert_data_got_restored, update_coredns_hosts
+from tests.conftest import (
+    assert_data_got_restored,
+    update_coredns_hosts,
+    wait_for_primary,
+)
 
 TEST_DATA = {"_id": "unique_id", "name": "John", "address": "Highway 37", "age": 30}
 
@@ -362,13 +365,20 @@ def project_one(
         )
 
     @fixture(scope="function")
-    def mongodb_multi_one_collection(self, mongodb_multi_one: MongoDBMulti):
-        # we instantiate the pymongo client per test to avoid flakiness as the primary and secondary might swap
-        collection = pymongo.MongoClient(
+    def mdb_client(self, mongodb_multi_one: MongoDBMulti):
+        return pymongo.MongoClient(
             mongodb_multi_one.tester(port=MONGODB_PORT).cnx_string,
             **mongodb_multi_one.tester(port=MONGODB_PORT).default_opts,
-        )["testdb"]
-        return collection["testcollection"]
+            readPreference="primary",  # let's read from the primary and not stale data from the secondary
+        )
+
+    @fixture(scope="function")
+    def mongodb_multi_one_collection(self, mdb_client):
+
+        # Ensure primary is available before proceeding
+        wait_for_primary(mdb_client)
+
+        return mdb_client["testdb"]["testcollection"]
 
     @fixture(scope="module")
     def mongodb_multi_one(
@@ -472,6 +482,7 @@ def test_add_test_data(self, mongodb_multi_one_collection):
                 print(e)
                 max_attempts -= 1
                 time.sleep(6)
+        raise Exception("❌ Failed to insert test data after multiple attempts")
 
     @mark.e2e_multi_cluster_backup_restore
     def test_mdb_backed_up(self, project_one: OMTester):
@@ -489,14 +500,17 @@ def test_pit_restore(self, project_one: OMTester):
         now_millis = time_to_millis(datetime.datetime.now())
         print("\nCurrent time (millis): {}".format(now_millis))
 
-        pit_datetme = datetime.datetime.now() - datetime.timedelta(seconds=15)
-        pit_millis = time_to_millis(pit_datetme)
-        print("Restoring back to the moment 15 seconds ago (millis): {}".format(pit_millis))
+        backup_completion_time = project_one.get_latest_backup_completion_time()
+        print("\nbackup_completion_time: {}".format(backup_completion_time))
+
+        pit_millis = backup_completion_time + 1500
+
+        print(f"Restoring back to: {pit_millis}")
 
         project_one.create_restore_job_pit(pit_millis)
 
     @mark.e2e_multi_cluster_backup_restore
-    def test_data_got_restored(self, mongodb_multi_one_collection):
+    def test_data_got_restored(self, mongodb_multi_one_collection, mdb_client):
         assert_data_got_restored(TEST_DATA, mongodb_multi_one_collection, timeout=1200)