It enables the new multipart transfer manager inside track downloader.

docs/configuration.rst

@@ -140,8 +140,8 @@ Define a secure connection to an Elastic Cloud Serverless project::
 storage
 ~~~~~~~
 
-This section defines how client is configured to transfer corpus files. The main
-advantages of this downloader implementation are:
+This section defines how client is configured to transfer corpus files. The main advantages of this downloader
+implementation are:
 
 * It supports configuring multiple mirror URLs. The client will load balance between these URLs giving priority to
   those with the lower latency. In case of failures the client will download files from the original URL.
@@ -166,11 +166,17 @@ Configuration options are:
   Here is an example of valid value for http(s) adapter::
 
     [storage]
-    storage.adapters = esrally.storage._http:HTTPAdapter
+    storage.adapters = esrally.storage.http:HTTPAdapter,esrally.storage.aws:S3Adapter
 
+  At this point in time ``esrally.storage.http:HTTPAdapter`` and ``esrally.storage.aws:S3Adapter`` are the only
+  known ``Adapter`` implementations intended for public use and they are both enabled by default. So it is required
+  to edit this option for special customizations (like for example remove one of them or adding a new
+  custom implementation for accessing a special infrastructure server).
 
-  At this point in time `esrally.storage._http:HTTPAdapter` is the only existing `Adapter` implementations intended
-  for public use and that is already the default one. So it is required to edit this option for special customizations.
+* ``storage.cache_ttl`` indicates the default time to live in seconds replies from head requests has to be considered
+  valid. In case this value is zero, the caching mechanism is disabled.
+
+* ``storage.chunk_size`` is used to specify the default size (in bytes) of data chunks to be downloaded using adapters.
 
 * ``storage.local_dir`` indicates the default directory where to store local files when no path has been specified.
 
@@ -208,17 +214,17 @@ Configuration options are:
   The mirroring of the files on mirrors servers has to be provided by the infrastructure. The esrally client will look
   for the files on the destination mirror endpoints URLs or use the original source endpoint URL in case the files
   are not mirrored or they have a different size from the source one. The client will prefer endpoints with the lower
-  latency fetching the head of the file.
+  latency measured by fetching the headers of the file.
 
-* ``storage.monitor_interval`` represents the time interval (in seconds) `TransferManager` should wait for consecutive
+* ``storage.monitor_interval`` represents the time interval (in seconds) ``TransferManager`` should wait for consecutive
   monitor operations (log transfer and connections statistics, adjust the maximum number of connections, etc.).
 
 * ``storage.multipart_size`` When the file size measured in bytes is greater than this value the file is split in chunk
-  of this size plus the last one ()that could be smaller). Each part will be downloaded separately and in parallel using
-  a dedicated connection by a worker thread and eventually from a different mirror server to load balance the network
-  traffic between multiple servers. If the resulting number of parts is greater than ``storage.max_workers`` and
-  ``storage.max_connections`` options, then the transfer of those parts exceeding these limits will be performed as
-  soon as a worker thread gets available or a HTTP connection get released by another thread.
+  of this size plus the last one. Each part will be downloaded separately and in parallel using a dedicated connection
+  by a worker thread and eventually from a different mirror server to load balance the network traffic between multiple
+  servers. If the resulting number of parts is greater than ``storage.max_workers`` and ``storage.max_connections``
+  options, then the transfer of those parts exceeding these limits will be performed as soon as a worker thread gets
+  available or a HTTP connection get released by another thread.
 
 * ``storage.random_seed`` a string used to initialize the client random number generator. This could be used to make
   problems easier to reproduce in continuous integration. In most of the cases it should be left empty.
@@ -229,10 +235,13 @@ HTTP Adapter
 
 This adapter can be used only to download files from public HTTP or HTTPS servers.
 
-* ``storage.http.chunk_size`` is used to specify the size of the buffer is being used for transferring chunk of files.
+It must be listed in the ``storage.adapters`` option::
+
+    [storage]
+    storage.adapters = esrally.storage.http:HTTPAdapter
 
 * ``storage.http.max_retries`` is used to configure the maximum number of retries for making HTTP adapter requests.
