diff --git a/osbenchmark/benchmark.py b/osbenchmark/benchmark.py index 73ee46527..ff5527e89 100644 --- a/osbenchmark/benchmark.py +++ b/osbenchmark/benchmark.py @@ -470,7 +470,7 @@ def add_workload_source(subparser): "(default: localhost:9200).", default="") # actually the default is pipeline specific and it is set later test_execution_parser.add_argument( - "--worker-ips", + "--load-worker-coordinator-hosts", help="Define a comma-separated list of hosts which should generate load (default: localhost).", default="localhost") test_execution_parser.add_argument( @@ -859,8 +859,8 @@ def dispatch_sub_command(arg_parser, args, cfg): cfg.add( config.Scope.applicationOverride, "worker_coordinator", - "worker_ips", - opts.csv_to_list(args.worker_ips)) + "load_worker_coordinator_hosts", + opts.csv_to_list(args.load_worker_coordinator_hosts)) cfg.add(config.Scope.applicationOverride, "workload", "test.mode.enabled", args.test_mode) configure_workload_params(arg_parser, args, cfg) configure_connection_params(arg_parser, args, cfg) diff --git a/osbenchmark/worker_coordinator/runner.py b/osbenchmark/worker_coordinator/runner.py index 0e76427d7..51ad3f735 100644 --- a/osbenchmark/worker_coordinator/runner.py +++ b/osbenchmark/worker_coordinator/runner.py @@ -77,7 +77,7 @@ def register_default_runners(): register_runner(workload.OperationType.RestoreSnapshot, RestoreSnapshot(), async_runner=True) # We treat the following as administrative commands and thus already start to wrap them in a retry. register_runner(workload.OperationType.ClusterHealth, Retry(ClusterHealth()), async_runner=True) - register_runner(workload.OperationType.CreateIngestPipeline, Retry(CreateIngestPipeline()), async_runner=True) + register_runner(workload.OperationType.PutPipeline, Retry(PutPipeline()), async_runner=True) register_runner(workload.OperationType.Refresh, Retry(Refresh()), async_runner=True) register_runner(workload.OperationType.CreateIndex, Retry(CreateIndex()), async_runner=True) register_runner(workload.OperationType.DeleteIndex, Retry(DeleteIndex()), async_runner=True) @@ -1231,7 +1231,7 @@ def __repr__(self, *args, **kwargs): return "cluster-health" -class CreateIngestPipeline(Runner): +class PutPipeline(Runner): async def __call__(self, opensearch, params): await opensearch.ingest.put_pipeline(id=mandatory(params, "id", self), body=mandatory(params, "body", self), @@ -1240,7 +1240,7 @@ async def __call__(self, opensearch, params): ) def __repr__(self, *args, **kwargs): - return "create-ingest-pipeline" + return "put-pipeline" # TODO: refactor it after python client support search pipeline https://github.com/opensearch-project/opensearch-py/issues/474 class CreateSearchPipeline(Runner): diff --git a/osbenchmark/worker_coordinator/worker_coordinator.py b/osbenchmark/worker_coordinator/worker_coordinator.py index ad8b77dc8..bcfa96d31 100644 --- a/osbenchmark/worker_coordinator/worker_coordinator.py +++ b/osbenchmark/worker_coordinator/worker_coordinator.py @@ -529,7 +529,7 @@ def __init__(self, target, config, os_client_factory_class=client.OsClientFactor self.workload = None self.test_procedure = None self.metrics_store = None - self.worker_ips = [] + self.load_worker_coordinator_hosts = [] self.workers = [] # which client ids are assigned to which workers? self.clients_per_worker = {} @@ -637,7 +637,7 @@ def prepare_benchmark(self, t): # are not useful and attempts to connect to a non-existing cluster just lead to exception traces in logs. self.prepare_telemetry(os_clients, enable=not uses_static_responses) - for host in self.config.opts("worker_coordinator", "worker_ips"): + for host in self.config.opts("worker_coordinator", "load_worker_coordinator_hosts"): host_config = { # for simplicity we assume that all benchmark machines have the same specs "cores": num_cores(self.config) @@ -647,9 +647,9 @@ def prepare_benchmark(self, t): else: host_config["host"] = host - self.worker_ips.append(host_config) + self.load_worker_coordinator_hosts.append(host_config) - self.target.prepare_workload([h["host"] for h in self.worker_ips], self.config, self.workload) + self.target.prepare_workload([h["host"] for h in self.load_worker_coordinator_hosts], self.config, self.workload) def start_benchmark(self): self.logger.info("Benchmark is about to start.") @@ -670,7 +670,7 @@ def start_benchmark(self): if allocator.clients < 128: self.logger.info("Allocation matrix:\n%s", "\n".join([str(a) for a in self.allocations])) - worker_assignments = calculate_worker_assignments(self.worker_ips, allocator.clients) + worker_assignments = calculate_worker_assignments(self.load_worker_coordinator_hosts, allocator.clients) worker_id = 0 for assignment in worker_assignments: host = assignment["host"] diff --git a/osbenchmark/workload/workload.py b/osbenchmark/workload/workload.py index d9b296846..1dc4ff530 100644 --- a/osbenchmark/workload/workload.py +++ b/osbenchmark/workload/workload.py @@ -590,7 +590,7 @@ class OperationType(Enum): # administrative actions ForceMerge = 1001 ClusterHealth = 1002 - CreateIngestPipeline = 1003 + PutPipeline = 1003 Refresh = 1004 CreateIndex = 1005 DeleteIndex = 1006 @@ -653,8 +653,8 @@ def from_hyphenated_string(cls, v): return OperationType.Bulk elif v == "raw-request": return OperationType.RawRequest - elif v == "create-ingest-pipeline": - return OperationType.CreateIngestPipeline + elif v == "put-pipeline": + return OperationType.PutPipeline elif v == "refresh": return OperationType.Refresh elif v == "create-index": diff --git a/tests/worker_coordinator/runner_test.py b/tests/worker_coordinator/runner_test.py index 728a663fe..01a2a3c3f 100644 --- a/tests/worker_coordinator/runner_test.py +++ b/tests/worker_coordinator/runner_test.py @@ -2737,13 +2737,13 @@ async def test_query_vector_search_with_custom_id_field_inside_source(self, open ) -class CreateIngestPipelineRunnerTests(TestCase): +class PutPipelineRunnerTests(TestCase): @mock.patch("opensearchpy.OpenSearch") @run_async async def test_create_pipeline(self, opensearch): opensearch.ingest.put_pipeline.return_value = as_future() - r = runner.CreateIngestPipeline() + r = runner.PutPipeline() params = { "id": "rename", @@ -2769,16 +2769,13 @@ async def test_create_pipeline(self, opensearch): async def test_param_body_mandatory(self, opensearch): opensearch.ingest.put_pipeline.return_value = as_future() - r = runner.CreateIngestPipeline() + r = runner.PutPipeline() params = { "id": "rename" } with self.assertRaisesRegex(exceptions.DataError, - "Parameter source " - "for operation 'create-ingest-pipeline' " - "did not provide the " - "mandatory parameter 'body'. " + "Parameter source for operation 'put-pipeline' did not provide the mandatory parameter 'body'. " "Add it to your parameter source and try again."): await r(opensearch, params) @@ -2789,15 +2786,13 @@ async def test_param_body_mandatory(self, opensearch): async def test_param_id_mandatory(self, opensearch): opensearch.ingest.put_pipeline.return_value = as_future() - r = runner.CreateIngestPipeline() + r = runner.PutPipeline() params = { "body": {} } with self.assertRaisesRegex(exceptions.DataError, - "Parameter source for " - "operation 'create-ingest-pipeline' did" - " not provide the mandatory parameter 'id'. " + "Parameter source for operation 'put-pipeline' did not provide the mandatory parameter 'id'. " "Add it to your parameter source and try again."): await r(opensearch, params) diff --git a/tests/worker_coordinator/worker_coordinator_test.py b/tests/worker_coordinator/worker_coordinator_test.py index 31391f3c4..bf7865288 100644 --- a/tests/worker_coordinator/worker_coordinator_test.py +++ b/tests/worker_coordinator/worker_coordinator_test.py @@ -110,7 +110,7 @@ def setUp(self): self.cfg.add(config.Scope.application, "client", "hosts", WorkerCoordinatorTests.Holder(all_hosts={"default": ["localhost:9200"]})) self.cfg.add(config.Scope.application, "client", "options", WorkerCoordinatorTests.Holder(all_client_options={"default": {}})) - self.cfg.add(config.Scope.application, "worker_coordinator", "worker_ips", ["localhost"]) + self.cfg.add(config.Scope.application, "worker_coordinator", "load_worker_coordinator_hosts", ["localhost"]) self.cfg.add(config.Scope.application, "results_publishing", "datastore.type", "in-memory") default_test_procedure = workload.TestProcedure("default", default=True, schedule=[ @@ -135,7 +135,7 @@ def create_test_worker_coordinator_target(self): @mock.patch("osbenchmark.utils.net.resolve") def test_start_benchmark_and_prepare_workload(self, resolve): # override load worker_coordinator host - self.cfg.add(config.Scope.applicationOverride, "worker_coordinator", "worker_ips", ["10.5.5.1", "10.5.5.2"]) + self.cfg.add(config.Scope.applicationOverride, "worker_coordinator", "load_worker_coordinator_hosts", ["10.5.5.1", "10.5.5.2"]) resolve.side_effect = ["10.5.5.1", "10.5.5.2"] target = self.create_test_worker_coordinator_target()