Workflow to run performance tests using opensearch-benchmark (#3415)

Signed-off-by: Rishabh Singh <sngri@amazon.com>

Author: rishabh6788

src/run_benchmark_test.py

@@ -0,0 +1,29 @@
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+import sys
+
+from manifests.bundle_manifest import BundleManifest
+from system import console
+from test_workflow.benchmark_test.benchmark_args import BenchmarkArgs
+from test_workflow.benchmark_test.benchmark_test_runners import BenchmarkTestRunners
+
+
+def main() -> int:
+    """
+        Entry point for Benchmark Test with bundle manifest, config file containing the required arguments for running
+        benchmarking test. Will call out in test.sh with benchmark as argument
+    """
+    benchmark_args = BenchmarkArgs()
+    console.configure(level=benchmark_args.logging_level)
+    manifest = BundleManifest.from_file(benchmark_args.bundle_manifest)
+    BenchmarkTestRunners.from_args(benchmark_args, manifest).run()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/test_workflow/README.md

@@ -6,6 +6,7 @@
     - [Identifying Regressions in Performance Tests](#identifying-regressions-in-performance-tests)
       - [Identifying Regressions in Nightly Performance Tests](#identifying-regressions-in-nightly-performance-tests)
     - [Identifying Issues in Longevity Tests](#identifying-issues-in-longevity-tests)
+  - [Benchmark Tests](#benchmarking-tests)
 - [Testing in CI/CD](#testing-in-cicd)
   - [Test Workflow (in development)](#test-workflow-in-development)
   - [Component-Level Details](#component-level-details)
@@ -174,6 +175,13 @@ Internal tools provide dashboards for monitoring cluster behavior during these t
 |Indexing Latency|Consistent during each test iteration|upward trends|
 |Query Latency|Varies based on the query being issued|upward trends|
 
+### Benchmarking Tests
+
+Runs benchmarking tests on a remote opensource OpenSearch cluster, uses [OpenSearch Benchmark](https://github.com/opensearch-project/OpenSearch-Benchmark).
+At a high-level the benchmarking test workflow uses [opensearch-cluster-cdk](https://github.com/opensearch-project/opensearch-cluster-cdk.git) to first set-up an OpenSearch cluster (single/multi-node) and then executes `opensearch-benchmark` to run benchmark test against that cluster. The performance metric that opensearch-benchmark generates during the run are ingested into another OS cluster for further analysis and dashboarding purpose.
+
+The benchmarking tests will be run nightly and if you have a feature in any released/un-released OpenSearch version that you want to benchmark periodically please create an issue and the team will reach out to you. In case you want to run the benchmarking test locally you can use `opensearch-cluster-cdk` repo to spin up an OS cluster in your personal AWS account and then use `opensearch-benchmark` to run performance test against it. The detailed instructions are available on respective GitHub repositories.
+
 ## Testing in CI/CD
 
 The CI/CD infrastructure is divided into two main workflows - `build` and `test`. The `build` workflow automates the process to generate all OpenSearch and OpenSearch Dashboards artifacts, and provide them as distributions to the `test` workflow, which runs exhaustive testing on the artifacts based on the artifact type. The next section talks in detail about the test workflow.

src/test_workflow/benchmark_test/__init__.py

@@ -0,0 +1,6 @@
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.

src/test_workflow/benchmark_test/benchmark_args.py

@@ -0,0 +1,108 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+import argparse
+import json
+import logging
+from typing import IO
+
+from test_workflow.json_args import JsonArgs
+
+
+# Contains the arguments required to run a perf test.
+class BenchmarkArgs:
+    bundle_manifest: IO
+    stack_suffix: str
+    config: IO
+    keep: bool
+    insecure: bool
+    single_node: bool
+    min_distribution: bool
+    manager_node_count: int
+    data_node_count: int
+    client_node_count: int
+    ingest_node_count: int
+    ml_node_count: int
+    data_node_storage: int
+    ml_node_storage: int
+    jvm_sys_props: str
+    additional_config: str
+    workload: str
+    benchmark_config: IO
+    user_tag: str
+    target_hosts: str
+    logging_level: int
+
+    def __init__(self) -> None:
+        parser = argparse.ArgumentParser(description="Test an OpenSearch Bundle")
+        parser.add_argument("--bundle-manifest", type=argparse.FileType("r"), help="Bundle Manifest file.",
+                            required=True)
+        parser.add_argument("--suffix", dest="suffix", help="Suffix to be added to stack name for performance test")
+        parser.add_argument("--component", dest="component", default="OpenSearch",
+                            help="Component name that needs to be performance tested")
+        parser.add_argument("--config", type=argparse.FileType("r"), help="Config file.", required=True)
+        parser.add_argument(
+            "--without-security", dest="insecure", action="store_true",
+            help="Force the security of the cluster to be disabled.", default=False)
+        parser.add_argument("--keep", dest="keep", action="store_true",
+                            help="Do not delete the working temporary directory.")
+        parser.add_argument("--single-node", dest="single_node", action="store_true",
+                            help="Is this a single node cluster")
+        parser.add_argument("--min-distribution", dest="min_distribution", action="store_true",
+                            help="Is it the minimal OpenSearch distribution with no security and plugins")
+        parser.add_argument("--manager-node-count", dest="manager_node_count",
+                            help="Number of cluster manager nodes, default is 3")
+        parser.add_argument("--data-node-count", dest="data_node_count", help="Number of data nodes, default is 2")
+        parser.add_argument("--client-node-count", dest="client_node_count",
+                            help="Number of dedicated client nodes, default is 0")
+        parser.add_argument("--ingest-node-count", dest="ingest_node_count",
+                            help="Number of dedicated ingest nodes, default is 0")
+        parser.add_argument("--ml-node-count", dest="ml_node_count",
+                            help="Number of dedicated machine learning nodes, default is 0")
+        parser.add_argument("--jvm-sys-props", dest="jvm_sys_props",
+                            help="A comma-separated list of key=value pairs that will be added to jvm.options as JVM system properties.")
+        parser.add_argument("--additional-config", nargs='*', action=JsonArgs, dest="additional_config",
+                            help="Additional opensearch.yml config parameters passed as JSON")
+        parser.add_argument("--ml-node-storage", dest="ml_node_storage",
+                            help="User provided ml-node ebs block storage size defaults to 100Gb")
+        parser.add_argument("--data-node-storage", dest="data_node_storage",
+                            help="User provided data-node ebs block storage size, defaults to 100Gb")
+        parser.add_argument("--workload", dest="workload", help="workload type for the OpenSearch benchmarking",
+                            required=True)
+        parser.add_argument("--benchmark-config", dest="benchmark_config",
+                            help="absolute filepath to custom opensearch-benchmark.ini config")
+        parser.add_argument("--user-tag", dest="user_tag",
+                            help="Attach arbitrary text to the meta-data of each metric record")
+        parser.add_argument(
+            "-v", "--verbose", help="Show more verbose output.", action="store_const", default=logging.INFO,
+            const=logging.DEBUG, dest="logging_level"
+        )
+
+        args = parser.parse_args()
+        self.bundle_manifest = args.bundle_manifest
+        self.stack_suffix = args.suffix if args.suffix else None
+        self.config = args.config
+        self.keep = args.keep
+        self.single_node = args.single_node
+        self.min_distribution = args.min_distribution
+        self.component = args.component
+        self.insecure = args.insecure
+        self.manager_node_count = args.manager_node_count if args.manager_node_count else None
+        self.data_node_count = args.data_node_count if args.data_node_count else None
+        self.client_node_count = args.client_node_count if args.client_node_count else None
+        self.ingest_node_count = args.ingest_node_count if args.ingest_node_count else None
+        self.ml_node_count = args.ml_node_count if args.ml_node_count else None
+        self.jvm_sys_props = args.jvm_sys_props if args.jvm_sys_props else None
+        self.data_node_storage = args.data_node_storage if args.data_node_storage else None
+        self.ml_node_storage = args.ml_node_storage if args.ml_node_storage else None
+        self.workload = args.workload
+        self.benchmark_config = args.benchmark_config if args.benchmark_config else None
+        self.user_tag = args.user_tag if args.user_tag else None
+        self.additional_config = json.dumps(args.additional_config) if args.additional_config is not None else None
+        self.logging_level = args.logging_level

src/test_workflow/benchmark_test/benchmark_test_cluster.py

@@ -0,0 +1,160 @@
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+
+import json
+import logging
+import os
+import subprocess
+from contextlib import contextmanager
+from typing import Any, Generator
+
+import requests
+from requests.auth import HTTPBasicAuth
+from retry.api import retry_call  # type: ignore
+
+from manifests.bundle_manifest import BundleManifest
+from test_workflow.benchmark_test.benchmark_args import BenchmarkArgs
+
+
+class BenchmarkTestCluster:
+    manifest: BundleManifest
+    work_dir: str
+    current_workspace: str
+    args: BenchmarkArgs
+    output_file: str
+    params: str
+    is_endpoint_public: bool
+    cluster_endpoint: str
+    cluster_endpoint_with_port: str
+
+    """
+    Represents a performance test cluster. This class deploys the opensearch bundle with CDK. Supports both single
+    and multi-node clusters
+    """
+
+    def __init__(
+            self,
+            bundle_manifest: BundleManifest,
+            config: dict,
+            args: BenchmarkArgs,
+            current_workspace: str
+    ) -> None:
+        self.manifest = bundle_manifest
+        self.current_workspace = current_workspace
+        self.args = args
+        self.output_file = "output.json"
+        role = config["Constants"]["Role"]
+        params_dict = self.setup_cdk_params(config)
+        params_list = []
+        for key, value in params_dict.items():
+            if value:
+                '''
+                TODO: To send json input to typescript code from command line it needs to be enclosed in
+                single-quotes, this is a temp fix to achieve that since the quoted string passed from command line in
+                tesh.sh wrapper script gets un-quoted and we need to handle it here.
+                '''
+                if key == 'additionalConfig':
+                    params_list.append(f" -c {key}=\'{value}\'")
+                else:
+                    params_list.append(f" -c {key}={value}")
+        role_params = (
+            f" --require-approval=never --plugin cdk-assume-role-credential-plugin"
+            f" -c assume-role-credentials:writeIamRoleName={role} -c assume-role-credentials:readIamRoleName={role} "
+        )
+        self.params = "".join(params_list) + role_params
+        self.is_endpoint_public = False
+        self.cluster_endpoint = None
+        self.cluster_endpoint_with_port = None
+        self.stack_name = f"opensearch-infra-stack-{self.args.stack_suffix}-{self.manifest.build.id}-{self.manifest.build.architecture}"
+
+    def start(self) -> None:
+        command = f"npm install && cdk deploy \"*\" {self.params} --outputs-file {self.output_file}"
+
+        logging.info(f'Executing "{command}" in {os.getcwd()}')
+        subprocess.check_call(command, cwd=os.getcwd(), shell=True)
+        with open(self.output_file, "r") as read_file:
+            load_output = json.load(read_file)
+            self.create_endpoint(load_output)
+        self.wait_for_processing()
+
+    def create_endpoint(self, cdk_output: dict) -> None:
+        loadbalancer_url = cdk_output[self.stack_name].get('loadbalancerurl', None)
+        if loadbalancer_url is None:
+            raise RuntimeError("Unable to fetch the cluster endpoint from cdk output")
+        self.cluster_endpoint = loadbalancer_url
+        self.cluster_endpoint_with_port = "".join([loadbalancer_url, ":", str(self.port)])
+
+    @property
+    def endpoint(self) -> str:
+        return self.cluster_endpoint
+
+    @property
+    def endpoint_with_port(self) -> str:
+        return self.cluster_endpoint_with_port
+
+    @property
+    def port(self) -> int:
+        return 80 if self.args.insecure else 443
+
+    def terminate(self) -> None:
+        command = f"cdk destroy {self.stack_name} {self.params} --force"
+        logging.info(f'Executing "{command}" in {os.getcwd()}')
+
+        subprocess.check_call(command, cwd=os.getcwd(), shell=True)
+
+    def wait_for_processing(self, tries: int = 3, delay: int = 15, backoff: int = 2) -> None:
+        logging.info(f"Waiting for domain at {self.endpoint} to be up")
+        protocol = "http://" if self.args.insecure else "https://"
+        url = "".join([protocol, self.endpoint, "/_cluster/health"])
+        request_args = {"url": url} if self.args.insecure else {"url": url, "auth": HTTPBasicAuth("admin", "admin"), "verify": False}  # type: ignore
+        retry_call(requests.get, fkwargs=request_args,
+                   tries=tries, delay=delay, backoff=backoff)
+
+    def setup_cdk_params(self, config: dict) -> dict:
+        if self.args.stack_suffix:
+            suffix = self.args.stack_suffix + '-' + self.manifest.build.id + '-' + self.manifest.build.architecture
+        else:
+            suffix = self.manifest.build.id + '-' + self.manifest.build.architecture
+        return {
+            "distributionUrl": self.manifest.build.location,
+            "vpcId": config["Constants"]["VpcId"],
+            "account": config["Constants"]["AccountId"],
+            "region": config["Constants"]["Region"],
+            "suffix": suffix,
+            "securityDisabled": str(self.args.insecure).lower(),
+            "cpuArch": self.manifest.build.architecture,
+            "singleNodeCluster": str(self.args.single_node).lower(),
+            "distVersion": self.manifest.build.version,
+            "minDistribution": str(self.args.min_distribution).lower(),
+            "serverAccessType": config["Constants"]["serverAccessType"],
+            "restrictServerAccessTo": config["Constants"]["restrictServerAccessTo"],
+            "additionalConfig": self.args.additional_config,
+            "managerNodeCount": self.args.manager_node_count,
+            "dataNodeCount": self.args.data_node_count,
+            "clientNodeCount": self.args.client_node_count,
+            "ingestNodeCount": self.args.ingest_node_count,
+            "mlNodeCount": self.args.ml_node_count,
+            "dataNodeStorage": self.args.data_node_storage,
+            "mlNodeStorage": self.args.ml_node_storage,
+            "jvmSysProps": self.args.jvm_sys_props
+        }
+
+    @classmethod
+    @contextmanager
+    def create(cls, *args: Any) -> Generator[Any, None, None]:
+        """
+        Set up the cluster. When this method returns, the cluster must be available to take requests.
+        Throws ClusterCreationException if the cluster could not start for some reason. If this exception is thrown, the caller does not need to call "destroy".
+        """
+        cluster = cls(*args)
+
+        try:
+            cluster.start()
+            yield cluster
+        finally:
+            cluster.terminate()

src/test_workflow/benchmark_test/benchmark_test_runner.py

@@ -0,0 +1,34 @@
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+import abc
+import os
+
+from manifests.bundle_manifest import BundleManifest
+from test_workflow.benchmark_test.benchmark_args import BenchmarkArgs
+
+
+class BenchmarkTestRunner(abc.ABC):
+    args: BenchmarkArgs
+    test_manifest: BundleManifest
+    security: bool
+    tests_dir: str
+
+    def __init__(self, args: BenchmarkArgs, test_manifest: BundleManifest) -> None:
+        self.args = args
+        self.test_manifest = test_manifest
+
+        self.security = "security" in self.test_manifest.components and not self.args.insecure
+        self.tests_dir = os.path.join(os.getcwd(), "test-results", "benchmark-test", f"{'with' if self.security else 'without'}-security")
+        os.makedirs(self.tests_dir, exist_ok=True)
+
+    @abc.abstractmethod
+    def run_tests(self) -> None:
+        pass
+
+    def run(self) -> None:
+        self.run_tests()