-  it accept a numeric value to simply specify total number of retries. Examples::
+  It accepts a numeric value to simply specify total number of retries. Examples::
 
     [storage]
     storage.http.max_retries = 3
@@ -256,6 +265,11 @@ It requires `Boto3 Client`_ to be installed and it accepts only URLs with the fo
 
   s3://<bucket-name>/[<object-key-prefix>]
 
+It must be listed in the ``storage.adapters`` option::
+
+    [storage]
+    storage.adapters = esrally.storage.aws:S3Adapter
+
 In the case the boto3 client is not installed, and S3 buckets are publicly readable without authentication, you can use
 the HTTP adapter instead, for example by using the following URL format::
 
@@ -294,6 +308,20 @@ Configuration options:
   .. _Boto3 Client: https://boto3.amazonaws.com/v1/documentation/api/latest/index.html
   .. _S3 Service Documentation: https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/DownloadDistS3AndCustomOrigins.html#concept_S3Origin
 
+
+track
+~~~~~
+
+This section specifies how tracks corpora files has to be fetched. Available options are:
+
+* ``track.downloader.multipart_enabled`` if ``true``, it will enable the use the new multipart ``esrally.storage`` package for
+  downloading corpora files. For more configuration options please have a look to the `storage`_ configuration section.
+
+.. warning::
+
+    Transfers manager implementation is experimental and under active development. Enable it only if you actually need it.
+
+
 tracks
 ~~~~~~
 

esrally/config.py

@@ -16,15 +16,21 @@
 # under the License.
 
 import configparser
+import contextvars
 import logging
 import os.path
 import shutil
+import typing
 from enum import Enum
 from string import Template
 
+from typing_extensions import Self
+
 from esrally import PROGRAM_NAME, exceptions, paths, types
 from esrally.utils import io
 
+LOG = logging.getLogger(__name__)
+
 
 class Scope(Enum):
     # Valid for all benchmarks, typically read from the configuration file
@@ -40,6 +46,7 @@ class Scope(Enum):
 
 
 class ConfigFile:
+
     def __init__(self, config_name=None, **kwargs):
         self.config_name = config_name
 
