SERVER-59712 Create ClusterToClusterFixture in resmoke

2 files changed, 151 insertions, 8 deletions

diff --git a/buildscripts/resmokelib/testing/fixtures/cluster_to_cluster.py b/buildscripts/resmokelib/testing/fixtures/cluster_to_cluster.py
new file mode 100644
index 00000000000..1c5f0540ded
--- /dev/null
+++ b/buildscripts/resmokelib/testing/fixtures/cluster_to_cluster.py
@@ -0,0 +1,139 @@
+"""Fixture with two clusters (for cluster to cluster replications) for executing JSTests against."""
+
+import copy
+import os.path
+
+import buildscripts.resmokelib.testing.fixtures.interface as interface
+
+
+class ClusterToClusterFixture(interface.Fixture):  # pylint: disable=too-many-instance-attributes
+    """Fixture which provides two clusters to perform a cluster to cluster replication."""
+
+    def __init__(  # pylint: disable=too-many-arguments,too-many-locals
+            self, logger, job_num, fixturelib, cluster0_options, cluster1_options,
+            dbpath_prefix=None, preserve_dbpath=False):
+        """Initialize with different options for the clusters."""
+
+        interface.Fixture.__init__(self, logger, job_num, fixturelib, dbpath_prefix=dbpath_prefix)
+
+        self.clusters = []
+        self.both_cluster_options = []
+        parsed_options = [
+            self.fixturelib.default_if_none(copy.deepcopy(cluster0_options), {}),
+            self.fixturelib.default_if_none(copy.deepcopy(cluster1_options), {})
+        ]
+
+        self.preserve_dbpath = preserve_dbpath
+
+        # The usual command line params can lead to messy behavior in unlike topologies, since they
+        # may override the cluster-to-cluster behavior specified elsewhere on both clusters in an
+        # unexpected way. Therefore, forbid them.
+        if any(
+                v is not None for v in (self.config.MIXED_BIN_VERSIONS, self.config.NUM_SHARDS,
+                                        self.config.NUM_REPLSET_NODES)):
+            raise ValueError(
+                "ClusterToClusterFixture options must be specified through 'cluster0_options' and 'cluster1_options'."
+            )
+
+        for i, cluster_options in enumerate(parsed_options):
+            cluster_options["settings"] = self.fixturelib.default_if_none(
+                cluster_options["settings"], {})
+            if "preserve_dbpath" not in cluster_options["settings"]\
+                or cluster_options["settings"]["preserve_dbpath"] is None:
+                cluster_options["settings"]["preserve_dbpath"] = self.preserve_dbpath
+
+            cluster_options["settings"]["dbpath_prefix"] = os.path.join(
+                self._dbpath_prefix, f"cluster{i}")
+
+            if cluster_options["class"] == "ReplicaSetFixture":
+                cluster_options["settings"]["replicaset_logging_prefix"] = f"cl{i}"
+            elif cluster_options["class"] == "ShardedClusterFixture":
+                cluster_options["settings"]["cluster_logging_prefix"] = f"cl{i}"
+            else:
+                raise ValueError(f"Illegal fixture class: {cluster_options['class']}")
+
+            self.logger.info(f"Cluster{i} configured with settings: {cluster_options}")
+
+        self.both_cluster_options = parsed_options
+
+    def setup(self):
+        """Set up the cluster to cluster fixture according to the options provided."""
+
+        for cluster_options in self.both_cluster_options:
+            cluster = self.fixturelib.make_fixture(cluster_options["class"], self.logger,
+                                                   self.job_num, **cluster_options["settings"])
+            self.clusters.append(cluster)
+
+        for i, cluster in enumerate(self.clusters):
+            self.logger.info(f"Setting up cluster {i}.")
+            cluster.setup()
+
+    def pids(self):
+        """:return: pids owned by this fixture if any."""
+        out = []
+        for i, cluster in enumerate(self.clusters):
+            self.logger.info(f"Gathering cluster {i} pids: {cluster.pids()}")
+            out.extend(cluster.pids())
+        if not out:
+            self.logger.debug('No clusters when gathering cluster to cluster fixture pids.')
+        return out
+
+    def await_ready(self):
+        """Block until the fixture can be used for testing."""
+        # Wait for each of the clusters.
+        for cluster in self.clusters:
+            cluster.await_ready()
+
+    def _do_teardown(self, mode=None):
+        """Shut down the clusters."""
+        self.logger.info("Stopping all clusters...")
+
+        running_at_start = self.is_running()
+        if not running_at_start:
+            self.logger.warning("All clusters were expected to be running, but weren't.")
+
+        teardown_handler = interface.FixtureTeardownHandler(self.logger)
+
+        for i, cluster in enumerate(self.clusters):
+            teardown_handler.teardown(cluster, f"cluster {i}", mode=mode)
+
+        if teardown_handler.was_successful():
+            self.logger.info("Successfully stopped all clusters.")
+        else:
+            self.logger.error("Stopping the fixture failed.")
+            raise self.fixturelib.ServerFailure(teardown_handler.get_error_message())
+
+    def is_running(self):
+        """Return true if all clusters are still operating."""
+        return all(cluster.is_running() for cluster in self.clusters)
+
+    def get_node_info(self):
+        """Return a list of dicts of NodeInfo objects."""
+        output = []
+        for cluster in self.clusters:
+            output += cluster.get_node_info()
+        return output
+
+    def get_driver_connection_url(self):
+        """Return the driver connection URL to the cluster that starts out owning the data."""
+        if not self.clusters:
+            raise ValueError("Must call setup() before calling get_driver_connection_url")
+        return self.clusters[0].get_driver_connection_url()
+
+    def get_internal_connection_string(self):
+        """Return the internal connection string to the cluster that starts out owning the data."""
+        if not self.clusters:
+            raise ValueError("Must call setup() before calling get_internal_connection_string")
+        return self.clusters[0].get_internal_connection_string()
+
+    def get_cluster0_connection_string(self):
+        """Return the connection string of cluster 0."""
+        if not self.clusters:
+            raise ValueError("Must call setup() before calling get_cluster0_connection_string")
+        return self.clusters[0].get_internal_connection_string()
+
+    def get_cluster1_connection_string(self):
+        """Return the connection string of cluster 1."""
+        if not self.clusters:
+            raise ValueError("Must call setup() before calling get_cluster1_connection_string")
+        return self.clusters[1].get_internal_connection_string()
diff --git a/buildscripts/resmokelib/testing/fixtures/shardedcluster.py b/buildscripts/resmokelib/testing/fixtures/shardedcluster.py
index 031fbc1b772..5b334236848 100644
--- a/buildscripts/resmokelib/testing/fixtures/shardedcluster.py
+++ b/buildscripts/resmokelib/testing/fixtures/shardedcluster.py
@@ -17,9 +17,6 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
     _CONFIGSVR_REPLSET_NAME = "config-rs"
     _SHARD_REPLSET_NAME_PREFIX = "shard-rs"
 
