[Perf] Support multiple test proxies (Azure#21203)

- Some scenarios require multiple test-proxy URLs to bottleneck client
- test-proxy URLs are assigned round-robin to parallel test instances
- Ported from Azure/azure-sdk-for-net#24265

Author: mikeharder

doc/dev/perfstress_tests.md

@@ -82,7 +82,7 @@ The framework has a series of common command line options built in:
 - `-w --warm-up=5` Number of seconds to spend warming up the connection before measuring begins. Default is 5.
 - `--sync` Whether to run the tests in sync or async. Default is False (async).
 - `--no-cleanup` Whether to keep newly created resources after test run. Default is False (resources will be deleted).
-- `-x --test-proxy` Whether to run the tests against the test proxy server. Specfiy the URL for the proxy endpoint (e.g. "https://localhost:5001").
+- `-x --test-proxies` Whether to run the tests against the test proxy server. Specify the URL(s) for the proxy endpoint(s) (e.g. "https://localhost:5001").
 - `--profile` Whether to run the perftest with cProfile. If enabled (default is False), the output file of the **last completed single iteration** will be written to the current working directory in the format `"cProfile-<TestClassName>-<TestID>-<sync/async>.pstats"`.
 
 

sdk/storage/azure-storage-blob/tests/perfstress_tests/README.md

@@ -44,7 +44,7 @@ These options are available for all perf tests:
 - `-w --warm-up=5` Number of seconds to spend warming up the connection before measuring begins. Default is 5.
 - `--sync` Whether to run the tests in sync or async. Default is False (async). This flag must be used for Storage legacy tests, which do not support async.
 - `--no-cleanup` Whether to keep newly created resources after test run. Default is False (resources will be deleted).
-- `-x --test-proxy` Whether to run the tests against the test proxy server. Specfiy the URL for the proxy endpoint (e.g. "https://localhost:5001"). WARNING: When using with Legacy tests - only HTTPS is supported.
+- `-x --test-proxies` Whether to run the tests against the test proxy server. Specfiy the URL(s) for the proxy endpoint(s) (e.g. "https://localhost:5001"). WARNING: When using with Legacy tests - only HTTPS is supported.
 - `--profile` Whether to run the perftest with cProfile. If enabled (default is False), the output file of the **last completed single iteration** will be written to the current working directory in the format `"cProfile-<TestClassName>-<TestID>-<sync/async>.pstats"`.
 
 ### Common Blob command line options

sdk/storage/azure-storage-blob/tests/perfstress_tests/T1_legacy_tests/_test_base_legacy.py

@@ -37,7 +37,7 @@ def __init__(self, arguments):
         super().__init__(arguments)
         connection_string = self.get_from_env("AZURE_STORAGE_CONNECTION_STRING")
         session = None
-        if self.args.test_proxy:
+        if self.args.test_proxies:
             session = requests.Session()
             session.verify = False
         if not _LegacyServiceTest.service_client or self.args.no_client_share:
@@ -50,7 +50,7 @@ def __init__(self, arguments):
         self.async_service_client = None
         self.service_client = _LegacyServiceTest.service_client
 
-        if self.args.test_proxy:
+        if self.args.test_proxies:
             self.service_client.request_callback = functools.partial(
                 test_proxy_callback,
                 self._test_proxy_policy

sdk/storage/azure-storage-blob/tests/perfstress_tests/_test_base.py

@@ -20,7 +20,7 @@ class _ServiceTest(PerfStressTest):
     def __init__(self, arguments):
         super().__init__(arguments)
         connection_string = self.get_from_env("AZURE_STORAGE_CONNECTION_STRING")
-        if self.args.test_proxy:
+        if self.args.test_proxies:
             self._client_kwargs['_additional_pipeline_policies'] = self._client_kwargs['per_retry_policies']
         self._client_kwargs['max_single_put_size'] = self.args.max_put_size
         self._client_kwargs['max_block_size'] = self.args.max_block_size

tools/azure-devtools/src/azure_devtools/perfstress_tests/perf_stress_runner.py

@@ -97,7 +97,8 @@ def _parse_args(self):
             "--profile", action="store_true", help="Run tests with profiler.  Default is False.", default=False
         )
         per_test_arg_parser.add_argument(
-            "-x", "--test-proxy", help="URI of TestProxy Server"
+            "-x", "--test-proxies", help="URIs of TestProxy Servers (separated by ';')",
+            type=lambda s: s.split(';')
         )
 
         # Per-test args
@@ -142,7 +143,7 @@ async def start(self):
                     await asyncio.gather(*[test.setup() for test in tests])
                     self.logger.info("")
 
-                    if self.per_test_args.test_proxy:
+                    if self.per_test_args.test_proxies:
                         self.logger.info("=== Record and Start Playback ===")
                         await asyncio.gather(*[test.record_and_start_playback() for test in tests])
                         self.logger.info("")
@@ -162,7 +163,7 @@ async def start(self):
                 except Exception as e:
                     print("Exception: " + str(e))
                 finally:
-                    if self.per_test_args.test_proxy:
+                    if self.per_test_args.test_proxies:
                         self.logger.info("=== Stop Playback ===")
                         await asyncio.gather(*[test.stop_playback() for test in tests])
                         self.logger.info("")

tools/azure-devtools/src/azure_devtools/perfstress_tests/perf_stress_test.py

@@ -4,6 +4,7 @@
 # --------------------------------------------------------------------------------------------
 
 import os
+import threading
 import aiohttp
 
 from urllib.parse import urljoin
@@ -21,15 +22,22 @@ class PerfStressTest:
     """
 
     args = {}
+    _global_parallel_index_lock = threading.Lock()
+    _global_parallel_index = 0
 
     def __init__(self, arguments):
         self.args = arguments
         self._session = None
+        self._test_proxy = None
         self._test_proxy_policy = None
         self._client_kwargs = {}
         self._recording_id = None
 
-        if self.args.test_proxy:
+        with PerfStressTest._global_parallel_index_lock:
+            self._parallel_index = PerfStressTest._global_parallel_index
+            PerfStressTest._global_parallel_index += 1
+
+        if self.args.test_proxies:
             self._session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(verify_ssl=False))
 
             # SSL will be disabled for the test proxy requests, so suppress warnings
@@ -38,7 +46,8 @@ def __init__(self, arguments):
             warnings.simplefilter('ignore', InsecureRequestWarning)
 
             # Add policy to redirect requests to the test proxy
-            self._test_proxy_policy = PerfTestProxyPolicy(self.args.test_proxy)
+            self._test_proxy = self.args.test_proxies[self._parallel_index % len(self.args.test_proxies)]
+            self._test_proxy_policy = PerfTestProxyPolicy(self._test_proxy)
             self._client_kwargs['per_retry_policies'] = [self._test_proxy_policy]
 
     async def global_setup(self):
@@ -74,7 +83,7 @@ async def stop_playback(self):
             "x-recording-id": self._recording_id,
             "x-purge-inmemory-recording": "true"
         }
-        url = urljoin(self.args.test_proxy, "/playback/stop")
+        url = urljoin(self._test_proxy, "/playback/stop")
         async with self._session.post(url, headers=headers) as resp:
             assert resp.status == 200
 
@@ -98,20 +107,20 @@ async def run_async(self):
         raise Exception("run_async must be implemented for {}".format(self.__class__.__name__))
 
     async def _start_recording(self):
-        url = urljoin(self.args.test_proxy, "/record/start")
+        url = urljoin(self._test_proxy, "/record/start")
         async with self._session.post(url) as resp:
             assert resp.status == 200
             self._recording_id = resp.headers["x-recording-id"]
 
     async def _stop_recording(self):
         headers = {"x-recording-id": self._recording_id}
-        url = urljoin(self.args.test_proxy, "/record/stop")
+        url = urljoin(self._test_proxy, "/record/stop")
         async with self._session.post(url, headers=headers) as resp:
             assert resp.status == 200
 
     async def _start_playback(self):
         headers = {"x-recording-id": self._recording_id}
-        url = urljoin(self.args.test_proxy, "/playback/start")
+        url = urljoin(self._test_proxy, "/playback/start")
         async with self._session.post(url, headers=headers) as resp:
             assert resp.status == 200
             self._recording_id = resp.headers["x-recording-id"]