无法从 Kinesis 流解码 CloudWatch Logs答案

【问题标题】：Unable to decode CloudWatch Logs from Kinesis stream无法从 Kinesis 流解码 CloudWatch Logs
【发布时间】：2020-11-10 09:55:07
【问题描述】：

我正在尝试通过 Kinesis“汇集”我的 cloudwatch 日志，然后将其发送到 lambda 进行处理，但是我找不到解码/解析传入日志的方法。到目前为止，我已经尝试过：

方法 1 使用 cloudwatch“类”

func function(request events.KinesisEvent) error {
    for _, record := range request.Records {
        fmt.Println(record.EventName)
        fmt.Println(string(record.Kinesis.Data))

        rawData := events.CloudwatchLogsRawData{
            Data: string(record.Kinesis.Data),
        }

        parse, err := rawData.Parse()
        fmt.Println(parse)
        fmt.Println(err)
    }
    return nil
}

func main() {
    lambda.Start(function)
}

方法2手动解码

var logData events.CloudwatchLogsData

func Base64Decode(message []byte) (b []byte, err error) {
    var l int
    b = make([]byte, base64.StdEncoding.DecodedLen(len(message)))
    l, err = base64.StdEncoding.Decode(b, message)
    if err != nil {
        return
    }
    return b[:l], nil
}

func Parse(rawData []byte, d events.CloudwatchLogsData) (err error) {
    data, err := Base64Decode(rawData)
    if err != nil {
        return
    }

    zr, err := gzip.NewReader(bytes.NewBuffer(data))
    if err != nil {
        return
    }
    defer zr.Close()
    fmt.Println(zr)
    dec := json.NewDecoder(zr)
    err = dec.Decode(&d)

    return
}

func function(request events.KinesisEvent) error {
    for _, record := range request.Records {
        fmt.Println(record.EventName)
        fmt.Println(string(record.Kinesis.Data))

        err = Parse(record.Kinesis.Data, logData)
        fmt.Println(err)
        fmt.Println(logData)
    }
    return nil
}

func main() {
    lambda.Start(function)
}

他们两个我都得到同样的错误：

illegal base64 data at input byte 0

因此，据我了解，在 Base64 中接收并压缩的日志格式，但我无法在网上找到专门针对 Go 的任何内容。

编辑：

添加 logData 类型

// CloudwatchLogsData is an unmarshal'd, ungzip'd, cloudwatch logs event
type CloudwatchLogsData struct {
    Owner               string                   `json:"owner"`
    LogGroup            string                   `json:"logGroup"`
    LogStream           string                   `json:"logStream"`
    SubscriptionFilters []string                 `json:"subscriptionFilters"`
    MessageType         string                   `json:"messageType"`
    LogEvents           []CloudwatchLogsLogEvent `json:"logEvents"`
}

Base64 解码解压后的数据格式为 JSON，结构如下：（根据 AWS：https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html）

{
    "owner": "111111111111",
    "logGroup": "logGroup_name",
    "logStream": "111111111111_logGroup_name_us-east-1",
    "subscriptionFilters": [
        "Destination"
    ],
    "messageType": "DATA_MESSAGE",
    "logEvents": [
        {
            "id": "31953106606966983378809025079804211143289615424298221568",
            "timestamp": 1432826855000,
            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
        },
        {
            "id": "31953106606966983378809025079804211143289615424298221569",
            "timestamp": 1432826855000,
            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
        },
        {
            "id": "31953106606966983378809025079804211143289615424298221570",
            "timestamp": 1432826855000,
            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
        }
    ]
}

【问题讨论】：

你能举一个你试图解码的日志数据的例子吗？
@Marcin 对不起，我忘了。我已经在描述中添加了，最后
你能否也发布实际的json 消息。
@Marcin 也添加了
好的，我想我已经解决了，还是谢谢！

标签： amazon-web-services go amazon-cloudwatch amazon-kinesis

【解决方案1】：

好的，原来我不必从base64解码，只需解压缩数据



func Unzip(data []byte) error {
    rdata := bytes.NewReader(data)
    r, err := gzip.NewReader(rdata)
    if err != nil {
        return err
    }
    uncompressedData, err := ioutil.ReadAll(r)
    if err != nil {
        return err
    }
    fmt.Println(string(uncompressedData))
    return nil
}

uncompressedData 是 cloudwatch 日志的 JSON 字符串

【讨论】：