Repository: AgnostiqHQ/covalent (language: python)

Issue:
create benchmark.yml workflow in core covalent



Discussion / hints:
Acceptance criteria

- [x] Discuss to see if this can be done in existing tests.yml workflow

- [ ] If not, setup a new workflow that to run the stress tests, locust load tests on a nightly schedule

- [ ] Upload benchmark results to S3 for further post-processing
Benchmark.yml
* Checkout develop
* pip isntall -r tests/requirements.txt
* run  the stress.ipynb notebook (jupyter --nbconvert notebook --execute tests/stress_tests/tests.ipynb)
* upload the metrics.json file to S3

Fix patch:
diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml
new file mode 100644
index 000000000..d5f0ed4a9
--- /dev/null
+++ b/.github/workflows/benchmark.yml
@@ -0,0 +1,92 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+name: benchmark
+
+on:
+  workflow_call:
+  workflow_dispatch:
+    inputs:
+      commit_sha:
+        description: "Commit SHA used for testing"
+        type: string
+        required: true
+
+jobs:
+  benchmarks:
+    runs-on: ubuntu-latest
+    env:
+      COVALENT_SERVER_IFACE_ANY: true
+    steps:
+      - name: Check out head
+        if: github.event_name == 'workflow_call'
+        uses: actions/checkout@v3
+        with:
+          persist-credentials: false
+
+      - name: Check out SHA
+        if: github.event_name == 'workflow_dispatch'
+        uses: actions/checkout@v3
+        with:
+          persist-credentials: false
+          fetch-depth: 0
+          ref: ${{ inputs.commit_sha }}
+
+      - name: Setup Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.8"
+
+      - name: Install Covalent
+        run: python setup.py sdist && pip install dist/*.tar.gz
+
+      - name: Install test requirements
+        run: pip install --no-cache-dir -r ./tests/requirements.txt
+
+      - name: Start Covalent dispatcher server
+        run: covalent start -d --ignore-migrations
+
+      - name: Run stress benchmarks
+        run: python -m pytest -vv tests/stress_tests/benchmarks
+
+      - name: Run Locust load tests
+        run: >
+          python -m locust
+          --user $(python -c 'import os; print(os.cpu_count())')
+          --spawn-rate 3
+          --run-time 10m
+          --csv=loadtest_results
+          --headless
+          -f ./load_tests/locustfiles
+        working-directory: ./tests
+
+      - name: Create tests result archive
+        run: |
+          tar -zcvf $GITHUB_SHA.tar.gz metrics*.log ./tests/loadtest_results*.csv
+
+      - name: Configure AWS credentials
+        uses: aws-actions/configure-aws-credentials@v1
+        with:
+          aws-access-key-id: ${{ secrets.AWS_KEY_ID }}
+          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+          aws-region: us-east-1
+
+      - name: Upload benchmark results to S3
+        run: aws s3 cp $GITHUB_SHA.tar.gz s3://${{ secrets.BENCHMARK_BUCKET }}/$GITHUB_SHA.tar.gz
diff --git a/.gitignore b/.gitignore
index c9ece6b78..e278f9841 100644
--- a/.gitignore
+++ b/.gitignore
@@ -72,7 +72,8 @@
 !README.md
 !VERSION
 
-
+# Ignore notebook checkpoints
+**/.ipynb_checkpoints
 
 # Ignore .sonarwork files
 covalent_ui/api/.scannerwork/.sonar_lock
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 14ce933ac..91ae5a94d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [UNRELEASED]
 
+### Operations
+
+- Adding `benchmark.yml` workflow
+
+### Tests
+
+- Adding scripts in `tests/stress_tests/benchmarks`
+
 ## [0.188.0] - 2022-08-31
 
 ### Authors
@@ -95,7 +103,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Co-authored-by: Alejandro Esquivel <ae@alejandro.ltd>
 - Venkat Bala <venkat@agnostiq.ai>
 
-
 ### Added
 
 - Adding `load_tests` subdirectory to tests to facilitate execution of Covalent benchmarks during nightly runs
