Add Jenkins CI for CCR Performance tests

Signed-off-by: Ankit Kala <ankikala@amazon.com>

jenkins/cross-cluster-replication/perf-test.jenkinsfile

@@ -0,0 +1,141 @@
+lib = library(identifier: "jenkins@20211118", retriever: legacySCM(scm))
+
+pipeline {
+    agent none
+    options {
+        timeout(time: 10, unit: 'HOURS')
+    }
+    environment {
+        AGENT_LABEL = 'Jenkins-Agent-al2-x64-c54xlarge-Docker-Host'
+        AGENT_IMAGE = 'opensearchstaging/ci-runner:ci-runner-centos7-v1'
+        BUNDLE_MANIFEST = 'bundle-manifest.yml'
+        JOB_NAME = 'ccr-perf-test'
+    }
+    parameters {
+        string(
+            name: 'GITHUB_TOKEN',
+            description: 'Github token for account access.',
+            trim: true
+        )
+        string(
+            name: 'BUNDLE_MANIFEST_URL',
+            description: 'The bundle manifest URL, e.g. https://ci.opensearch.org/ci/dbc/distribution-build-opensearch/1.2.2/98/linux/x64/builds/opensearch/manifest.yml.',
+            trim: true
+        )
+    }
+
+    stages {
+        stage('validate-and-set-parameters') {
+            agent {
+                docker {
+                    label AGENT_LABEL
+                    image AGENT_IMAGE
+                    alwaysPull true
+                }
+            }
+            steps {
+                script {
+                    if (BUNDLE_MANIFEST_URL == '') {
+                        currentBuild.result = 'ABORTED'
+                        error("Performance Tests failed to start. Missing parameter: BUNDLE_MANIFEST_URL.")
+                    }
+                    if (GITHUB_TOKEN == '') {
+                        currentBuild.result = 'ABORTED'
+                        error("Performance Tests failed to start. Missing parameter: GITHUB_TOKEN.")
+                    }
+                    def bundleManifestObj = downloadBuildManifest(
+                        url: BUNDLE_MANIFEST_URL,
+                        path: BUNDLE_MANIFEST
+                    )
+                    String buildId = bundleManifestObj.getArtifactBuildId()
+                    env.BUILD_ID = buildId
+                    env.HAS_SECURITY = bundleManifestObj.components.containsKey("security")
+                    env.ARCHITECTURE = bundleManifestObj.getArtifactArchitecture()
+                    echo "HAS_SECURITY: ${env.HAS_SECURITY}"
+                    lib.jenkins.Messages.new(this).add(JOB_NAME, "CCR Performance tests for #${BUILD_ID}")
+                }
+            }
+        }
+        stage('perf-test') {
+            agent {
+                docker {
+                    label AGENT_LABEL
+                    image AGENT_IMAGE
+                    alwaysPull true
+                }
+            }
+            when {
+                expression { return env.HAS_SECURITY }
+            }
+            steps {
+                script {
+                    def bundleManifestObj = downloadBuildManifest(
+                        url: BUNDLE_MANIFEST_URL,
+                        path: BUNDLE_MANIFEST
+                    )
+                    echo "BUNDLE_MANIFEST: ${BUNDLE_MANIFEST}"
+                    echo "BUILD_ID: ${BUILD_ID}"
+                    echo "Architecture: ${ARCHITECTURE}"
+
+                    runPerfTestScript(bundleManifest: BUNDLE_MANIFEST,
+                        buildId: BUILD_ID,
+                        architecture: ARCHITECTURE,
+                        component: "cross-cluster-replication")
+
+                    lib.jenkins.Messages.new(this).add(JOB_NAME,
+                        lib.jenkins.Messages.new(this).get([JOB_NAME]) +
+                        "\nCCR Performance tests for ${BUILD_ID} completed")
+                }
+            }
+            post {
+                success {
+                    script {
+                        uploadTestResults(
+                            buildManifestFileName: BUNDLE_MANIFEST,
+                            jobName: JOB_NAME,
+                            buildNumber: BUILD_ID
+                        )
+                    }
+                    postCleanup()
+                }
+                failure {
+                    postCleanup()
+                }
+                aborted {
+                    postCleanup()
+                }
+            }
+        }
+    }
+
+    post {
+        success {
+            node(AGENT_LABEL) {
+                script {
+                    def stashed = lib.jenkins.Messages.new(this).get([JOB_NAME])
+                    publishNotification(
+                        icon: ':white_check_mark:',
+                        message: 'CCR Performance Tests Successful',
+                        extra: stashed,
+                        credentialsId: 'INTEG_TEST_WEBHOOK',
+                    )
+                    postCleanup()
+                }
+            }
+        }
+        failure {
+            node(AGENT_LABEL) {
+                script {
+                    def stashed = lib.jenkins.Messages.new(this).get([JOB_NAME])
+                    publishNotification(
+                        icon: ':warning:',
+                        message: 'Failed CCR Performance Tests',
+                        extra: stashed,
+                        credentialsId: 'INTEG_TEST_WEBHOOK',
+                    )
+                    postCleanup()
+                }
+            }
+        }
+    }
+}

