将Syslog转换为JSON的Python Lambda代码报错排查求助

这个错误核心是用字符串正则模式去匹配字节对象，哪怕你做了base64解码转UTF-8，大概率是某个环节漏了字符串转换，或者处理过程中又把字符串转回了字节。

常见问题点 #

• base64解码后没及时转成字符串，直接用字节对象喂给正则
• 中间预处理步骤（比如清洗日志内容）里，不小心把字符串转回了字节
• Lambda事件中的payload提取路径错误，拿到的不是预期的解码后数据

修正后的代码片段 #

import base64
import re
import json

def lambda_handler(event, context):
    for record in event['Records']:
        # 先解码base64，再转UTF-8字符串，用replace处理编码异常
        payload_bytes = base64.b64decode(record['kinesis']['data'])
        payload_str = payload_bytes.decode('utf-8', errors='replace')
        
        # 编译正则（可选，但复用更高效），匹配字符串对象
        syslog_regex = re.compile(r'<(\d+)>(.*)')
        match_result = syslog_regex.search(payload_str)
        
        if match_result:
            syslog_json = {
                'priority': int(match_result.group(1)),
                'message': match_result.group(2).strip()
            }
            print(json.dumps(syslog_json))
        else:
            print(f"不匹配Syslog格式: {payload_str[:50]}...")

额外排查技巧 #

• 加日志打印变量类型：比如print(type(payload_str))，确认是<class 'str'>而非<class 'bytes'>
• 如果日志有特殊编码，尝试用latin-1解码（Syslog有时会用这个编码）
• 检查Lambda事件结构：如果是SQS触发，payload路径是record['body']，而非Kinesis的record['kinesis']['data']

内容的提问来源于stack exchange，提问作者Divya