diff --git a/docs/how-to-guides/online-server-performance-tuning.md b/docs/how-to-guides/online-server-performance-tuning.md index f10d7bff8c8..50e746a6f03 100644 --- a/docs/how-to-guides/online-server-performance-tuning.md +++ b/docs/how-to-guides/online-server-performance-tuning.md @@ -324,6 +324,7 @@ online_store: batch_size: 100 max_read_workers: 10 consistent_reads: false + warmup_connections: true max_pool_connections: 100 keepalive_timeout: 30.0 connect_timeout: 3 @@ -337,6 +338,7 @@ Key knobs: - **`batch_size`**: DynamoDB's `BatchGetItem` accepts up to 100 items per request. For 500 entities, this means 5 batches. Keep at 100 unless hitting the 16 MB response limit. - **`max_read_workers`**: Controls parallelism for batch reads. With 10 workers, those 5 batches run concurrently (~10 ms) instead of sequentially (~50 ms). - **`consistent_reads: false`**: Eventually consistent reads are faster and cheaper. Use `true` only if you need read-after-write consistency. +- **`warmup_connections: true`**: Pre-warms the DynamoDB connection pool on server startup by making a lightweight call (`describe_limits`). This avoids a cold-start latency penalty (~20ms) on the very first feature request. - **`max_pool_connections`**: Increase for high-throughput deployments to improve HTTP connection reuse to the DynamoDB endpoint. - **`keepalive_timeout`**: Longer keep-alive reduces TLS handshake overhead on reused connections. - **`connect_timeout` / `read_timeout`**: Lower values fail fast, improving p99. Set aggressively if your retry strategy covers transient failures. diff --git a/docs/reference/online-stores/dynamodb.md b/docs/reference/online-stores/dynamodb.md index a7f6b7392c9..faf2f9a15b8 100644 --- a/docs/reference/online-stores/dynamodb.md +++ b/docs/reference/online-stores/dynamodb.md @@ -37,6 +37,7 @@ online_store: batch_size: 100 max_read_workers: 10 consistent_reads: false + warmup_connections: true ``` {% endcode %} @@ -49,6 +50,7 @@ online_store: | `batch_size` | int | `100` | Number of items per BatchGetItem/BatchWriteItem request (max 100) | | `max_read_workers` | int | `10` | Maximum parallel threads for batch read operations. Higher values improve throughput for large batch reads but increase resource usage | | `consistent_reads` | bool | `false` | Whether to use strongly consistent reads (higher latency, guaranteed latest data) | +| `warmup_connections` | bool | `false` | Whether to pre-warm the async connection pool on startup with a lightweight call (`describe_limits`) | | `tags` | dict | `null` | AWS resource tags added to each table | | `session_based_auth` | bool | `false` | Use AWS session-based client authentication | @@ -63,6 +65,8 @@ For high-throughput workloads with large entity counts, increase `max_read_worke **Batch Size**: Increase `batch_size` up to 100 to reduce the number of API calls. However, larger batches may hit DynamoDB's 16MB response limit for tables with large feature values. +**Connection Warmup**: The DynamoDB async client does not establish actual TCP/TLS connections to the AWS endpoint on initialization. The very first feature retrieval request is penalized with a cold-start overhead (~20ms). Setting `warmup_connections: true` establishes the TCP connection pool during server startup. + ## Permissions Feast requires the following permissions in order to execute commands for DynamoDB online store: diff --git a/sdk/python/feast/infra/online_stores/dynamodb.py b/sdk/python/feast/infra/online_stores/dynamodb.py index b67c8a81f6b..905da844261 100644 --- a/sdk/python/feast/infra/online_stores/dynamodb.py +++ b/sdk/python/feast/infra/online_stores/dynamodb.py @@ -76,6 +76,9 @@ class DynamoDBOnlineStoreConfig(FeastConfigBaseModel): consistent_reads: StrictBool = False """Whether to read from Dynamodb by forcing consistent reads""" + warmup_connections: StrictBool = False + """Whether to warm up the connection pool with a lightweight call on initialization""" + tags: Union[Dict[str, str], None] = None """AWS resource tags added to each table""" @@ -146,7 +149,7 @@ def __init__(self): async def initialize(self, config: RepoConfig): online_config = config.online_store - await self._get_aiodynamodb_client( + client = await self._get_aiodynamodb_client( online_config.region, online_config.max_pool_connections, online_config.keepalive_timeout, @@ -157,6 +160,14 @@ async def initialize(self, config: RepoConfig): online_config.endpoint_url, ) + if online_config.warmup_connections: + try: + await client.describe_limits() + except Exception: + logger.warning( + "Failed to warmup DynamoDB connection pool", exc_info=True + ) + async def close(self): await self._aiodynamodb_close() diff --git a/sdk/python/tests/unit/infra/online_store/test_dynamodb_online_store.py b/sdk/python/tests/unit/infra/online_store/test_dynamodb_online_store.py index 3b84b1a3547..da47c34a821 100644 --- a/sdk/python/tests/unit/infra/online_store/test_dynamodb_online_store.py +++ b/sdk/python/tests/unit/infra/online_store/test_dynamodb_online_store.py @@ -75,6 +75,7 @@ def test_dynamodb_online_store_config_default(): assert dynamodb_store_config.read_timeout == 10 assert dynamodb_store_config.total_max_retry_attempts == 3 assert dynamodb_store_config.retry_mode == "adaptive" + assert dynamodb_store_config.warmup_connections is False def test_dynamodb_online_store_config_custom_params(): @@ -88,12 +89,58 @@ def test_dynamodb_online_store_config_custom_params(): batch_size=batch_size, endpoint_url=endpoint_url, table_name_template=table_name_template, + warmup_connections=True, ) assert dynamodb_store_config.type == "dynamodb" assert dynamodb_store_config.batch_size == batch_size assert dynamodb_store_config.endpoint_url == endpoint_url assert dynamodb_store_config.region == aws_region assert dynamodb_store_config.table_name_template == table_name_template + assert dynamodb_store_config.warmup_connections is True + + +@pytest.mark.asyncio +async def test_dynamodb_online_store_warmup_connections(): + """Test DynamoDBOnlineStore warmup connections in initialize method.""" + from unittest.mock import AsyncMock + + online_store = DynamoDBOnlineStore() + + # Mock _get_aiodynamodb_client to return a mock client + mock_client = AsyncMock() + mock_client.describe_limits = AsyncMock() + online_store._get_aiodynamodb_client = AsyncMock(return_value=mock_client) + + # Test case 1: warmup_connections=True + config_warmup = RepoConfig( + registry=REGISTRY, + project=PROJECT, + provider=PROVIDER, + online_store=DynamoDBOnlineStoreConfig(region=REGION, warmup_connections=True), + offline_store=DaskOfflineStoreConfig(), + entity_key_serialization_version=3, + ) + await online_store.initialize(config_warmup) + mock_client.describe_limits.assert_called_once() + + # Test case 2: warmup_connections=False + mock_client.describe_limits.reset_mock() + config_no_warmup = RepoConfig( + registry=REGISTRY, + project=PROJECT, + provider=PROVIDER, + online_store=DynamoDBOnlineStoreConfig(region=REGION, warmup_connections=False), + offline_store=DaskOfflineStoreConfig(), + entity_key_serialization_version=3, + ) + await online_store.initialize(config_no_warmup) + mock_client.describe_limits.assert_not_called() + + # Test case 3: warmup_connections=True and describe_limits raises an exception (should catch and log warning) + mock_client.describe_limits.reset_mock() + mock_client.describe_limits.side_effect = Exception("Connection failed") + await online_store.initialize(config_warmup) + mock_client.describe_limits.assert_called_once() def test_dynamodb_online_store_config_dynamodb_client(dynamodb_online_store):