feat: Emit worker pool size as a DEBUG log event during init on Lambda Managed Instances (#219)

* feat: Emit worker pool size as a DEBUG log event during init on Lambda Managed Instances

* fix: Stringify non-serializable JSON log values and close init log sink

Address PR review feedback:
- Add default=str to the JSON log encoder so dict messages (and extra
  attributes) containing non-serializable values are stringified instead
  of raising and dropping the log record.
- Close the log sink opened by init_logging deterministically after the
  parent's handler is removed, instead of relying on GC.

* fix: Remove redundant output redirect from parent process

RAPID wires the runtime main process's stdout/stderr to the log egress
at spawn, so the parent does not need to dial the telemetry FD provider
socket — that socket exists for the forked workers, which continue to
redirect in run_single. Addresses PR review feedback.

Author: vip-amzn

RELEASE.CHANGELOG.md

@@ -1,3 +1,7 @@
+### September 2, 2026
+`4.0.3`
+- Emit a structured `runtime_worker_pool_initializing` DEBUG log event once per execution environment during INIT in multi-concurrent (Lambda Managed Instances) mode, reporting `workerCount` and `executionEnvironmentMaxConcurrency` for worker pool observability. Only visible when the function log level is DEBUG or lower; no impact on the standard on-demand path.
+
 ### July 15, 2026
 `4.0.2`
 - Add `Lambda-Runtime-Invocation-Id` header support for cross-wiring protection. The RIC now echoes the invocation ID received from RAPID on `/next` back on `/response` and `/error`, enabling RAPID to detect and reject stale responses from timed-out invocations.

awslambdaric/__init__.py

@@ -2,4 +2,4 @@
 Copyright 2021 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 """
 
-__version__ = "4.0.2"
+__version__ = "4.0.3"

awslambdaric/bootstrap.py

@@ -493,6 +493,16 @@ def _log_preview_runtime_warning():
         logging.warning(get_lambda_preview_runtime_warning_message())
 
 
+def init_logging():
+    """Setup logging for the parent process before forking (LMI only)."""
+    sys.stdout = Unbuffered(sys.stdout)
+    sys.stderr = Unbuffered(sys.stderr)
+    log_sink = create_log_sink()
+    log_sink.__enter__()
+    _setup_logging(_AWS_LAMBDA_LOG_FORMAT, _AWS_LAMBDA_LOG_LEVEL, log_sink)
+    return log_sink
+
+
 def run(handler, lambda_runtime_client):
     sys.stdout = Unbuffered(sys.stdout)
     sys.stderr = Unbuffered(sys.stderr)

awslambdaric/lambda_multi_concurrent_utils.py

@@ -2,6 +2,7 @@
 Copyright 2025 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 """
 
+import logging
 import os
 import sys
 import socket
@@ -10,6 +11,8 @@
 from . import bootstrap
 from .lambda_runtime_client import LambdaMultiConcurrentRuntimeClient
 
+WORKER_POOL_INITIALIZING_EVENT = "runtime_worker_pool_initializing"
+
 
 class MultiConcurrentRunner:
     @staticmethod
@@ -32,6 +35,27 @@ def run_single(
         client = LambdaMultiConcurrentRuntimeClient(api_addr, use_thread)
         bootstrap.run(handler, client)
 
+    @classmethod
+    def _emit_worker_pool_event(cls, max_concurrency: int):
+        """Emit worker pool DEBUG event once from the parent before forking.
+
+        No output redirection here: RAPID wires the runtime main process's
+        stdout/stderr to the log egress at spawn. The FD provider socket is
+        only for the forked workers, which redirect in run_single.
+        """
+        log_sink = bootstrap.init_logging()
+        logging.getLogger().debug(
+            {
+                "event": WORKER_POOL_INITIALIZING_EVENT,
+                "workerCount": max_concurrency,
+                "executionEnvironmentMaxConcurrency": max_concurrency,
+            }
+        )
+        logging.getLogger().handlers.clear()
+        # Close the sink deterministically now that its handler is gone
+        # (no-op for StandardLogSink; releases the fd for framed sinks).
+        log_sink.__exit__(None, None, None)
+
     @classmethod
     def run_concurrent(
         cls,
@@ -41,6 +65,8 @@ def run_concurrent(
         socket_path: str,
         max_concurrency: int,
     ):
+        cls._emit_worker_pool_event(max_concurrency)
+
         processes = []
         for _ in range(max_concurrency):
             p = multiprocessing.Process(

awslambdaric/lambda_runtime_log_utils.py

@@ -68,7 +68,10 @@ def _get_log_level_from_env_var(log_level):
 }
 _DEFAULT_FRAME_TYPE = _TEXT_FRAME_TYPES[logging.NOTSET]
 
-_json_encoder = json.JSONEncoder(ensure_ascii=False)
+# default=str keeps formatting resilient: non-JSON-serializable values in
+# dict messages or `extra` attributes are stringified instead of raising and
+# dropping the whole log record.
+_json_encoder = json.JSONEncoder(ensure_ascii=False, default=str)
 _encode_json = _json_encoder.encode
 
 
@@ -117,7 +120,11 @@ def format(self, record: logging.LogRecord) -> str:
         result = {
             "timestamp": self.formatTime(record, self.datefmt),
             "level": record.levelname,
-            "message": record.getMessage(),
+            "message": (
+                record.msg
+                if isinstance(record.msg, dict) and not record.args
+                else record.getMessage()
+            ),
             "logger": record.name,
             "stackTrace": self.__format_stacktrace(record.exc_info),
             "errorType": self.__format_exception_name(record.exc_info),

tests/test_bootstrap.py

@@ -1540,6 +1540,106 @@ def test_set_log_level_with_dictConfig(self, mock_stderr, mock_stdout):
         self.assertEqual(mock_stdout.getvalue(), "")
 
 
+class TestWorkerPoolInitializedLog(unittest.TestCase):
+    def setUp(self):
+        logging.getLogger().handlers.clear()
+        logging.getLogger().level = logging.NOTSET
+
+    def tearDown(self):
+        logging.getLogger().handlers.clear()
+        logging.getLogger().level = logging.NOTSET
+
+    def _setup_json_logging(self, log_level):
+        bootstrap._setup_logging(
+            LogFormat.from_str("JSON"), log_level, bootstrap.StandardLogSink()
+        )
+
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_dict_message_serialized_as_nested_json_at_debug(self, mock_stdout):
+        self._setup_json_logging("DEBUG")
+
+        logging.getLogger().debug(
+            {
+                "event": "runtime_worker_pool_initializing",
+                "workerCount": 17,
+                "executionEnvironmentMaxConcurrency": 34,
+            }
+        )
+
+        data = json.loads(mock_stdout.getvalue().strip())
+        self.assertEqual(data["level"], "DEBUG")
+        self.assertEqual(
+            data["message"],
+            {
+                "event": "runtime_worker_pool_initializing",
+                "workerCount": 17,
+                "executionEnvironmentMaxConcurrency": 34,
+            },
+        )
+
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_not_emitted_at_higher_log_levels(self, mock_stdout):
+        for log_level in ("INFO", "WARN", "ERROR", "FATAL"):
+            with self.subTest(log_level):
+                logging.getLogger().handlers.clear()
+                logging.getLogger().level = logging.NOTSET
+                self._setup_json_logging(_get_log_level_from_env_var(log_level))
+
+                logging.getLogger().debug({"event": "test"})
+
+                self.assertEqual(mock_stdout.getvalue(), "")
+
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_init_logging_enables_parent_emission(self, mock_stdout):
+        with patch.dict(
+            os.environ,
+            {"AWS_LAMBDA_LOG_FORMAT": "JSON", "AWS_LAMBDA_LOG_LEVEL": "DEBUG"},
+            clear=True,
+        ):
+            importlib.reload(bootstrap)
+            bootstrap.init_logging()
+
+            logging.getLogger().debug(
+                {
+                    "event": "runtime_worker_pool_initializing",
+                    "workerCount": 4,
+                    "executionEnvironmentMaxConcurrency": 4,
+                }
+            )
+
+        importlib.reload(bootstrap)
+
+        data = json.loads(mock_stdout.getvalue())
+        self.assertEqual(data["level"], "DEBUG")
+        self.assertEqual(data["message"]["event"], "runtime_worker_pool_initializing")
+
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_dict_message_with_non_serializable_values_is_not_dropped(
+        self, mock_stdout
+    ):
+        import datetime
+        import decimal
+
+        self._setup_json_logging("DEBUG")
+
+        logging.getLogger().debug(
+            {
+                "event": "custom_event",
+                "when": datetime.datetime(2026, 9, 3, 12, 0, 0),
+                "amount": decimal.Decimal("1.5"),
+                "blob": b"bytes",
+            }
+        )
+
+        # The record must not be dropped: it serializes with values
+        # stringified via the encoder's default=str fallback.
+        data = json.loads(mock_stdout.getvalue())
+        self.assertEqual(data["message"]["event"], "custom_event")
+        self.assertEqual(data["message"]["when"], "2026-09-03 12:00:00")
+        self.assertEqual(data["message"]["amount"], "1.5")
+        self.assertEqual(data["message"]["blob"], "b'bytes'")
+
+
 class TestBootstrapModule(unittest.TestCase):
     def test_run(self):
         expected_handler = "app.my_test_handler"

tests/test_concurrency.py

@@ -38,6 +38,8 @@ def fake_bootstrap_run(handler, lambda_runtime_client):
 
         with patch(
             "awslambdaric.lambda_multi_concurrent_utils.MultiConcurrentRunner._redirect_output"
+        ), patch(
+            "awslambdaric.lambda_multi_concurrent_utils.MultiConcurrentRunner._emit_worker_pool_event"
         ), patch(
             "awslambdaric.lambda_multi_concurrent_utils.bootstrap.run",
             side_effect=fake_bootstrap_run,

tests/test_multi_concurrent_runner.py

@@ -4,7 +4,7 @@
 
 import sys
 import unittest
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch, MagicMock, call
 
 from awslambdaric.lambda_multi_concurrent_utils import MultiConcurrentRunner
 
@@ -58,8 +58,9 @@ def test_run_single_creates_client_and_calls_bootstrap(
         mock_client_cls.assert_called_once_with("addr", True)
         mock_bootstrap.run.assert_called_once_with("h.fn", mock_client)
 
+    @patch.object(MultiConcurrentRunner, "_emit_worker_pool_event")
     @patch("multiprocessing.Process")
-    def test_run_concurrent_spawns_and_joins(self, mock_process):
+    def test_run_concurrent_spawns_and_joins(self, mock_process, mock_emit):
         fake_proc = MagicMock()
         mock_process.return_value = fake_proc
 
@@ -77,6 +78,46 @@ def test_run_concurrent_spawns_and_joins(self, mock_process):
             self.assertEqual(target, MultiConcurrentRunner.run_single)
             self.assertEqual(args, ("h", "a", False, "/sock"))
 
+    @patch("multiprocessing.Process")
+    def test_run_concurrent_emits_worker_pool_event_once_before_spawning(
+        self, mock_process
+    ):
+        mock_process.return_value = MagicMock()
+        order_tracker = MagicMock()
+        order_tracker.attach_mock(mock_process, "process")
+
+        with patch.object(
+            MultiConcurrentRunner, "_emit_worker_pool_event"
+        ) as mock_emit:
+            order_tracker.attach_mock(mock_emit, "emit")
+            MultiConcurrentRunner.run_concurrent(
+                "h", "a", False, "/sock", max_concurrency=3
+            )
+
+        mock_emit.assert_called_once_with(3)
+        self.assertEqual(order_tracker.mock_calls[0], call.emit(3))
+
+    @patch("awslambdaric.lambda_multi_concurrent_utils.logging")
+    @patch("awslambdaric.lambda_multi_concurrent_utils.bootstrap")
+    def test_emit_worker_pool_event_sets_up_parent_logging_and_emits(
+        self, mock_bootstrap, mock_logging
+    ):
+        with patch.object(MultiConcurrentRunner, "_redirect_output") as mock_redirect:
+            MultiConcurrentRunner._emit_worker_pool_event(16)
+
+        # Parent never redirects: RAPID wires its stdout at spawn.
+        mock_redirect.assert_not_called()
+        mock_bootstrap.init_logging.assert_called_once_with()
+        mock_logging.getLogger.return_value.debug.assert_called_once()
+        event = mock_logging.getLogger.return_value.debug.call_args[0][0]
+        self.assertEqual(event["workerCount"], 16)
+        self.assertEqual(event["executionEnvironmentMaxConcurrency"], 16)
+        mock_logging.getLogger.return_value.handlers.clear.assert_called_once_with()
+        # Sink is closed deterministically after the handler is removed.
+        mock_bootstrap.init_logging.return_value.__exit__.assert_called_once_with(
+            None, None, None
+        )
+
     @patch(
         "awslambdaric.lambda_multi_concurrent_utils.LambdaMultiConcurrentRuntimeClient"
     )