Introduce LOGGING_OVERRIDE config var

localstack/config.py

@@ -418,6 +418,10 @@ def in_docker():
 LS_LOG = eval_log_type("LS_LOG")
 DEBUG = is_env_true("DEBUG") or LS_LOG in TRACE_LOG_LEVELS
 
+# EXPERIMENTAL
+# allow setting custom log levels for individual loggers
+LOGGING_OVERRIDE = os.environ.get("LOGGING_OVERRIDE", "")
+
 # whether to enable debugpy
 DEVELOP = is_env_true("DEVELOP")
 

localstack/logging/setup.py

@@ -4,6 +4,7 @@
 
 from localstack import config, constants
 
+from ..utils.collections import parse_key_value_pairs
 from .format import AddFormattedAttributes, DefaultFormatter
 
 # The log levels for modules are evaluated incrementally for logging granularity,
@@ -78,6 +79,24 @@ def setup_logging_from_config():
         for name, level in trace_internal_log_levels.items():
             logging.getLogger(name).setLevel(level)
 
+    raw_logging_override = config.LOGGING_OVERRIDE
+    if raw_logging_override:
+        try:
+            logging_overrides = parse_key_value_pairs(raw_logging_override)
+            for logger, level_name in logging_overrides.items():
+                level = getattr(logging, level_name, None)
+                if not level:
+                    raise RuntimeError(
+                        f"Failed to configure logging overrides ({raw_logging_override}): '{level_name}' is not a valid log level"
+                    )
+                logging.getLogger(logger).setLevel(level)
+        except RuntimeError:
+            raise
+        except Exception as e:
+            raise RuntimeError(
+                f"Failed to configure logging overrides ({raw_logging_override})"
+            ) from e
+
 
 def create_default_handler(log_level: int):
     log_handler = logging.StreamHandler(stream=sys.stderr)

localstack/utils/collections.py

@@ -533,3 +533,24 @@ def is_comma_delimited_list(string: str, item_regex: Optional[str] = None) -> bo
     if pattern.match(string) is None:
         return False
     return True
+
+
+def parse_key_value_pairs(raw_text: str) -> dict:
+    """
+    Parse a series of key-value pairs, in an environment variable format into a dictionary
+
+    >>> input = "a=b,c=d"
+    >>> assert parse_key_value_pairs(input) == {"a": "b", "c": "d"}
+    """
+    result = {}
+    for pair in raw_text.split(","):
+        items = pair.split("=")
+        if len(items) != 2:
+            raise ValueError(f"invalid key/value pair: '{pair}'")
+        raw_key, raw_value = items[0].strip(), items[1].strip()
+        if not raw_key:
+            raise ValueError(f"missing key: '{pair}'")
+        if not raw_value:
+            raise ValueError(f"missing value: '{pair}'")
+        result[raw_key] = raw_value
+    return result

tests/unit/utils/test_collections.py

@@ -9,6 +9,7 @@
     ImmutableList,
     convert_to_typed_dict,
     is_comma_delimited_list,
+    parse_key_value_pairs,
     select_from_typed_dict,
 )
 
@@ -193,3 +194,28 @@ def test_is_comma_limited_list():
     assert not is_comma_delimited_list("foo, bar baz")
     assert not is_comma_delimited_list("foo,")
     assert not is_comma_delimited_list("")
+
+
+@pytest.mark.parametrize(
+    "input_text,expected",
+    [
+        ("a=b", {"a": "b"}),
+        ("a=b,c=d", {"a": "b", "c": "d"}),
+    ],
+)
+def test_parse_key_value_pairs(input_text, expected):
+    assert parse_key_value_pairs(input_text) == expected
+
+
+@pytest.mark.parametrize(
+    "input_text,message",
+    [
+        ("a=b,", "invalid key/value pair: ''"),
+        ("a=b,c=", "missing value: 'c='"),
+    ],
+)
+def test_parse_key_value_pairs_error_messages(input_text, message):
+    with pytest.raises(ValueError) as exc_info:
+        parse_key_value_pairs(input_text)
+
+    assert str(exc_info.value) == message