diff --git a/audbackend/core/backend/artifactory.py b/audbackend/core/backend/artifactory.py index 9076a4c0..07cf356e 100644 --- a/audbackend/core/backend/artifactory.py +++ b/audbackend/core/backend/artifactory.py @@ -1,9 +1,11 @@ from collections.abc import Iterator +import configparser import os import artifactory import dohq_artifactory import requests +from requests.adapters import HTTPAdapter import audeer @@ -70,6 +72,16 @@ def _download( class Artifactory(Base): r"""Backend for Artifactory. + Connection pool settings can be configured + via the config file (see :meth:`get_config`): + + * ``pool_connections``: number of connection pools to cache (default: 10) + * ``pool_maxsize``: max connections per pool (default: 10) + * ``max_retries``: max retries per connection (default: 0) + + For bulk operations downloading many files in parallel, + increase ``pool_maxsize`` to match the number of workers. + Args: host: host address repository: repository name @@ -167,6 +179,62 @@ def get_authentication(cls, host: str) -> tuple[str, str]: return username, api_key + @classmethod + def get_config(cls, host: str) -> dict: + """Configuration of Artifactory server. + + The default path of the config file + (:file:`~/.artifactory_python.cfg`) + can be overwritten with the environment variable + ``ARTIFACTORY_CONFIG_FILE``. + + If no config file can be found, + or no entry for the requested host, + an empty dictionary is returned. + + The config file + expects one section per host, + e.g. + + .. code-block:: ini + + [artifactory.example.com/artifactory] + pool_connections = 10 + pool_maxsize = 100 + max_retries = 3 + + Connection pool settings (for bulk operations with many files): + + * ``pool_connections``: number of connection pools to cache (default: 10) + * ``pool_maxsize``: maximum number of connections per pool. + Increase this for parallel downloads of many files (default: 10) + * ``max_retries``: maximum number of retries per connection (default: 0) + + Args: + host: hostname + + Returns: + config entries as dictionary + + """ + config_file = os.getenv( + "ARTIFACTORY_CONFIG_FILE", + artifactory.default_config_path, + ) + config_file = audeer.path(config_file) + + if os.path.exists(config_file): + config = configparser.ConfigParser(allow_no_value=True) + config.read(config_file) + try: + config = dict(config.items(host)) + except configparser.NoSectionError: + config = {} + else: + config = {} + + return config + def _checksum( self, path: str, @@ -327,6 +395,37 @@ def _open( r"""Open connection to backend.""" self._session = requests.Session() self._session.auth = self.authentication + + # Configure connection pooling for better performance + # with parallel downloads of many files. + # Settings can be tuned via backend config: + # - "pool_connections": number of pools to cache (default: 10) + # - "pool_maxsize": max connections per pool (default: 10) + # - "max_retries": max retries per connection (default: 0) + config = self.get_config(self.host) + pool_connections = utils.parse_config_int( + config.get("pool_connections", 10), + name="pool_connections", + default=10, + ) + pool_maxsize = utils.parse_config_int( + config.get("pool_maxsize", 10), + name="pool_maxsize", + default=10, + ) + max_retries = utils.parse_config_int( + config.get("max_retries", 0), + name="max_retries", + default=0, + ) + adapter = HTTPAdapter( + pool_connections=pool_connections, + pool_maxsize=pool_maxsize, + max_retries=max_retries, + ) + self._session.mount("https://", adapter) + self._session.mount("http://", adapter) + path = artifactory.ArtifactoryPath(self.host, session=self._session) self._repo = path.find_repository(self.repository) if self._repo is None: diff --git a/audbackend/core/backend/minio.py b/audbackend/core/backend/minio.py index adbdd7d4..6e632dfe 100644 --- a/audbackend/core/backend/minio.py +++ b/audbackend/core/backend/minio.py @@ -18,12 +18,18 @@ class Minio(Base): r"""Backend for MinIO. - HTTP timeouts can be configured via the config file - (see :meth:`get_config`): + HTTP timeouts and connection pool settings can be configured + via the config file (see :meth:`get_config`): * ``connect_timeout``: seconds for connection establishment (default: 10.0) * ``read_timeout``: seconds for read operations; ``None`` means no timeout (default: ``None``) + * ``num_pools``: number of connection pools to cache (default: 10) + * ``pool_maxsize``: max connections per pool (default: 10) + * ``pool_block``: block when pool is full (default: ``false``) + + For bulk operations downloading many files in parallel, + increase ``pool_maxsize`` to match the number of workers. Alternatively, provide a custom ``http_client`` object as ``kwargs`` @@ -84,12 +90,16 @@ def __init__( if secure is None: secure = config.get("secure", True) - # Configure HTTP client with timeouts to prevent hanging connections. + # Configure HTTP client with timeouts and connection pooling. # Users can override by passing their own http_client in kwargs. - # Timeouts can be tuned via backend config: + # Settings can be tuned via backend config: # - "connect_timeout": seconds for connection establishment (default: 10.0) # - "read_timeout": seconds for read operations; None means no timeout # (default: None) + # - "num_pools": number of connection pools to cache (default: 10) + # - "pool_maxsize": max connections per pool (default: 10) + # - "pool_block": block when pool is full instead of creating new + # connection (default: False) if "http_client" not in kwargs: connect_timeout = _parse_timeout( config.get("connect_timeout", 10.0), @@ -102,7 +112,27 @@ def __init__( default=None, ) timeout = urllib3.Timeout(connect=connect_timeout, read=read_timeout) - kwargs["http_client"] = urllib3.PoolManager(timeout=timeout) + num_pools = utils.parse_config_int( + config.get("num_pools", 10), + name="num_pools", + default=10, + ) + pool_maxsize = utils.parse_config_int( + config.get("pool_maxsize", 10), + name="pool_maxsize", + default=10, + ) + pool_block = _parse_bool( + config.get("pool_block", False), + name="pool_block", + default=False, + ) + kwargs["http_client"] = urllib3.PoolManager( + timeout=timeout, + num_pools=num_pools, + maxsize=pool_maxsize, + block=pool_block, + ) # Open MinIO client self._client = minio.Minio( @@ -170,7 +200,7 @@ def get_config(cls, host: str) -> dict: access_key = "Q3AM3UQ867SPQQA43P2F" secret_key = "zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG" - Optional timeout settings can also be configured: + Optional timeout and connection pool settings can also be configured: .. code-block:: ini @@ -179,12 +209,25 @@ def get_config(cls, host: str) -> dict: secret_key = "zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG" connect_timeout = 10.0 read_timeout = 60.0 + num_pools = 10 + pool_maxsize = 100 + pool_block = false + + Timeout settings: * ``connect_timeout``: seconds for connection establishment (default: 10.0) * ``read_timeout``: seconds for read operations; use ``None`` for no timeout (default: ``None``) + Connection pool settings (for bulk operations with many files): + + * ``num_pools``: number of connection pools to cache (default: 10) + * ``pool_maxsize``: maximum number of connections per pool. + Increase this for parallel downloads of many files (default: 10) + * ``pool_block``: if ``true``, block and wait when pool is exhausted + instead of creating a new connection outside the pool (default: ``false``) + Args: host: hostname @@ -568,6 +611,41 @@ def _metadata(checksum: str): } +def _parse_bool( + value: str | bool, + *, + name: str, + default: bool, +) -> bool: + """Parse a boolean value from config. + + Converts string values to bool, handling common boolean strings. + If parsing fails, logs a warning and returns the default value. + + Args: + value: boolean value (string from config or bool) + name: name of the setting (for warning messages) + default: default value to use if parsing fails + + Returns: + parsed boolean value or default + + """ + if isinstance(value, bool): + return value + if isinstance(value, str): + if value.lower() in ("true", "1", "yes", "on"): + return True + if value.lower() in ("false", "0", "no", "off"): + return False + warnings.warn( + f"Invalid {name} value '{value}' in config, using default: {default}", + UserWarning, + stacklevel=4, + ) + return default + + def _parse_timeout( value: str | float | None, *, diff --git a/audbackend/core/utils.py b/audbackend/core/utils.py index 1807d79c..13350f8a 100644 --- a/audbackend/core/utils.py +++ b/audbackend/core/utils.py @@ -4,6 +4,7 @@ import os import re import time +import warnings from audbackend.core.errors import BackendError @@ -148,3 +149,36 @@ def raise_is_a_directory(path: str): os.strerror(errno.EISDIR), path, ) + + +def parse_config_int( + value: str | int, + *, + name: str, + default: int, +) -> int: + """Parse an integer value from config. + + Converts string values to int. + If parsing fails, logs a warning and returns the default value. + + Args: + value: integer value (string from config or int) + name: name of the setting (for warning messages) + default: default value to use if parsing fails + + Returns: + parsed integer value or default + + """ + if isinstance(value, int): + return value + try: + return int(value) + except (ValueError, TypeError): + warnings.warn( + f"Invalid {name} value '{value}' in config, using default: {default}", + UserWarning, + stacklevel=4, + ) + return default diff --git a/tests/test_backend_artifactory.py b/tests/test_backend_artifactory.py index a1fec44a..c71d3c57 100644 --- a/tests/test_backend_artifactory.py +++ b/tests/test_backend_artifactory.py @@ -1,4 +1,6 @@ import os +from unittest import mock +import warnings import pytest @@ -325,3 +327,177 @@ def test_get_archive_streaming(tmpdir, interface): "/archive.zip", dst_root_validated, validate=True ) assert sorted(extracted_validated) == ["file1.txt", "file2.txt"] + + +def test_get_config(tmpdir, hosts, hide_credentials): + r"""Test parsing of connection pool configuration. + + The `get_config()` class method is responsible + for parsing an Artifactory backend config file. + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["artifactory"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["ARTIFACTORY_CONFIG_FILE"] = config_path + + # config file does not exist + config = audbackend.backend.Artifactory.get_config(host) + assert config == {} + + # config file is empty + audeer.touch(config_path) + config = audbackend.backend.Artifactory.get_config(host) + assert config == {} + + # config file has different host + with open(config_path, "w") as fp: + fp.write(f"[{host}.abc]\n") + config = audbackend.backend.Artifactory.get_config(host) + assert config == {} + + # config file entry without variables + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + config = audbackend.backend.Artifactory.get_config(host) + assert config == {} + + # config file entry with pool variables + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("pool_connections = 20\n") + fp.write("pool_maxsize = 100\n") + fp.write("max_retries = 3\n") + config = audbackend.backend.Artifactory.get_config(host) + assert config["pool_connections"] == "20" + assert config["pool_maxsize"] == "100" + assert config["max_retries"] == "3" + + +def test_default_pool_configuration(tmpdir, hosts, hide_credentials): + r"""Test that default connection pool configuration is applied. + + When no pool config is set, the backend should create a session + with default pool settings: + - pool_connections: 10 + - pool_maxsize: 10 + - max_retries: 0 + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["artifactory"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["ARTIFACTORY_CONFIG_FILE"] = config_path + + # Create empty config file + audeer.touch(config_path) + + backend = audbackend.backend.Artifactory(host, "repository") + + # Mock find_repository to avoid actual network call + with mock.patch("artifactory.ArtifactoryPath.find_repository") as mock_find: + mock_find.return_value = mock.MagicMock() + backend.open() + + # Check that HTTPAdapter was mounted with default settings + adapter = backend._session.get_adapter("https://") + # Check that HTTPAdapter was mounted with default settings + adapter = backend._session.get_adapter("https://") + assert adapter._pool_connections == 10 + assert adapter._pool_maxsize == 10 + # Default retry configuration: no automatic retries + assert adapter.max_retries.total == 0 + + backend.close() + + +def test_custom_pool_from_config(tmpdir, hosts, hide_credentials): + r"""Test that custom connection pool values from config are honored. + + When pool values are specified in the config file, + they should be used instead of the defaults. + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["artifactory"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["ARTIFACTORY_CONFIG_FILE"] = config_path + + # Create config file with custom pool settings + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("pool_connections = 20\n") + fp.write("pool_maxsize = 100\n") + fp.write("max_retries = 5\n") + + backend = audbackend.backend.Artifactory(host, "repository") + + # Mock find_repository to avoid actual network call + with mock.patch("artifactory.ArtifactoryPath.find_repository") as mock_find: + mock_find.return_value = mock.MagicMock() + backend.open() + + # Check that HTTPAdapter was mounted with custom settings + adapter = backend._session.get_adapter("https://") + assert adapter._pool_connections == 20 + assert adapter._pool_maxsize == 100 + + backend.close() + + +def test_invalid_pool_warning(tmpdir, hosts, hide_credentials): + r"""Test that invalid pool values emit a warning and use defaults. + + When a non-numeric string is provided as a pool value, + a warning should be emitted and the default value should be used. + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["artifactory"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["ARTIFACTORY_CONFIG_FILE"] = config_path + + # Create config file with invalid pool values + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("pool_connections = invalid\n") + fp.write("pool_maxsize = large\n") + fp.write("max_retries = many\n") + + backend = audbackend.backend.Artifactory(host, "repository") + + # Mock find_repository to avoid actual network call + with mock.patch("artifactory.ArtifactoryPath.find_repository") as mock_find: + mock_find.return_value = mock.MagicMock() + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + backend.open() + + # Verify warnings were emitted for invalid values + warning_messages = [str(warning.message) for warning in w] + assert any("Invalid pool_connections" in msg for msg in warning_messages) + assert any("Invalid pool_maxsize" in msg for msg in warning_messages) + assert any("Invalid max_retries" in msg for msg in warning_messages) + + # Verify default pool values were used + adapter = backend._session.get_adapter("https://") + assert adapter._pool_connections == 10 # default + assert adapter._pool_maxsize == 10 # default + + backend.close() diff --git a/tests/test_backend_minio.py b/tests/test_backend_minio.py index 6c134550..8ed7de25 100644 --- a/tests/test_backend_minio.py +++ b/tests/test_backend_minio.py @@ -769,6 +769,148 @@ def test_invalid_timeout_warning(tmpdir, hosts, hide_credentials): assert timeout.read_timeout is None # default +def test_default_pool_configuration(tmpdir, hosts, hide_credentials): + r"""Test that default connection pool configuration is applied. + + When no pool config is set, the backend should create an http_client + with default pool settings: + - num_pools: 10 + - pool_maxsize: 10 + - pool_block: False + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["minio"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["MINIO_CONFIG_FILE"] = config_path + + # Create minimal config file without pool settings + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("access_key = test\n") + fp.write("secret_key = test\n") + + with capture_minio_kwargs() as captured: + audbackend.backend.Minio(host, "repository") + + # Verify default pool values + http_client = captured["http_client"] + assert isinstance(http_client, urllib3.PoolManager) + assert http_client.connection_pool_kw.get("maxsize") == 10 + assert http_client.connection_pool_kw.get("block") is False + + +def test_custom_pool_from_config(tmpdir, hosts, hide_credentials): + r"""Test that custom connection pool values from config are honored. + + When pool values are specified in the config file, + they should be used instead of the defaults. + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["minio"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["MINIO_CONFIG_FILE"] = config_path + + # Create config file with custom pool settings + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("access_key = test\n") + fp.write("secret_key = test\n") + fp.write("num_pools = 20\n") + fp.write("pool_maxsize = 100\n") + fp.write("pool_block = true\n") + + with capture_minio_kwargs() as captured: + audbackend.backend.Minio(host, "repository") + + # Verify the custom pool values from config + http_client = captured["http_client"] + assert http_client.connection_pool_kw.get("maxsize") == 100 + assert http_client.connection_pool_kw.get("block") is True + + +def test_pool_block_false_from_config(tmpdir, hosts, hide_credentials): + r"""Test that pool_block = false is parsed correctly. + + When pool_block is set to 'false' (or '0', 'no', 'off') in the config file, + it should be parsed as Python False. + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["minio"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["MINIO_CONFIG_FILE"] = config_path + + # Create config file with pool_block = false + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("access_key = test\n") + fp.write("secret_key = test\n") + fp.write("pool_block = false\n") + + with capture_minio_kwargs() as captured: + audbackend.backend.Minio(host, "repository") + + # Verify pool_block is False + http_client = captured["http_client"] + assert http_client.connection_pool_kw.get("block") is False + + +def test_invalid_pool_warning(tmpdir, hosts, hide_credentials): + r"""Test that invalid pool values emit a warning and use defaults. + + When a non-numeric string is provided as a pool value, + a warning should be emitted and the default value should be used. + + Args: + tmpdir: tmpdir fixture + hosts: hosts fixture + hide_credentials: hide_credentials fixture + + """ + host = hosts["minio"] + config_path = audeer.path(tmpdir, "config.cfg") + os.environ["MINIO_CONFIG_FILE"] = config_path + + # Create config file with invalid pool values + with open(config_path, "w") as fp: + fp.write(f"[{host}]\n") + fp.write("access_key = test\n") + fp.write("secret_key = test\n") + fp.write("num_pools = invalid\n") + fp.write("pool_maxsize = large\n") + fp.write("pool_block = maybe\n") + + with capture_minio_kwargs() as captured: + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + audbackend.backend.Minio(host, "repository") + + # Verify warnings were emitted for invalid values + warning_messages = [str(warning.message) for warning in w] + assert any("Invalid num_pools" in msg for msg in warning_messages) + assert any("Invalid pool_maxsize" in msg for msg in warning_messages) + assert any("Invalid pool_block" in msg for msg in warning_messages) + + # Verify default pool values were used + http_client = captured["http_client"] + assert http_client.connection_pool_kw.get("maxsize") == 10 # default + assert http_client.connection_pool_kw.get("block") is False # default + + @pytest.mark.parametrize( "interface", [(audbackend.backend.Minio, audbackend.interface.Versioned)],