diff --git a/covalent/_shared_files/metrics.py b/covalent/_shared_files/metrics.py
new file mode 100644
index 000000000..56e76f0bf
--- /dev/null
+++ b/covalent/_shared_files/metrics.py
@@ -0,0 +1,63 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+import platform
+
+from pydantic import BaseModel
+
+
+class PlatformMetadata(BaseModel):
+    """Information about the platform used to run the benchmarks"""
+
+    arch: str = platform.architecture()
+    system: str = platform.system()
+    machine: str = platform.machine()
+    os: str = platform.node()
+    python_version: str = platform.python_version()
+
+
+class PerformanceMetrics(BaseModel):
+    """Performance metrics to be gathered"""
+
+    workflow_runtime: float
+    covalent_runtime: float
+    covalent_speedup: float
+    covalent_overhead: float
+    covalent_disk_io_time: float = 0.0
+    covalent_network_io_time: float = 0.0
+    covalent_fraction_disk_io: float = 0.0
+    covalent_fraction_idle: float = 0.0
+    covalent_fraction_user_mode: float = 0.0
+    covalent_fraction_system_mode: float = 0.0
+    covalent_dispatch_latency: float = 0.0
+    covalent_dispatch_throughput: float = 0.0
+    covalent_total_db_reads: float = 0.0
+    covalent_total_db_writes: float = 0.0
+    covalent_electron_throughput: float = 0.0
+    covalent_electron_latency: float = 0.0
+
+
+class WorkflowBenchmarkResult(BaseModel):
+    """Base object representing the results from a single workflow benchmark run"""
+
+    run_id: int
+    workflow_name: str
+    metadata: PlatformMetadata = PlatformMetadata()
+    metrics: PerformanceMetrics


