Search: e2e for sharded with single mongot (#823)

Missing e2e test for sharded search with one mongot instance

Author: lsierant

.evergreen-tasks.yml

@@ -1359,3 +1359,8 @@ tasks:
     tags: [ "patch-run" ]
     commands:
       - func: "e2e_test"
+
+  - name: e2e_search_sharded_external_mongod_single_mongot
+    tags: [ "patch-run" ]
+    commands:
+      - func: "e2e_test"

.evergreen.yml

@@ -811,6 +811,7 @@ task_groups:
       - e2e_search_enterprise_basic
       - e2e_search_enterprise_tls
       - e2e_search_enterprise_x509_cluster_auth
+      - e2e_search_sharded_external_mongod_single_mongot
     <<: *teardown_group
 
   # this task group contains just a one task, which is smoke testing whether the operator
@@ -1318,7 +1319,7 @@ buildvariants:
     display_name: e2e_mdb_kind_ubi_cloudqa
     tags: [ "pr_patch", "staging", "e2e_test_suite", "cloudqa", "cloudqa_non_static" ]
     run_on:
-      - ubuntu2404-medium
+      - ubuntu2404-large
     <<: *base_no_om_image_dependency
     tasks:
       - name: e2e_mdb_kind_cloudqa_task_group
@@ -1336,7 +1337,7 @@ buildvariants:
     display_name: e2e_static_mdb_kind_ubi_cloudqa
     tags: [ "pr_patch", "staging", "e2e_test_suite", "cloudqa", "static" ]
     run_on:
-      - ubuntu2404-medium
+      - ubuntu2404-large
     <<: *base_no_om_image_dependency
     tasks:
       - name: e2e_mdb_kind_cloudqa_task_group

docker/mongodb-kubernetes-tests/kubetester/certs.py

@@ -487,6 +487,7 @@ def create_sharded_cluster_certs(
     shard_distribution: Optional[List[int]] = None,
     mongos_distribution: Optional[List[int]] = None,
     config_srv_distribution: Optional[List[int]] = None,
+    mongos_service_dns_names: Optional[List[str]] = None,
 ):
     cert_generation_func = create_mongodb_tls_certs
     if x509_certs:
@@ -573,9 +574,8 @@ def create_sharded_cluster_certs(
             secret_backend=secret_backend,
         )
 
-    additional_domains_for_mongos = None
+    additional_domains_for_mongos = []
     if additional_domains is not None:
-        additional_domains_for_mongos = []
         for domain in additional_domains:
             if mongos_distribution is None:
                 for pod_idx in range(mongos):
@@ -585,6 +585,10 @@ def create_sharded_cluster_certs(
                     for pod_idx in range(pod_count or 0):
                         additional_domains_for_mongos.append(f"{resource_name}-mongos-{cluster_idx}-{pod_idx}.{domain}")
 
+    # Add service DNS names directly (e.g., for mongot to connect to mongos service)
+    if mongos_service_dns_names is not None:
+        additional_domains_for_mongos.extend(mongos_service_dns_names)
+
     secret_name = f"{resource_name}-mongos-cert"
     if secret_prefix is not None:
         secret_name = secret_prefix + secret_name
@@ -596,7 +600,7 @@ def create_sharded_cluster_certs(
         service_name=resource_name + "-svc",
         replicas=mongos,
         replicas_cluster_distribution=mongos_distribution,
-        additional_domains=additional_domains_for_mongos,
+        additional_domains=additional_domains_for_mongos if additional_domains_for_mongos else None,
         secret_backend=secret_backend,
     )
 

docker/mongodb-kubernetes-tests/kubetester/tests/test___init__.py

@@ -5,7 +5,7 @@
 from kubeobject import CustomObject
 
 
-class TestCreateOrUpdate(ctx, unittest.TestCase):
+class TestCreateOrUpdate(unittest.TestCase):
     def test_create_or_update_is_not_bound(self):
         api_client = MagicMock()
         custom_object = CustomObject(

docker/mongodb-kubernetes-tests/tests/common/mongodb_tools_pod/__init__.py

@@ -0,0 +1 @@
+

docker/mongodb-kubernetes-tests/tests/common/mongodb_tools_pod/mongodb_tools_pod.py

@@ -0,0 +1,108 @@
+import tarfile
+import tempfile
+
+from kubernetes import client, config
+from kubernetes.stream import stream
+from kubetester import get_pod_when_ready
+from tests import test_logger
+
+logger = test_logger.get_test_logger(__name__)
+
+TOOLS_POD_NAME = "mongodb-tools-pod"
+TOOLS_POD_IMAGE = "mongodb/mongodb-community-server:8.0-ubi9"
+
+
+class ToolsPod:
+    """A pod running MongoDB tools for executing commands like mongorestore inside the cluster."""
+
+    def __init__(self, namespace: str):
+        self.namespace = namespace
+        self.pod_name = TOOLS_POD_NAME
+        self.core_v1 = client.CoreV1Api()
+
+    def run_command(self, cmd: list[str]):
+        """Execute a command in the tools pod and return the output."""
+        logger.debug(f"Running command in {self.pod_name}: {' '.join(cmd)}")
+        resp = stream(
+            self.core_v1.connect_get_namespaced_pod_exec,
+            self.pod_name,
+            self.namespace,
+            command=cmd,
+            stderr=True,
+            stdin=False,
+            stdout=True,
+            tty=False,
+        )
+        logger.debug(f"Command output: {resp}")
+        return resp
+
+    def copy_file_to_pod(self, src_path: str, dest_path: str):
+        """Copy a file from the local filesystem to the tools pod."""
+        logger.debug(f"Copying {src_path} to {self.pod_name}:{dest_path}")
+
+        # Create a tar archive containing the file
+        with tempfile.NamedTemporaryFile(suffix=".tar") as tar_file:
+            with tarfile.open(tar_file.name, "w") as tar:
+                tar.add(src_path, arcname=dest_path.split("/")[-1])
+
+            tar_file.seek(0)
+            tar_data = tar_file.read()
+
+        # Extract the tar archive in the pod
+        exec_command = ["tar", "xf", "-", "-C", "/".join(dest_path.split("/")[:-1]) or "/"]
+        resp = stream(
+            self.core_v1.connect_get_namespaced_pod_exec,
+            self.pod_name,
+            self.namespace,
+            command=exec_command,
+            stderr=True,
+            stdin=True,
+            stdout=True,
+            tty=False,
+            _preload_content=False,
+        )
+
+        # Send the tar data
+        resp.write_stdin(tar_data)
+        resp.close()
+        logger.debug(f"File copied to {self.pod_name}:{dest_path}")
+
+    def run_pod_and_wait(self):
+        """Create the tools pod and wait for it to be ready."""
+        pod_body = client.V1Pod(
+            api_version="v1",
+            kind="Pod",
+            metadata=client.V1ObjectMeta(name=self.pod_name, labels={"app": "mongodb-tools"}),
+            spec=client.V1PodSpec(
+                containers=[
+                    client.V1Container(
+                        name="mongodb-tools",
+                        image=TOOLS_POD_IMAGE,
+                        command=["/bin/bash", "-c"],
+                        args=["sleep infinity"],
+                    )
+                ],
+                restart_policy="Never",
+            ),
+        )
+
+        try:
+            self.core_v1.create_namespaced_pod(namespace=self.namespace, body=pod_body)
+            logger.info(f"Created {self.pod_name} in namespace {self.namespace}")
+        except client.exceptions.ApiException as e:
+            if e.status == 409:
+                logger.info(f"Pod {self.pod_name} already exists")
+            else:
+                raise
+
+        pod = get_pod_when_ready(self.namespace, "app=mongodb-tools", default_retry=120)
+        if pod is None:
+            raise TimeoutError(f"Timed out waiting for {self.pod_name} to be ready")
+        logger.info(f"{self.pod_name} is ready")
+
+
+def get_tools_pod(namespace: str) -> ToolsPod:
+    """Create and return a ready tools pod in the given namespace."""
+    tools_pod = ToolsPod(namespace)
+    tools_pod.run_pod_and_wait()
+    return tools_pod

docker/mongodb-kubernetes-tests/tests/common/search/movies_search_helper.py

@@ -1,8 +1,7 @@
-import logging
-
 import pymongo.errors
 from kubetester import kubetester
 from tests import test_logger
+from tests.common.mongodb_tools_pod.mongodb_tools_pod import ToolsPod
 from tests.common.search.search_tester import SearchTester
 
 logger = test_logger.get_test_logger(__name__)
@@ -13,15 +12,19 @@ class SampleMoviesSearchHelper:
     db_name: str
     col_name: str
     archive_url: str
+    tools_pod: ToolsPod
 
-    def __init__(self, search_tester: SearchTester):
+    def __init__(self, search_tester: SearchTester, tools_pod: ToolsPod):
         self.search_tester = search_tester
+        self.tools_pod = tools_pod
         self.db_name = "sample_mflix"
         self.col_name = "movies"
 
     def restore_sample_database(self):
         self.search_tester.mongorestore_from_url(
-            "https://atlas-education.s3.amazonaws.com/sample_mflix.archive", f"{self.db_name}.*"
+            "https://atlas-education.s3.amazonaws.com/sample_mflix.archive",
+            f"{self.db_name}.*",
+            self.tools_pod,
         )
 
     def create_search_index(self):

docker/mongodb-kubernetes-tests/tests/common/search/search_tester.py

@@ -1,12 +1,10 @@
-import os
-import tempfile
+from typing import Optional, Self
 
 import kubetester
-import requests
-from kubetester.helm import process_run_and_check
 from kubetester.mongotester import MongoTester
 from pymongo.operations import SearchIndexModel
 from tests import test_logger
+from tests.common.mongodb_tools_pod.mongodb_tools_pod import ToolsPod
 
 logger = test_logger.get_test_logger(__name__)
 
@@ -16,23 +14,100 @@ def __init__(
         self,
         connection_string: str,
         use_ssl: bool = False,
-        ca_path: str | None = None,
+        ca_path: Optional[str] = None,
     ):
         super().__init__(connection_string, use_ssl, ca_path)
 
-    def mongorestore_from_url(self, archive_url: str, ns_include: str, mongodb_tools_dir: str = ""):
-        logger.debug(f"running mongorestore from {archive_url}")
-        with tempfile.NamedTemporaryFile(delete=False) as sample_file:
-            resp = requests.get(archive_url)
-            size = sample_file.write(resp.content)
-            logger.debug(f"Downloaded sample file from {archive_url} to {sample_file.name} (size: {size})")
-            mongorestore_path = os.path.join(mongodb_tools_dir, "mongorestore")
-            mongorestore_cmd = f"{mongorestore_path} --archive={sample_file.name} --verbose=1 --drop --nsInclude {ns_include} --uri={self.cnx_string}"
-            if self.default_opts.get("tls", False):
-                mongorestore_cmd += " --ssl"
-            if ca_path := self.default_opts.get("tlsCAFile"):
-                mongorestore_cmd += " --sslCAFile=" + ca_path
-            process_run_and_check(mongorestore_cmd.split(), capture_output=True)
+    @classmethod
+    def for_replicaset(
+        cls,
+        mdb,
+        user_name: str,
+        password: str,
+        use_ssl: bool = False,
+        ca_path: Optional[str] = None,
+    ) -> Self:
+        """Create SearchTester for a replica set MongoDB resource.
+
+        Args:
+            mdb: MongoDB or MongoDBCommunity resource with name and namespace attributes
+            user_name: Username for authentication
+            password: Password for authentication
+            use_ssl: Whether to use TLS/SSL connection
+            ca_path: Path to CA certificate file (required if use_ssl=True)
+
+        Returns:
+            SearchTester instance configured for the replica set
+        """
+        conn_str = (
+            f"mongodb://{user_name}:{password}@"
+            f"{mdb.name}-0.{mdb.name}-svc.{mdb.namespace}.svc.cluster.local:27017/"
+            f"?replicaSet={mdb.name}"
+        )
+        return cls(conn_str, use_ssl=use_ssl, ca_path=ca_path)
+
+    @classmethod
+    def for_sharded(
+        cls,
+        mdb,
+        user_name: str,
+        password: str,
+        use_ssl: bool = False,
+        ca_path: Optional[str] = None,
+    ) -> Self:
+        """Create SearchTester for a sharded MongoDB resource (connects to mongos).
+
+        Args:
+            mdb: MongoDB resource with name and namespace attributes (sharded cluster)
+            user_name: Username for authentication
+            password: Password for authentication
+            use_ssl: Whether to use TLS/SSL connection
+            ca_path: Path to CA certificate file (required if use_ssl=True)
+
+        Returns:
+            SearchTester instance configured for the sharded cluster via mongos
+        """
+        conn_str = (
+            f"mongodb://{user_name}:{password}@"
+            f"{mdb.name}-mongos-0.{mdb.name}-svc.{mdb.namespace}.svc.cluster.local:27017/"
+            f"?authSource=admin"
+        )
+        return cls(conn_str, use_ssl=use_ssl, ca_path=ca_path)
+
+    def mongorestore_from_url(self, archive_url: str, ns_include: str, tools_pod: ToolsPod):
+        """Run mongorestore from a URL using the tools pod.
+
+        Args:
+            archive_url: URL to download the archive from
+            ns_include: Namespace include pattern for mongorestore
+            tools_pod: ToolsPod instance to run the command in
+        """
+        logger.debug(f"running mongorestore from {archive_url} via tools pod")
+        archive_path = "/tmp/sample.archive"
+
+        # Download the archive directly in the pod using curl
+        tools_pod.run_command(["curl", "-o", archive_path, "-L", archive_url])
+
+        # Build mongorestore command
+        mongorestore_cmd = [
+            "mongorestore",
+            f"--archive={archive_path}",
+            "--verbose=1",
+            "--drop",
+            "--nsInclude",
+            ns_include,
+            f"--uri={self.cnx_string}",
+        ]
+
+        if self.default_opts.get("tls", False):
+            mongorestore_cmd.append("--ssl")
+        if ca_path := self.default_opts.get("tlsCAFile"):
+            # Copy CA cert to pod and use it
+            pod_ca_path = "/tmp/ca.crt"
+            tools_pod.copy_file_to_pod(ca_path, pod_ca_path)
+            mongorestore_cmd.append(f"--sslCAFile={pod_ca_path}")
+
+        tools_pod.run_command(mongorestore_cmd)
 
     def create_search_index(self, database_name: str, collection_name: str):
         database = self.client[database_name]
@@ -87,9 +162,19 @@ def search_indexes_ready(self, database_name: str, collection_name: str):
             return False
 
         for idx in search_indexes:
-            if idx.get("status") != "READY":
-                logger.debug(f"{database_name}/{collection_name}: search index {idx} is not ready")
-                return False
+            status = idx.get("status")
+            queryable = idx.get("queryable")
+            if status == "READY" or queryable is True:
+                continue
+            if status is None and queryable is None and idx.get("latestDefinition") is not None:
+                logger.debug(
+                    f"{database_name}/{collection_name}: search index {idx.get('name')} has no status but has latestDefinition, considering ready"
+                )
+                continue
+            logger.debug(
+                f"{database_name}/{collection_name}: search index {idx} is not ready (status={status}, queryable={queryable})"
+            )
+            return False
         return True
 
     def get_search_indexes(self, database_name, collection_name):