@@ -124,22 +131,66 @@ def auto_load_local_config(base_config, additional_sections=None, config_file_cl
     return cfg
 
 
-class Config:
-    EARLIEST_SUPPORTED_VERSION = 17
+CONFIG = contextvars.ContextVar[typing.Optional[types.Config]](f"{__name__}.config", default=None)
+
+
+def get_config() -> types.Config:
+    cfg = CONFIG.get()
+    if cfg is None:
+        raise exceptions.ConfigError("Config not initialized.")
+    return cfg
+
+
+def init_config(cfg: types.Config, *, force=False) -> types.Config:
+    if not force and CONFIG.get():
+        raise exceptions.ConfigError(f"Config already set: {cfg}")
+    cfg = Config.from_config(cfg)
+    CONFIG.set(cfg)
+    return cfg
 
-    CURRENT_CONFIG_VERSION = 17
 
+def clear_config() -> None:
+    CONFIG.set(None)
+
+
+class Config(types.Config):
     """
     Config is the main entry point to retrieve and set benchmark properties. It provides multiple scopes to allow overriding of values on
     different levels (e.g. a command line flag can override the same configuration property in the config file). These levels are
     transparently resolved when a property is retrieved and the value on the most specific level is returned.
     """
 
-    def __init__(self, config_name=None, config_file_class=ConfigFile, **kwargs):
+    EARLIEST_SUPPORTED_VERSION = 17
+
+    CURRENT_CONFIG_VERSION = 17
+
+    @classmethod
+    def from_config(cls, cfg: types.Config | None = None) -> Self:
+        if cfg is None:
+            cfg = get_config()
+        if isinstance(cfg, cls):
+            return cfg
+        if isinstance(cfg, types.Config):
+            return cls(opts_from=cfg)
+        raise TypeError(f"unexpected cfg: got type {type(cfg).__name__}, expected types.Config")
+
+    def __init__(self, config_name: str | None = None, config_file_class=ConfigFile, copy_from: types.Config | None = None, **kwargs):
         self.name = config_name
         self.config_file = config_file_class(config_name, **kwargs)
         self._opts = {}
-        self._clear_config()
+        if copy_from is not None:
+            self.update(copy_from)
+        self._override_config()
+
+    def update(self, cfg: types.Config):
+        if isinstance(cfg, Config):
+            self.name = cfg.name
+            self.config_file = cfg.config_file
+            self._opts.update(cfg._opts)  # pylint: disable=protected-access
+            return
+        for section in cfg.all_sections():
+            for name, value in cfg.all_opts(section).items():
+                self.add(Scope.application, section, name, value)
 
     def add(self, scope, section: types.Section, key: types.Key, value):
         """
@@ -185,7 +236,10 @@ def opts(self, section: types.Section, key: types.Key, default_value=None, manda
             else:
                 raise exceptions.ConfigError(f"No value for mandatory configuration: section='{section}', key='{key}'")
 
-    def all_opts(self, section: types.Section):
+    def all_sections(self) -> list[types.Section]:
+        return list(typing.get_args(types.Section))
+
+    def all_opts(self, section: types.Section) -> dict[str, typing.Any]:
         """
         Finds all options in a section and returns them in a dict.
 
@@ -233,21 +287,24 @@ def load_config(self, auto_upgrade=False):
     def _do_load_config(self):
         config = self.config_file.load()
         # It's possible that we just reload the configuration
-        self._clear_config()
+        self._opts = {}
+        self._override_config()
         self._fill_from_config_file(config)
 
-    def _clear_config(self):
+    def _override_config(self):
         # This map contains default options that we don't want to sprinkle all over the source code but we don't want users to change
         # them either
-        self._opts = {
-            (Scope.application, "source", "distribution.dir"): "distributions",
-            (Scope.application, "benchmarks", "track.repository.dir"): "tracks",
-            (Scope.application, "benchmarks", "track.default.repository"): "default",
-            (Scope.application, "provisioning", "node.name.prefix"): "rally-node",
-            (Scope.application, "provisioning", "node.http.port"): 39200,
-            (Scope.application, "mechanic", "team.repository.dir"): "teams",
-            (Scope.application, "mechanic", "team.default.repository"): "default",
-        }
+        self._opts.update(
+            {
+                (Scope.application, "source", "distribution.dir"): "distributions",
+                (Scope.application, "benchmarks", "track.repository.dir"): "tracks",
+                (Scope.application, "benchmarks", "track.default.repository"): "default",
+                (Scope.application, "provisioning", "node.name.prefix"): "rally-node",
+                (Scope.application, "provisioning", "node.http.port"): 39200,
+                (Scope.application, "mechanic", "team.repository.dir"): "teams",
+                (Scope.application, "mechanic", "team.default.repository"): "default",
+            }
+        )
 
     def _fill_from_config_file(self, config):
         for section in config.sections():
@@ -279,6 +336,15 @@ def _k(self, scope, section: types.Section, key: types.Key):
         else:
             return scope, section, key
 
+    def __eq__(self, other):
+        if not isinstance(other, Config):
+            return False
+        if other.name != self.name:
+            return False
+        if other._opts != self._opts:
+            return False
+        return True
+
 
 def migrate(config_file, current_version, target_version, out=print, i=input):
     logger = logging.getLogger(__name__)

esrally/storage/__init__.py

@@ -14,13 +14,12 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from esrally.storage._adapter import AdapterRegistry, Head
-from esrally.storage._client import Client
-from esrally.storage._executor import Executor, ThreadPoolExecutor
-from esrally.storage._http import HTTPAdapter
+from esrally.storage._adapter import Adapter, AdapterRegistry, Head
+from esrally.storage._config import StorageConfig
 from esrally.storage._manager import (
     TransferManager,
+    get_transfer_manager,
     init_transfer_manager,
-    quit_transfer_manager,
-    transfer_manager,
+    shutdown_transfer_manager,
 )
+from esrally.storage._transfer import Transfer