Test patch:
diff --git a/tests/covalent_tests/shared_files/metrics_test.py b/tests/covalent_tests/shared_files/metrics_test.py
new file mode 100644
index 000000000..94ffb41b3
--- /dev/null
+++ b/tests/covalent_tests/shared_files/metrics_test.py
@@ -0,0 +1,43 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""Unit tests for metrics module."""
+
+from covalent._shared_files.metrics import (
+    PerformanceMetrics,
+    PlatformMetadata,
+    WorkflowBenchmarkResult,
+)
+
+
+def test_platform_metdata():
+    """Test the platform metadata object."""
+
+    pmd = PlatformMetadata()
+    assert pmd.arch is not None
+    assert pmd.system is not None
+    assert pmd.machine is not None
+    assert pmd.os is not None
+    assert pmd.python_version is not None
+    print(pmd.arch)
+    print(pmd.system)
+    print(pmd.machine)
+    print(pmd.os)
+    print(pmd.python_version)
diff --git a/tests/stress_tests/benchmarks/__init__.py b/tests/stress_tests/benchmarks/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/stress_tests/benchmarks/conftest.py b/tests/stress_tests/benchmarks/conftest.py
new file mode 100644
index 000000000..8bf37cf56
--- /dev/null
+++ b/tests/stress_tests/benchmarks/conftest.py
@@ -0,0 +1,61 @@
+import logging
+import time
+from datetime import datetime
+
+import pytest
+
+import covalent as ct
+from covalent._shared_files.metrics import PerformanceMetrics, WorkflowBenchmarkResult
+
+
+def run_benchmark(benchmark_id: int, lattice, *args, **kwargs):
+    """
+    Benchmark the lattice using against a Covalent server
+    """
+    # Run as normal function
+    start = time.time()
+    lattice(*args, **kwargs)
+    end = time.time()
+    normal_runtime = end - start
+
+    # Dispatch to Covalent
+    dispatch_id = ct.dispatch(lattice)(*args, **kwargs)
+    dispatch_result = ct.get_result(dispatch_id=dispatch_id, wait=True)
+    covalent_execution_time = (
+        dispatch_result.end_time - dispatch_result.start_time
+    ).total_seconds()
+
+    # Covalent overhead (relative increase)
+    overhead = (covalent_execution_time - normal_runtime) / normal_runtime
+
+    metrics = PerformanceMetrics(
+        workflow_runtime=normal_runtime,
+        covalent_runtime=covalent_execution_time,
+        covalent_speedup=normal_runtime / covalent_execution_time,
+        covalent_overhead=overhead,
+    )
+
+    return (
+        WorkflowBenchmarkResult(
+            run_id=benchmark_id, workflow_name=lattice.__name__, metrics=metrics
+        ),
+        dispatch_result.status,
+    )
+
+
+@pytest.fixture(scope="module")
+def benchmark():
+    logger = logging.getLogger("metricsLogger")
+    logger.setLevel(logging.DEBUG)
+
+    fileHandler = logging.FileHandler("metrics.log")
+    fileHandler.setFormatter(logging.DEBUG)
+
+    formatter = logging.Formatter("%(asctime)s-%(name)s-%(levelname)s-%(message)s")
+    fileHandler.setFormatter(formatter)
+
+    logger.addHandler(fileHandler)
+    yield (
+        run_benchmark,
+        logger,
+    )
diff --git a/tests/stress_tests/benchmarks/cpu_intensive_workflows_test.py b/tests/stress_tests/benchmarks/cpu_intensive_workflows_test.py
new file mode 100644
index 000000000..acded928d
--- /dev/null
+++ b/tests/stress_tests/benchmarks/cpu_intensive_workflows_test.py
@@ -0,0 +1,43 @@
+import random
+
+import pytest
+
+import covalent as ct
+
+
+@pytest.mark.parametrize("iteration", range(5))
+def test_benchmark_primality_test(benchmark, iteration):
+    run_benchmark = benchmark[0]
+    logger = benchmark[1]
+
+    @ct.electron
+    def is_prime(n: int) -> bool:
+        """Primality test using 6k+-1 optimization."""
+        if n <= 3:
+            return n > 1
+        if not n % 2 or not n % 3:
+            return False
+        i = 5
+        stop = int(n**0.5)
+        while i <= stop:
+            if not n % i or not n % (i + 2):
+                return False
+            i += 6
+        return True
+
+    @ct.lattice
+    def primality_tests(nums_to_test):
+        res = []
+        for i in nums_to_test:
+            entry = {}
+            entry["num"] = i
+            entry["is_prime"] = is_prime(i)
+            res.append(entry)
+        return res
+
+    nums_to_test = [random.randint(1000, 10000) for i in range(50)]
+
+    results, status = run_benchmark(iteration, primality_tests, *[nums_to_test])
+    logger.debug(results.dict())
+
+    assert status == "COMPLETED"
diff --git a/tests/stress_tests/benchmarks/memory_intensive_workflow_test.py b/tests/stress_tests/benchmarks/memory_intensive_workflow_test.py
new file mode 100644
index 000000000..fae0c3cf6
--- /dev/null
+++ b/tests/stress_tests/benchmarks/memory_intensive_workflow_test.py
@@ -0,0 +1,32 @@
+import numpy as np
+import pytest
+
+import covalent as ct
+
+
+@pytest.mark.parametrize("iteration", range(5))
+def test_benchmark_matrix_multiplication(benchmark, iteration):
+    run_benchmark = benchmark[0]
+    logger = benchmark[1]
+    arraysizes = [128, 256, 512, 1024]
+
+    @ct.electron
+    def create_matrix(arraysize: int):
+        return np.random.random((arraysize, arraysize))
+
+    @ct.electron
+    def matrix_multiply(a: np.ndarray, b: np.ndarray):
+        return np.matmul(a, b)
+
+    @ct.lattice
+    def matrix_multiplication(arraysizes):
+        for arraysize in arraysizes:
+            a = create_matrix(arraysize)
+            b = create_matrix(arraysize)
+            matrix_multiply(a, b)
+        arraysizes = [256, 512, 1024]
+
+    results, status = run_benchmark(iteration, matrix_multiplication, *[arraysizes])
+    logger.debug(results.dict())
+
+    assert status == "COMPLETED"
diff --git a/tests/stress_tests/benchmarks/parallel_workflows_test.py b/tests/stress_tests/benchmarks/parallel_workflows_test.py
new file mode 100644
index 000000000..acd93a22e
--- /dev/null
+++ b/tests/stress_tests/benchmarks/parallel_workflows_test.py
@@ -0,0 +1,29 @@
+import secrets
+import tempfile
+
+import pytest
+
+import covalent as ct
+
+
+@pytest.mark.parametrize("iteration", range(5))
+def test_benchmark_parallel_fileio_test(benchmark, iteration):
+    run_benchmark, logger = benchmark
+
+    @ct.electron
+    def create_delete_tempfile():
+        fp = tempfile.NamedTemporaryFile(delete=True)
+        # thousand lines per file
+        for i in range(1000):
+            fp.write(secrets.token_bytes(16384))
+        fp.close()
+
+    @ct.lattice
+    def parallel_fileio(N: int):
+        for i in range(N):
+            create_delete_tempfile()
+
+    result, status = run_benchmark(iteration, parallel_fileio, *[50])
+    logger.debug(result.dict())
+
+    assert status == "COMPLETED"
diff --git a/tests/stress_tests/benchmarks/two_node_workflows_test.py b/tests/stress_tests/benchmarks/two_node_workflows_test.py
new file mode 100644
index 000000000..b9fe429f8
--- /dev/null
+++ b/tests/stress_tests/benchmarks/two_node_workflows_test.py
@@ -0,0 +1,29 @@
+import pytest
+
+import covalent as ct
+
+
+@pytest.mark.parametrize("iteration", range(5))
+def test_add_multiply_workflow(benchmark, iteration: int):
+    print(iteration)
+    run_benchmark = benchmark[0]
+    logger = benchmark[1]
+
+    @ct.electron
+    def add(x, y):
+        return x + y
+
+    @ct.electron
+    def multiply(x, y):
+        return x * y
+
+    @ct.lattice
+    def add_multiply_workflow(x, y):
+        r1 = add(x, y)
+        r2 = multiply(r1, y)
+        return r2
+
+    metrics, dispatch_status = run_benchmark(iteration, add_multiply_workflow, *[1, 2])
+    logger.debug(metrics.dict())
+
+    assert dispatch_status == "COMPLETED"
diff --git a/tests/stress_tests/benchmarks/vertical_workflows_test.py b/tests/stress_tests/benchmarks/vertical_workflows_test.py
new file mode 100644
index 000000000..cadf9aa1e
--- /dev/null
+++ b/tests/stress_tests/benchmarks/vertical_workflows_test.py
@@ -0,0 +1,51 @@
+import time
+
+import pytest
+
+import covalent as ct
+
+
+@pytest.mark.parametrize("iteration", range(5))
+def test_benchmark_fully_vertical_add_workflow(benchmark, iteration):
+    run_benchmark, logger = benchmark
+
+    @ct.electron
+    def add(x: int, y: int):
+        return x + y
+
+    @ct.lattice
+    def vertical_add_workflow(N: int):
+        for i in range(N):
+            if i == 0:
+                r1 = add(1, 1)
+            else:
+                r1 = add(r1, 1)
+        return r1
+
+    result, status = run_benchmark(iteration, vertical_add_workflow, *[50])
+    logger.debug(result.dict())
+
+    assert status == "COMPLETED"
+
+
+@pytest.mark.parametrize("iteration", range(5))
+def test_benchmark_fully_vertical_multiply_workflow(benchmark, iteration):
+    run_benchmark, logger = benchmark
+
+    @ct.electron
+    def multiply(x: int, y: int):
+        return x + y
+
+    @ct.lattice
+    def vertical_multiply_workflow(N: int):
+        for i in range(N):
+            if i == 0:
+                r1 = multiply(2, 2)
+            else:
+                r1 = multiply(r1, 2)
+        return r1
+
+    result, status = run_benchmark(iteration, vertical_multiply_workflow, *[50])
+    logger.debug(result.dict())
+
+    assert status == "COMPLETED"