-    _CONFIGSVR_SHARD_LOGGING_PREFIX = "configsvr"
-    _RS_SHARD_LOGGING_PREFIX = "shard"
-
     AWAIT_SHARDING_INITIALIZATION_TIMEOUT_SECS = 60
 
     def __init__(  # pylint: disable=too-many-arguments,too-many-locals
@@ -27,7 +24,7 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
             mongod_executable=None, mongod_options=None, dbpath_prefix=None, preserve_dbpath=False,
             num_shards=1, num_rs_nodes_per_shard=1, num_mongos=1, enable_sharding=None,
             enable_balancer=True, enable_autosplit=True, auth_options=None, configsvr_options=None,
-            shard_options=None):
+            shard_options=None, cluster_logging_prefix=None):
         """Initialize ShardedClusterFixture with different options for the cluster processes."""
 
         interface.Fixture.__init__(self, logger, job_num, fixturelib, dbpath_prefix=dbpath_prefix)
@@ -58,6 +55,13 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
         self.shard_options = self.fixturelib.make_historic(
             self.fixturelib.default_if_none(shard_options, {}))
 
+        # The logging prefix used in cluster to cluster replication.
+        self.cluster_logging_prefix = "" if cluster_logging_prefix is None else f"{cluster_logging_prefix}:"
+
+        self.configsvr_shard_logging_prefix = f"{self.cluster_logging_prefix}configsvr"
+        self.rs_shard_logging_prefix = f"{self.cluster_logging_prefix}shard"
+        self.mongos_logging_prefix = f"{self.cluster_logging_prefix}mongos"
+
         if self.num_rs_nodes_per_shard is None:
             raise TypeError("num_rs_nodes_per_shard must be an integer but found None")
         elif isinstance(self.num_rs_nodes_per_shard, int):
@@ -253,7 +257,7 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
     def get_configsvr_logger(self):
         """Return a new logging.Logger instance used for a config server shard."""
         return self.fixturelib.new_fixture_node_logger(self.__class__.__name__, self.job_num,
-                                                       self._CONFIGSVR_SHARD_LOGGING_PREFIX)
+                                                       self.configsvr_shard_logging_prefix)
 
     def get_configsvr_kwargs(self):
         """Return args to create replicaset.ReplicaSetFixture configured as the config server."""
@@ -278,7 +282,7 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
             "mongod_options": mongod_options, "mongod_executable": self.mongod_executable,
             "preserve_dbpath": preserve_dbpath, "num_nodes": num_nodes,
             "auth_options": auth_options, "replset_config_options": replset_config_options,
-            "shard_logging_prefix": self._CONFIGSVR_SHARD_LOGGING_PREFIX, **configsvr_options
+            "shard_logging_prefix": self.configsvr_shard_logging_prefix, **configsvr_options
         }
 
     def install_configsvr(self, configsvr):
@@ -287,7 +291,7 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
 
     def _get_rs_shard_logging_prefix(self, index):
         """Return replica set shard logging prefix."""
-        return f"{self._RS_SHARD_LOGGING_PREFIX}{index}"
+        return f"{self.rs_shard_logging_prefix}{index}"
 
     def get_rs_shard_logger(self, index):
         """Return a new logging.Logger instance used for a replica set shard."""
@@ -328,7 +332,7 @@ class ShardedClusterFixture(interface.Fixture):  # pylint: disable=too-many-inst
 
     def get_mongos_logger(self, index, total):
         """Return a new logging.Logger instance used for a mongos."""
-        logger_name = "mongos" if total == 1 else f"mongos{index}"
+        logger_name = self.mongos_logging_prefix if total == 1 else f"{self.mongos_logging_prefix}{index}"
         return self.fixturelib.new_fixture_node_logger(self.__class__.__name__, self.job_num,
                                                        logger_name)