src/test_workflow/perf_test/perf_test_cluster.py

@@ -13,32 +13,20 @@
 
 class PerfTestCluster(TestCluster):
     """
-    Represents a performance test cluster. This class deploys the opensearch bundle with CDK and returns the private IP.
+    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, config, stack_name, security, current_workspace):
+    def __init__(self, bundle_manifest, config, stack_name, cluster_config, current_workspace):
         self.manifest = bundle_manifest
-        self.work_dir = os.path.join(current_workspace, "opensearch-cluster", "cdk", "single-node")
+        self.work_dir = os.path.join(current_workspace, "opensearch-cluster", "cdk",
+                                     "single-node" if cluster_config.is_single_node_cluster() else "multi-node")
         self.current_workspace = current_workspace
         self.stack_name = stack_name
-        self.cluster_endpoint = None
-        self.cluster_port = None
         self.output_file = "output.json"
-        self.private_ip = None
-        self.security = security
+        self.cluster_config = cluster_config
         role = config["Constants"]["Role"]
-        params_dict = {
-            "url": self.manifest.build.location,
-            "security_group_id": config["Constants"]["SecurityGroupId"],
-            "vpc_id": config["Constants"]["VpcId"],
-            "account_id": config["Constants"]["AccountId"],
-            "region": config["Constants"]["Region"],
-            "stack_name": self.stack_name,
-            "security": "enable" if self.security else "disable",
-            "platform": self.manifest.build.platform,
-            "architecture": self.manifest.build.architecture,
-            "public_ip": config["Constants"].get("PublicIp", "disable")
-        }
+        params_dict = self.setup_cdk_params(config)
         params_list = []
         for key, value in params_dict.items():
             params_list.append(f" -c {key}={value}")
@@ -47,8 +35,9 @@ def __init__(self, bundle_manifest, config, stack_name, security, current_worksp
             f" -c assume-role-credentials:writeIamRoleName={role} -c assume-role-credentials:readIamRoleName={role} "
         )
         self.params = "".join(params_list) + role_params
-        self.cluster_endpoint = None
-        self.public_ip = None
+        self.is_endpoint_public = False
+        self.cluster_endpoint_with_port = None
+        self.endpoint = None
 
     def start(self):
         os.chdir(self.work_dir)
@@ -57,22 +46,31 @@ def start(self):
         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.private_ip = load_output[self.stack_name]["PrivateIp"]
-        logging.info(f"Private IP: {self.private_ip}")
-        self.public_ip = load_output[self.stack_name].get("PublicIp", None)
+            self.create_endpoint(load_output)
+
+    def create_endpoint(self, cdk_output):
+        scheme = "https://" if self.cluster_config.security else "http://"
+        if self.cluster_config.is_single_node_cluster():
+            private_ip = cdk_output[self.stack_name]["PrivateIp"]
+            public_ip = cdk_output[self.stack_name].get("PublicIp", None)
+            self.is_endpoint_public = public_ip is not None
+            host = public_ip if public_ip is not None else private_ip
+        else:
+            host = cdk_output[self.stack_name]["LoadBalancerEndpoint"]
+            self.is_endpoint_public = True
+
+        if host is not None:
+            self.endpoint = host
+            self.cluster_endpoint_with_port = "".join([scheme, host, ":", str(self.port())])
+
+    def endpoint_with_port(self):
+        return self.cluster_endpoint_with_port
 
     def endpoint(self):
-        if self.cluster_endpoint is None:
-            scheme = "https://" if self.security else "http://"
-            # If instances are configured to have public ip, use that instead.
-            host = self.private_ip if self.public_ip is None else self.public_ip
-            if host is not None:
-                self.cluster_endpoint = "".join([scheme, host, ":", str(self.port())])
-        return self.cluster_endpoint
+        return self.endpoint
 
     def port(self):
-        self.cluster_port = 443 if self.security else 9200
-        return self.cluster_port
+        return 443 if self.cluster_config.security else 80
 
     def terminate(self):
         os.chdir(os.path.join(self.current_workspace, self.work_dir))
@@ -87,12 +85,43 @@ def dependencies(self):
         return []
 
     def wait_for_processing(self, tries=3, delay=15, backoff=2):
-        if self.public_ip is None:
-            return
-        url = "".join([self.endpoint(), "/_cluster/health"])
+        # Should be invoked only if the endpoint is public.
+        assert self.is_endpoint_public, "wait_for_processing should be invoked only when cluster is public"
+        print("Waiting for domain to be up")
+        url = "".join([self.endpoint_with_port(), "/_cluster/health"])
         retry_call(requests.get, fkwargs={"url": url, "auth": HTTPBasicAuth('admin', 'admin'), "verify": False},
                    tries=tries, delay=delay, backoff=backoff)
 
+    def setup_cdk_params(self, config):
+        if self.cluster_config.is_single_node_cluster():
+            return {
+                "url": self.manifest.build.location,
+                "security_group_id": config["Constants"]["SecurityGroupId"],
+                "vpc_id": config["Constants"]["VpcId"],
+                "account_id": config["Constants"]["AccountId"],
+                "region": config["Constants"]["Region"],
+                "stack_name": self.stack_name,
+                "security": "enable" if self.cluster_config.security else "disable",
+                "platform": self.manifest.build.platform,
+                "architecture": self.manifest.build.architecture,
+                "public_ip": config["Constants"].get("PublicIp", "disable")
+            }
+        else:
+            return {
+                "url": self.manifest.build.location,
+                "security_group_id": config["Constants"]["SecurityGroupId"],
+                "vpc_id": config["Constants"]["VpcId"],
+                "account_id": config["Constants"]["AccountId"],
+                "region": config["Constants"]["Region"],
+                "cluster_stack_name": self.stack_name,
+                "security": "enable" if self.cluster_config.security else "disable",
+                "architecture": self.manifest.build.architecture,
+                "master_node_count": int(self.cluster_config.master_nodes),
+                "data_node_count": int(self.cluster_config.data_nodes),
+                "ingest_node_count": int(self.cluster_config.ingest_nodes),
+                "client_node_count": int(self.cluster_config.client_nodes)
+            }
+
     @classmethod
     @contextmanager
     def create(cls, *args):

src/test_workflow/perf_test/perf_test_cluster_config.py

@@ -0,0 +1,20 @@
+# 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.
+
+class PerfTestClusterConfig():
+    """
+    Maintains the cluster level configuration.
+    """
+    def __init__(self, security=False, data_nodes=1, master_nodes=0, ingest_nodes=0, client_nodes=0):
+        self.security = security
+        self.data_nodes = data_nodes
+        self.master_nodes = master_nodes
+        self.ingest_nodes = ingest_nodes
+        self.client_nodes = client_nodes
+
+    def is_single_node_cluster(self):
+        return True if (self.data_nodes == 1 and self.master_nodes == 0
+                        and self.ingest_nodes == 0 and self.client_nodes == 0) else False

src/test_workflow/perf_test/perf_test_runner_opensearch.py

@@ -16,6 +16,7 @@
 from system.working_directory import WorkingDirectory
 from test_workflow.perf_test.perf_args import PerfArgs
 from test_workflow.perf_test.perf_test_cluster import PerfTestCluster
+from test_workflow.perf_test.perf_test_cluster_config import PerfTestClusterConfig
 from test_workflow.perf_test.perf_test_runner import PerfTestRunner
 from test_workflow.perf_test.perf_test_suite import PerfTestSuite
 
@@ -40,6 +41,6 @@ def run_tests(self):
             logging.info("current_workspace is " + str(current_workspace))
             with GitRepository(self.get_infra_repo_url(), "main", current_workspace):
                 with WorkingDirectory(current_workspace):
-                    with PerfTestCluster.create(self.test_manifest, config, self.args.stack, self.security, current_workspace) as test_cluster:
-                        perf_test_suite = PerfTestSuite(self.test_manifest, test_cluster.endpoint(), self.security, current_workspace, self.tests_dir, self.args)
+                    with PerfTestCluster.create(self.test_manifest, config, self.args.stack, PerfTestClusterConfig(self.security), current_workspace) as test_cluster:
+                        perf_test_suite = PerfTestSuite(self.test_manifest, test_cluster.endpoint_with_port(), self.security, current_workspace, self.tests_dir, self.args)
                         retry_call(perf_test_suite.execute, tries=3, delay=60, backoff=2)

src/test_workflow/perf_test/perf_test_runner_opensearch_plugins.py

@@ -7,8 +7,6 @@
 import os
 import subprocess
 
-import yaml
-
 from git.git_repository import GitRepository
 from manifests.bundle_manifest import BundleManifest
 from system.temporary_directory import TemporaryDirectory
@@ -23,20 +21,21 @@ class PerfTestRunnerOpenSearchPlugins(PerfTestRunner):
     """
     def __init__(self, args: PerfArgs, test_manifest: BundleManifest):
         super().__init__(args, test_manifest)
