From a78899f3e17df3ff2c8ebe20b9751ebce33274e0 Mon Sep 17 00:00:00 2001 From: Lukas Phaf Date: Tue, 9 Jan 2024 14:40:03 +0100 Subject: [PATCH] Load test support for setting up message "per variable". --- datastore/load-test/locustfile_write.py | 5 ++++- datastore/load-test/netcdf_file_to_requests.py | 13 ++++++++++--- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/datastore/load-test/locustfile_write.py b/datastore/load-test/locustfile_write.py index d71202f..cab1969 100644 --- a/datastore/load-test/locustfile_write.py +++ b/datastore/load-test/locustfile_write.py @@ -76,7 +76,7 @@ class IngestionGrpcUser(grpc_user.GrpcUser): stub_class = dstore_grpc.DatastoreStub wait_time = between(1.5, 2.5) user_nr = 0 - dummy_observations_all_stations = generate_dummy_requests_from_netcdf_per_station_per_timestamp(file_path) + dummy_observations_all_stations = generate_dummy_requests_from_netcdf_per_station_per_timestamp(file_path, False) weight = 7 def on_start(self): @@ -84,6 +84,9 @@ def on_start(self): self.dummy_observations_per_station = IngestionGrpcUser.dummy_observations_all_stations[ IngestionGrpcUser.user_nr ] + print(f"Number of messages to send: {len(self.dummy_observations_per_station)}") + print(f'Number of observations in first: {len(self.dummy_observations_per_station[0]["observations"])}') + print(f'Number of observations in last: {len(self.dummy_observations_per_station[-1]["observations"])}') IngestionGrpcUser.user_nr += 1 self.index = 0 diff --git a/datastore/load-test/netcdf_file_to_requests.py b/datastore/load-test/netcdf_file_to_requests.py index de2b335..831262c 100644 --- a/datastore/load-test/netcdf_file_to_requests.py +++ b/datastore/load-test/netcdf_file_to_requests.py @@ -67,7 +67,9 @@ def timerange(start_time, end_time, interval_minutes): current_time += timedelta(minutes=interval_minutes) -def generate_dummy_requests_from_netcdf_per_station_per_timestamp(file_path: Path | str) -> Tuple[List, List]: +def generate_dummy_requests_from_netcdf_per_station_per_timestamp( + file_path: Path | str, per_variable: bool = False +) -> Tuple[List, List]: print("Starting with creating the time series and observations requests.") create_requests_start = perf_counter() obs_per_station = [] @@ -90,7 +92,7 @@ def generate_dummy_requests_from_netcdf_per_station_per_timestamp(file_path: Pat ts.FromDatetime(generated_timestamp) for param_id in knmi_parameter_names: param = station_slice[param_id] - obs_value = station_slice[param_id].data[idx] # Use 10 minute data value for each + obs_value = station_slice[param_id].data[idx] # Use 10-minute data value for each obs_value = 0 if math.isnan(obs_value) else obs_value # dummy data so obs_value doesn't matter ts_mdata = dstore.TSMetadata( platform=station_id, @@ -107,7 +109,12 @@ def generate_dummy_requests_from_netcdf_per_station_per_timestamp(file_path: Pat ) observation = dstore.Metadata1(ts_mdata=ts_mdata, obs_mdata=obs_mdata) obs_per_parameter.append(observation) - obs_per_timestamp.append({"time": generated_timestamp, "observations": obs_per_parameter}) + if per_variable: + obs_per_timestamp.append({"time": generated_timestamp, "observations": obs_per_parameter}) + obs_per_parameter = [] + + if not per_variable: + obs_per_timestamp.append({"time": generated_timestamp, "observations": obs_per_parameter}) obs_per_station.append(obs_per_timestamp) print("Finished creating the time series and observation requests " f"{perf_counter() - create_requests_start}.")