diff --git a/RELEASE.CHANGELOG.md b/RELEASE.CHANGELOG.md index 9153a5b..a3fe32c 100644 --- a/RELEASE.CHANGELOG.md +++ b/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. diff --git a/awslambdaric/__init__.py b/awslambdaric/__init__.py index 96b7844..f00368b 100644 --- a/awslambdaric/__init__.py +++ b/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" diff --git a/awslambdaric/bootstrap.py b/awslambdaric/bootstrap.py index aa3ae65..2f36177 100644 --- a/awslambdaric/bootstrap.py +++ b/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) diff --git a/awslambdaric/lambda_multi_concurrent_utils.py b/awslambdaric/lambda_multi_concurrent_utils.py index b6678d9..6440844 100644 --- a/awslambdaric/lambda_multi_concurrent_utils.py +++ b/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,21 @@ def run_single( client = LambdaMultiConcurrentRuntimeClient(api_addr, use_thread) bootstrap.run(handler, client) + @classmethod + def _emit_worker_pool_event(cls, socket_path: str, max_concurrency: int): + """Emit worker pool DEBUG event once from the parent before forking.""" + if socket_path: + cls._redirect_output(socket_path) + bootstrap.init_logging() + logging.getLogger().debug( + { + "event": WORKER_POOL_INITIALIZING_EVENT, + "workerCount": max_concurrency, + "executionEnvironmentMaxConcurrency": max_concurrency, + } + ) + logging.getLogger().handlers.clear() + @classmethod def run_concurrent( cls, @@ -41,6 +59,8 @@ def run_concurrent( socket_path: str, max_concurrency: int, ): + cls._emit_worker_pool_event(socket_path, max_concurrency) + processes = [] for _ in range(max_concurrency): p = multiprocessing.Process( diff --git a/awslambdaric/lambda_runtime_log_utils.py b/awslambdaric/lambda_runtime_log_utils.py index 9ddbcfb..dff0387 100644 --- a/awslambdaric/lambda_runtime_log_utils.py +++ b/awslambdaric/lambda_runtime_log_utils.py @@ -117,7 +117,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), diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index f4d78b6..43888c5 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -1540,6 +1540,80 @@ 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") + + class TestBootstrapModule(unittest.TestCase): def test_run(self): expected_handler = "app.my_test_handler" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index bc104fd..61ceeca 100644 --- a/tests/test_concurrency.py +++ b/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, diff --git a/tests/test_multi_concurrent_runner.py b/tests/test_multi_concurrent_runner.py index 4a9d023..678e6b4 100644 --- a/tests/test_multi_concurrent_runner.py +++ b/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,53 @@ 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("/sock", 3) + self.assertEqual(order_tracker.mock_calls[0], call.emit("/sock", 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("/sock", 16) + + mock_redirect.assert_called_once_with("/sock") + 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() + + @patch("awslambdaric.lambda_multi_concurrent_utils.logging") + @patch("awslambdaric.lambda_multi_concurrent_utils.bootstrap") + def test_emit_worker_pool_event_skips_redirect_when_no_socket( + self, mock_bootstrap, mock_logging + ): + with patch.object(MultiConcurrentRunner, "_redirect_output") as mock_redirect: + MultiConcurrentRunner._emit_worker_pool_event(None, 4) + + mock_redirect.assert_not_called() + mock_bootstrap.init_logging.assert_called_once_with() + mock_logging.getLogger.return_value.debug.assert_called_once() + @patch( "awslambdaric.lambda_multi_concurrent_utils.LambdaMultiConcurrentRuntimeClient" )