feat: add time based benchmarks

tests/perf/microbenchmarks/time_based/conftest.py

@@ -0,0 +1,85 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import pytest
+import os
+import multiprocessing
+import logging
+from google.cloud import storage
+
+_OBJECT_NAME_PREFIX = "time_based_tests"
+
+
+# def _upload_worker(args):
+#     bucket_name, object_name, object_size = args
+#     storage_client = storage.Client()
+#     bucket = storage_client.bucket(bucket_name)
+#     blob = bucket.blob(object_name)
+
+#     try:
+#         blob.reload()
+#         if blob.size >= object_size:
+#             logging.info(f"Object {object_name} already exists and has the required size.")
+#             return object_name, object_size
+#     except Exception:
+#         pass
+
+#     logging.info(f"Creating object {object_name} of size {object_size} bytes.")
+#     # For large objects, it's better to upload in chunks.
+#     # Using urandom is slow, so for large objects, we will write the same chunk over and over.
+#     chunk_size = 100 * 1024 * 1024  # 100 MiB
+#     data_chunk = os.urandom(chunk_size)
+#     num_chunks = object_size // chunk_size
+#     remaining_bytes = object_size % chunk_size
+
+#     from io import BytesIO
+#     with BytesIO() as f:
+#         for _ in range(num_chunks):
+#             f.write(data_chunk)
+#         if remaining_bytes > 0:
+#             f.write(data_chunk[:remaining_bytes])
+
+#         f.seek(0)
+#         blob.upload_from_file(f, size=object_size)
+
+#     logging.info(f"Finished creating object {object_name}.")
+#     return object_name, object_size
+
+
+# def _create_files(num_files, bucket_name, object_size):
+#     """
+#     Create/Upload objects for benchmarking and return a list of their names.
+#     """
+#     object_names = [f"{_OBJECT_NAME_PREFIX}_{i}" for i in range(num_files)]
+
+#     args_list = [
+#         (bucket_name, object_names[i], object_size) for i in range(num_files)
+#     ]
+
+#     # Don't use a pool to avoid contention writing the same objects.
+#     # The check for existence should make this fast on subsequent runs.
+#     results = [_upload_worker(arg) for arg in args_list]
+
+#     return [r[0] for r in results]
+
+
+@pytest.fixture
+def workload_params(request):
+    params = request.param
+    files_names = [f'fio-go_storage_fio.0.{i}' for i in range(0, params.num_processes)]
+    # files_names = _create_files(
+    #     params.num_processes, # One file per process
+    #     params.bucket_name,
+    #     params.file_size_bytes,
+    # )
+    return params, files_names

tests/perf/microbenchmarks/time_based/reads/config.py

@@ -0,0 +1,102 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import itertools
+import os
+from typing import Dict, List
+
+import yaml
+
+try:
+    from tests.perf.microbenchmarks.time_based.reads.parameters import (
+        TimeBasedReadParameters,
+    )
+except ModuleNotFoundError:
+    from reads.parameters import TimeBasedReadParameters
+
+
+def _get_params() -> Dict[str, List[TimeBasedReadParameters]]:
+    """Generates a dictionary of benchmark parameters for time based read operations."""
+    params: Dict[str, List[TimeBasedReadParameters]] = {}
+    config_path = os.path.join(os.path.dirname(__file__), "config.yaml")
+    with open(config_path, "r") as f:
+        config = yaml.safe_load(f)
+
+    common_params = config["common"]
+    bucket_types = common_params["bucket_types"]
+    file_sizes_mib = common_params["file_sizes_mib"]
+    chunk_sizes_mib = common_params["chunk_sizes_mib"]
+    rounds = common_params["rounds"]
+    duration = common_params["duration"]
+    warmup_duration = common_params["warmup_duration"]
+
+    bucket_map = {
+        "zonal": os.environ.get(
+            "DEFAULT_RAPID_ZONAL_BUCKET",
+            config["defaults"]["DEFAULT_RAPID_ZONAL_BUCKET"],
+        ),
+        "regional": os.environ.get(
+            "DEFAULT_STANDARD_BUCKET", config["defaults"]["DEFAULT_STANDARD_BUCKET"]
+        ),
+    }
+
+    for workload in config["workload"]:
+        workload_name = workload["name"]
+        params[workload_name] = []
+        pattern = workload["pattern"]
+        processes = workload["processes"]
+        coros = workload["coros"]
+
+        # Create a product of all parameter combinations
+        product = itertools.product(
+            bucket_types,
+            file_sizes_mib,
+            chunk_sizes_mib,
+            processes,
+            coros,
+        )
+
+        for (
+            bucket_type,
+            file_size_mib,
+            chunk_size_mib,
+            num_processes,
+            num_coros,
+        ) in product:
+            file_size_bytes = file_size_mib * 1024 * 1024
+            chunk_size_bytes = chunk_size_mib * 1024 * 1024
+            bucket_name = bucket_map[bucket_type]
+
+            num_files = num_processes * num_coros
+
+            # Create a descriptive name for the parameter set
+            name = f"{pattern}_{bucket_type}_{num_processes}p_{file_size_mib}MiB_{chunk_size_mib}MiB"
+
+            params[workload_name].append(
+                TimeBasedReadParameters(
+                    name=name,
+                    workload_name=workload_name,
+                    pattern=pattern,
+                    bucket_name=bucket_name,
+                    bucket_type=bucket_type,
+                    num_coros=num_coros,
+                    num_processes=num_processes,
+                    num_files=num_files,
+                    rounds=rounds,
+                    chunk_size_bytes=chunk_size_bytes,
+                    file_size_bytes=file_size_bytes,
+                    duration=duration,
+                    warmup_duration=warmup_duration,
+                )
+            )
+    return params

tests/perf/microbenchmarks/time_based/reads/config.yaml

@@ -0,0 +1,26 @@
+common:
+  bucket_types:
+    - "zonal"
+  file_sizes_mib:
+    - 10240 # 10GiB
+  chunk_sizes_mib: [1, 16, 100, 200] # 16MiB
+  rounds: 1
+  duration: 60 # seconds
+  warmup_duration: 5 # seconds
+
+workload:
+  ############# multi process multi coroutine #########
+  - name: "read_seq_multi_process"
+    pattern: "seq"
+    coros: [1]
+    processes: [1, 48]
+
+
+  - name: "read_rand_multi_process"
+    pattern: "rand"
+    coros: [1]
+    processes: [1, 48]
+
+defaults:
+  DEFAULT_RAPID_ZONAL_BUCKET: "chandrasiri-benchmarks-zb"
+  DEFAULT_STANDARD_BUCKET: "chandrasiri-benchmarks-rb"

tests/perf/microbenchmarks/time_based/reads/parameters.py

@@ -0,0 +1,22 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from dataclasses import dataclass
+from tests.perf.microbenchmarks.parameters import IOBenchmarkParameters
+
+
+@dataclass
+class TimeBasedReadParameters(IOBenchmarkParameters):
+    pattern: str
+    duration: int
+    warmup_duration: int