+        self.tests_dir = os.path.join(os.getcwd(), "test-results", "perf-test", self.args.component)
+        os.makedirs(self.tests_dir, exist_ok=True)
+        security_flag = "--without-security" if not self.security else ""
         self.command = (
-            f"python3 run_perf_test.py --config {yaml.safe_load(self.args.config)} "
-            f"--bundle-manifest {str(self.args.bundle_manifest.name)}"
+            f"./run_perf_test.sh --config {str(os.path.abspath(self.args.config.name))} "
+            f"--bundle-manifest {str(os.path.abspath(self.args.bundle_manifest.name))} "
+            f"--test-result-dir {str(self.tests_dir)} {security_flag}"
         )
 
     def get_plugin_repo_url(self):
         return f"https://github.com/opensearch-project/{self.args.component}.git"
 
     def run_tests(self):
         with TemporaryDirectory(keep=self.args.keep, chdir=True) as work_dir:
-            current_workspace = os.path.join(work_dir.name, "plugin")
+            current_workspace = os.path.join(work_dir.name, self.args.component)
             with GitRepository(self.get_plugin_repo_url(), "main", current_workspace):
                 with WorkingDirectory(current_workspace):
-                    if self.security:
-                        subprocess.check_call(f"{self.command} -s", cwd=os.getcwd(), shell=True)
-                    else:
-                        subprocess.check_call(f"{self.command}", cwd=os.getcwd(), shell=True)
+                    subprocess.check_call(f"{self.command}", cwd=os.getcwd(), shell=True)

