AWS Lambda Python 런타임의 Logging 시스템 뜯어보기

bootstrap_1.py

# https://github.com/aws/aws-lambda-python-runtime-interface-client/blob/849e874de01776cb386c18fb2c1731d43cd2b2f4/awslambdaric/bootstrap.py#L478C1-L534C14

def run(app_root, handler, lambda_runtime_api_addr):
    sys.stdout = Unbuffered(sys.stdout)
    sys.stderr = Unbuffered(sys.stderr)

    use_thread_for_polling_next = os.environ.get("AWS_EXECUTION_ENV") in {
        "AWS_Lambda_python3.12",
        "AWS_Lambda_python3.13",
    }

    with create_log_sink() as log_sink:
        lambda_runtime_client = LambdaRuntimeClient(
            lambda_runtime_api_addr, use_thread_for_polling_next
        )
        error_result = None

        try:
            _setup_logging(_AWS_LAMBDA_LOG_FORMAT, _AWS_LAMBDA_LOG_LEVEL, log_sink)
            global _GLOBAL_AWS_REQUEST_ID

            request_handler = _get_handler(handler)
        except FaultException as e:
            error_result = make_error(
                e.msg,
                e.exception_type,
                e.trace,
            )
        except Exception:
            error_result = build_fault_result(sys.exc_info(), None)

        if error_result is not None:
            log_error(error_result, log_sink)
            lambda_runtime_client.post_init_error(error_result)

            sys.exit(1)

        if os.environ.get(AWS_LAMBDA_INITIALIZATION_TYPE) == INIT_TYPE_SNAP_START:
            on_init_complete(lambda_runtime_client, log_sink)

        while True:
            event_request = lambda_runtime_client.wait_next_invocation()

            _GLOBAL_AWS_REQUEST_ID = event_request.invoke_id

            update_xray_env_variable(event_request.x_amzn_trace_id)

            handle_event_request(
                lambda_runtime_client,
                request_handler,
                event_request.invoke_id,
                event_request.event_body,
                event_request.content_type,
                event_request.client_context,
                event_request.cognito_identity,
                event_request.invoked_function_arn,
                event_request.deadline_time_in_ms,
                log_sink,
            )

bootstrap_2.py

def _setup_logging(log_format, log_level, log_sink):
    logging.Formatter.converter = time.gmtime
    logger = logging.getLogger()

    if log_format == LogFormat.JSON or log_level:
        logger_handler = LambdaLoggerHandlerWithFrameType(log_sink)
    else:
        logger_handler = LambdaLoggerHandler(log_sink)

    if log_format == LogFormat.JSON:
        logger_handler.setFormatter(JsonFormatter())
    else:
        logger_handler.setFormatter(
            logging.Formatter(
                "[%(levelname)s]\t%(asctime)s.%(msecs)03dZ\t%(aws_request_id)s\t%(message)s\n",
                "%Y-%m-%dT%H:%M:%S",
            )
        )

    if log_level in logging._nameToLevel:
        logger.setLevel(log_level)

    logger_handler.addFilter(LambdaLoggerFilter())
    logger.addHandler(logger_handler)

bootstrap_3.py

# https://github.com/aws/aws-lambda-python-runtime-interface-client/blob/849e874de01776cb386c18fb2c1731d43cd2b2f4/awslambdaric/bootstrap.py#L478C1-L534C14

class LambdaLoggerFilter(logging.Filter):
    def filter(self, record):
        record.aws_request_id = _GLOBAL_AWS_REQUEST_ID or ""
        return True

bootstrap_4.py

# https://github.com/aws/aws-lambda-python-runtime-interface-client/blob/849e874de01776cb386c18fb2c1731d43cd2b2f4/awslambdaric/bootstrap.py#L153C1-L214C10

def handle_event_request(
    lambda_runtime_client,
    request_handler,
    invoke_id,
    event_body,
    content_type,
    client_context_json,
    cognito_identity_json,
    invoked_function_arn,
    epoch_deadline_time_in_ms,
    log_sink,
):
    error_result = None
    try:
        lambda_context = create_lambda_context(
            client_context_json,
            cognito_identity_json,
            epoch_deadline_time_in_ms,
            invoke_id,
            invoked_function_arn,
        )
        event = lambda_runtime_client.marshaller.unmarshal_request(
            event_body, content_type
        )
        response = request_handler(event, lambda_context)
        result, result_content_type = lambda_runtime_client.marshaller.marshal_response(
            response
        )
    except FaultException as e:
        xray_fault = make_xray_fault("LambdaValidationError", e.msg, os.getcwd(), [])
        error_result = make_error(
            e.msg,
            e.exception_type,
            e.trace,
            invoke_id,
        )

    except Exception:
        etype, value, tb = sys.exc_info()
        tb_tuples = extract_traceback(tb)
        for i in range(len(tb_tuples)):
            if "/bootstrap.py" not in tb_tuples[i][0]:  # filename of the tb tuple
                tb_tuples = tb_tuples[i:]
                break

        xray_fault = make_xray_fault(etype.__name__, str(value), os.getcwd(), tb_tuples)
        error_result = make_error(
            str(value), etype.__name__, traceback.format_list(tb_tuples), invoke_id
        )

    if error_result is not None:
        from .lambda_literals import lambda_unhandled_exception_warning_message

        log_sink.log(lambda_unhandled_exception_warning_message, _WARNING_FRAME_TYPE)
        log_error(error_result, log_sink)
        lambda_runtime_client.post_invocation_error(
            invoke_id, to_json(error_result), to_json(xray_fault)
        )
    else:
        lambda_runtime_client.post_invocation_result(
            invoke_id, result, result_content_type
        )