Loki处理JSON负载时抛出unmarshalerDecoder解码错误求助

错误信息 #

loghttp.PushRequest.Streams: []*loghttp.Stream: unmarshalerDecoder: Value looks like Number/Boolean/None, but can't find its end: ',' or '}' symbol, error found in #10 byte of ...| ]
      }
   ]
}|..., bigger context ...|
               }
            }
         ]
      }
   ]
}|...

上传的JSON负载 #

{
   "streams":[
      {
         "stream":{
            "application":"fabric-sso",
            "job":"aws-lambda",
            "level":"info",
            "namespace":"oauth"
         },
         "values":[
            {
               "ts":"2022-12-10T01:36:44.971933+05:30",
               "message":{
                  "type":"h2",
                  "timestamp":"2022-09-17T11:00:03.828554Z",
                  "alb":"app/fabric-sso/sdjsjhdjhdshksdhf",
                  "client_ip":"999.999.999.999",
                  "client_port":"7392",
                  "backend_ip":"",
                  "backend_port":"",
                  "request_processing_time":"-1",
                  "backend_processing_time":"-1",
                  "response_processing_time":"-1",
                  "alb_status_code":"404",
                  "backend_status_code":"-",
                  "received_bytes":"706",
                  "sent_bytes":"67",
                  "request_verb":"GET",
                  "request_url":"https://foozy.dev.gabbar.com:443/gabbar",
                  "request_proto":"HTTP/2.0",
                  "user_agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36",
                  "ssl_cipher":"ECDHE-RSA-AES128-GCM-SHA256",
                  "ssl_protocol":"TLSv1.2",
                  "target_group_arn":"-",
                  "trace_id":"Root=1-6325a8b3-1980ccbd244b83c35ec5b543",
                  "domain_name":"foozy.dev.gabbar.com",
                  "chosen_cert_arn":"arn:aws:acm:us-east-1:23232323:certificate/0000",
                  "matched_rule_priority":"0",
                  "request_creation_time":"2022-09-17T11:00:03.818000Z",
                  "actions_executed":"waf,fixed-response",
                  "redirect_url":"-",
                  "lambda_error_reason":"-",
                  "target_port_list":"-",
                  "target_status_code_list":"-",
                  "classification":"-",
                  "classification_reason":"-",
                  "application":"fabric-sso",
                  "env":"dev"
               }
            }
         ]
      }
   ]
}

构造负载的Python代码 #

def build_loki_request_payload(alb_log_data):
    entries = []
    for entry in alb_log_data:
        curr_datetime = datetime.datetime.now(pytz.timezone('Asia/Kolkata'))
        curr_datetime = curr_datetime.isoformat('T')
        entries.append({'ts': curr_datetime, 'message': entry})

    payload = {
        'streams': [{
            "stream": {
                "application": alb_log_data[0]['application'],
                "job": "aws-lambda",
                "level": "info",
                "namespace": "oauth"
            },
            "values": entries
        }]
    }
    payload = json.dumps(payload)
    logger.debug('Created Payload %s', payload)
    return payload

问题原因与解决方法 #

Loki的Push API对streams[].values字段有严格格式要求：每个元素必须是**[时间戳字符串, 日志内容字符串]**的二元数组，而非字典结构。当前代码中values的元素是字典{'ts': ..., 'message': ...}，完全不符合Loki的解码规则，因此触发错误。

修正后的代码 #

import json
import datetime
import pytz

def build_loki_request_payload(alb_log_data):
    entries = []
    for entry in alb_log_data:
        curr_datetime = datetime.datetime.now(pytz.timezone('Asia/Kolkata'))
        curr_datetime = curr_datetime.isoformat('T')
        # 将日志字典序列化为JSON字符串，作为values的第二个元素
        entries.append([curr_datetime, json.dumps(entry)])

    payload = {
        'streams': [{
            "stream": {
                "application": alb_log_data[0]['application'],
                "job": "aws-lambda",
                "level": "info",
                "namespace": "oauth"
            },
            "values": entries
        }]
    }
    payload = json.dumps(payload)
    logger.debug('Created Payload %s', payload)
    return payload

关键修改点 #

• 将entries中每个元素从字典改为二元数组，第一个值为时间戳字符串，第二个值为序列化后的日志JSON字符串
• 确保日志内容以字符串形式传递给Loki，保留结构化信息的同时符合API格式要求

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