Merge pull request #656 from aiven/jjaakola-aiven-fix-producer-max-co…

…nfig-on-aiohttp-application

fix: set aiohttp Application client max size according with producer config

README.rst

@@ -308,6 +308,9 @@ Keys to take special care are the ones needed to configure Kafka and advertised_
    * - Parameter
      - Default Value
      - Description
+   * - ``http_request_max_size``
+     - ``1048576``
+     - The maximum client HTTP request size. This value controls how large (POST) payloads are allowed. When configuration of ``karapace_rest`` is set to `true` and ``http_request_max_size`` is not set, Karapace configuration adapts the allowed client max size from the ``producer_max_request_size``. In cases where automatically selected size is not enough the configuration can be overridden by setting a value in configuration. For schema registry operation set the client max size according to expected size of schema payloads if default size is not enough.
    * - ``advertised_protocol``
      - ``http``
      - The protocol being advertised to other instances of Karapace that are attached to the same Kafka group.

karapace/config.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from enum import Enum, unique
-from karapace.constants import DEFAULT_SCHEMA_TOPIC
+from karapace.constants import DEFAULT_AIOHTTP_CLIENT_MAX_SIZE, DEFAULT_PRODUCER_MAX_REQUEST, DEFAULT_SCHEMA_TOPIC
 from karapace.utils import json_decode, json_encode, JSONDecodeError
 from pathlib import Path
 from typing import IO, Mapping
@@ -103,6 +103,7 @@ class ConfigDefaults(Config, total=False):
     "consumer_idle_disconnect_timeout": 0,
     "fetch_min_bytes": 1,
     "group_id": "schema-registry",
+    "http_request_max_size": None,
     "host": "127.0.0.1",
     "port": 8081,
     "server_tls_certfile": None,
@@ -136,7 +137,7 @@ class ConfigDefaults(Config, total=False):
     "producer_compression_type": None,
     "producer_count": 5,
     "producer_linger_ms": 100,
-    "producer_max_request_size": 1048576,
+    "producer_max_request_size": DEFAULT_PRODUCER_MAX_REQUEST,
     "session_timeout_ms": 10000,
     "karapace_rest": False,
     "karapace_registry": False,
@@ -186,6 +187,24 @@ def set_config_defaults(config: ConfigDefaults) -> Config:
     new_config.setdefault("tags", {})
     new_config["tags"]["app"] = "Karapace"
 
+    # Set the aiohttp client max size if REST Proxy is enabled and producer max request configuration is altered from default
+    # and aiohttp client max size is not set
+    # Use the http request max size from the configuration without altering if set.
+    if (
+        new_config["karapace_rest"]
+        and new_config["producer_max_request_size"] > DEFAULT_PRODUCER_MAX_REQUEST
+        and new_config["http_request_max_size"] is None
+    ):
+        # REST Proxy API configuration for producer max request size must be taken into account
+        # also for the aiohttp.web.Application client max size.
+        # Always add the aiohttp default client max size as the headroom above the producer max request size.
+        # The input JSON size for REST Proxy is not easy to estimate, lot of small records in single request has
+        # a lot of overhead due to JSON structure.
+        new_config["http_request_max_size"] = new_config["producer_max_request_size"] + DEFAULT_AIOHTTP_CLIENT_MAX_SIZE
+    elif new_config["http_request_max_size"] is None:
+        # Set the default aiohttp client max size
+        new_config["http_request_max_size"] = DEFAULT_AIOHTTP_CLIENT_MAX_SIZE
+
     set_settings_from_environment(new_config)
     set_sentry_dsn_from_environment(new_config)
     validate_config(new_config)

karapace/constants.py

@@ -10,6 +10,8 @@
 API_VERSION_AUTO_TIMEOUT_MS: Final = 30000
 TOPIC_CREATION_TIMEOUT_MS: Final = 20000
 DEFAULT_SCHEMA_TOPIC: Final = "_schemas"
+DEFAULT_PRODUCER_MAX_REQUEST: Final = 1048576
+DEFAULT_AIOHTTP_CLIENT_MAX_SIZE: Final = 1048576
 
 SECOND: Final = 1000
 MINUTE: Final = 60 * SECOND

karapace/rapu.py

@@ -164,12 +164,15 @@ def __init__(
         self.app_name = app_name
         self.config = config
         self.app_request_metric = f"{app_name}_request"
-        self.app = aiohttp.web.Application()
+        self.app = self._create_aiohttp_application(config=config)
         self.log = logging.getLogger(self.app_name)
         self.stats = StatsClient(config=config)
         self.app.on_cleanup.append(self.close_by_app)
         self.not_ready_handler = not_ready_handler
 
+    def _create_aiohttp_application(self, *, config: Config) -> aiohttp.web.Application:
+        return aiohttp.web.Application(client_max_size=config["http_request_max_size"])
+
     async def close_by_app(self, app: aiohttp.web.Application) -> None:  # pylint: disable=unused-argument
         await self.close()
 

tests/unit/test_config.py

@@ -0,0 +1,58 @@
+"""
+Test config
+
+Copyright (c) 2023 Aiven Ltd
+See LICENSE for details
+"""
+from karapace.config import set_config_defaults
+from karapace.constants import DEFAULT_AIOHTTP_CLIENT_MAX_SIZE, DEFAULT_PRODUCER_MAX_REQUEST
+
+
+def test_http_request_max_size() -> None:
+    config = set_config_defaults(
+        {
+            "karapace_rest": False,
+            "producer_max_request_size": DEFAULT_PRODUCER_MAX_REQUEST + 1024,
+        }
+    )
+    assert config["http_request_max_size"] == DEFAULT_AIOHTTP_CLIENT_MAX_SIZE
+
+    config = set_config_defaults(
+        {
+            "karapace_rest": False,
+            "http_request_max_size": 1024,
+        }
+    )
+    assert config["http_request_max_size"] == 1024
+
+    config = set_config_defaults(
+        {
+            "karapace_rest": True,
+        }
+    )
+    assert config["http_request_max_size"] == DEFAULT_AIOHTTP_CLIENT_MAX_SIZE
+
+    config = set_config_defaults(
+        {
+            "karapace_rest": True,
+            "producer_max_request_size": 1024,
+        }
+    )
+    assert config["http_request_max_size"] == DEFAULT_AIOHTTP_CLIENT_MAX_SIZE
+
+    config = set_config_defaults(
+        {
+            "karapace_rest": True,
+            "producer_max_request_size": DEFAULT_PRODUCER_MAX_REQUEST + 1024,
+        }
+    )
+    assert config["http_request_max_size"] == DEFAULT_PRODUCER_MAX_REQUEST + 1024 + DEFAULT_AIOHTTP_CLIENT_MAX_SIZE
+
+    config = set_config_defaults(
+        {
+            "karapace_rest": True,
+            "producer_max_request_size": DEFAULT_PRODUCER_MAX_REQUEST + 1024,
+            "http_request_max_size": 1024,
+        }
+    )
+    assert config["http_request_max_size"] == 1024