tests/jenkins/TestRunNonSecurityPerfTestScript.groovy

@@ -88,7 +88,7 @@ class TestRunNonSecurityPerfTestScript extends BuildPipelineTest {
 
         assertThat(testScriptCommands.size(), equalTo(1))
         assertThat(testScriptCommands, hasItem(
-            "./test.sh perf-test --stack test-single-1236-x64 --bundle-manifest tests/jenkins/data/opensearch-1.3.0-non-security-bundle.yml --config config.yml --without-security --workload nyc_taxis --test-iters 1 --warmup-iters 1".toString()
+            "./test.sh perf-test --stack test-single-1236-x64 --bundle-manifest tests/jenkins/data/opensearch-1.3.0-non-security-bundle.yml --config config.yml --without-security --workload nyc_taxis --test-iters 1 --warmup-iters 1 ".toString()
         ))
 
         def resultUploadScriptCommands = getCommandExecutions('s3Upload', 'test-results').findAll {

tests/jenkins/jenkinsjob-regression-files/opensearch/perf-test-with-security.jenkinsfile.txt

@@ -51,7 +51,7 @@
     )
                   runPerfTestScript.withAWS({role=opensearch-test, roleAccount=dummy_account, duration=900, roleSessionName=jenkins-session}, groovy.lang.Closure)
                      runPerfTestScript.s3Download({file=config.yml, bucket=test_bucket, path=test_config/config.yml, force=true})
-                  runPerfTestScript.sh(./test.sh perf-test --stack test-single-security-1236-x64 --bundle-manifest tests/jenkins/data/opensearch-1.3.0-bundle.yml --config config.yml  --workload nyc_taxis --test-iters 1 --warmup-iters 1)
+                  runPerfTestScript.sh(./test.sh perf-test --stack test-single-security-1236-x64 --bundle-manifest tests/jenkins/data/opensearch-1.3.0-bundle.yml --config config.yml  --workload nyc_taxis --test-iters 1 --warmup-iters 1 )
                Messages.asBoolean()
                Messages.asBoolean()
                Messages.get([perf-test])

tests/jenkins/jenkinsjob-regression-files/opensearch/perf-test.jenkinsfile.txt

@@ -52,7 +52,7 @@
     )
                   runPerfTestScript.withAWS({role=opensearch-test, roleAccount=dummy_account, duration=900, roleSessionName=jenkins-session}, groovy.lang.Closure)
                      runPerfTestScript.s3Download({file=config.yml, bucket=test_bucket, path=test_config/config.yml, force=true})
-                  runPerfTestScript.sh(./test.sh perf-test --stack test-single-1236-x64 --bundle-manifest tests/jenkins/data/opensearch-1.3.0-non-security-bundle.yml --config config.yml --without-security --workload nyc_taxis --test-iters 1 --warmup-iters 1)
+                  runPerfTestScript.sh(./test.sh perf-test --stack test-single-1236-x64 --bundle-manifest tests/jenkins/data/opensearch-1.3.0-non-security-bundle.yml --config config.yml --without-security --workload nyc_taxis --test-iters 1 --warmup-iters 1 )
                Messages.asBoolean()
                Messages.asBoolean()
                Messages.get([perf-test])