解析不了 Kinesis 流中的 CloudWatch Logs

哈喽！今天心血来潮给大家带来了《解析不了 Kinesis 流中的 CloudWatch Logs》，想必大家应该对Golang都不陌生吧，那么阅读本文就都不会很困难，以下内容主要涉及到，若是你正在学习Golang，千万别错过这篇文章~希望能帮助到你！

我正在尝试通过 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\"}"
        }
    ]
}

解决方案

好吧，事实证明我不必从 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字符串

好了，本文到此结束，带大家了解了《解析不了 Kinesis 流中的 CloudWatch Logs》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！