如何从 Kinesis Firehose 向 Splunk 发送非 cloudwatch JSON 事件？答案

【问题标题】：How do I send non-cloudwatch JSON events to Splunk from Kinesis Firehose?如何从 Kinesis Firehose 向 Splunk 发送非 cloudwatch JSON 事件？
【发布时间】：2020-08-12 17:37:20
【问题描述】：

我正在尝试从 Kinesis Firehose 向 Splunk 发送非 cloudwatch 事件。我正在使用 Lambda 处理事件并以下列格式 (required for Firehose) 将其反馈回 Firehose：

{ 
    "records": [
        {
          "recordId": "2345678",
          "result": "Ok",
          "data": [base64-encoded custom JSON]
        }
    ]
}

但是，一旦到达 Splunk，它就会抛出一个模糊的解析错误，并且帮助链接无处可去：

"errorCode":"Splunk.InvalidDataFormat","errorMessage":"The data is not formatted correctly. To see how to properly format data for Raw or Event HEC endpoints, see Splunk Event Data (http://dev.splunk.com/view/event-collector/SP-CAAAE6P#data)"

我在这里缺少什么？ HEC 端点无法以标准格式解析来自 Firehose 的消息，这似乎很奇怪。

我正在使用 aws_kinesis_firehose_delivery_stream Terraform module 中的 splunk_configuration 块将消息发送到 HEC 事件端点。

【问题讨论】：

标签： json parsing terraform splunk amazon-kinesis-firehose

【解决方案1】：

想通了！对于后代，因为这没有很好的记录：

您在 Kinesis Firehose 负载中的 data 字段必须是一个 base64 编码的对象，跟在 Splunk event collector spec 之后。

只要两者 Firehose 和 Splunk 都可以读取 Lambda 返回的负载，它不应该抛出错误。

这是 Kinesis Firehose 转换器 Lambda（node12 运行时）的代码：

/*
* Transformer for sending Kinesis Firehose events to Splunk
*
* Properly formats incoming messages for Splunk ingestion
* Returned object gets fed back into Kinesis Firehose and sent to Splunk
*/

'use strict';
console.log('Loading function');

exports.handler = (event, context, callback) => {
    let success = 0; // Number of valid entries found
    let failure = 0; // Number of invalid entries found
    let dropped = 0; // Number of dropped entries

    /* Process the list of records and transform them to adhere to Splunk specs */
    const output = event.records.map((record) => {
        try {
            const entry = (Buffer.from(record.data, 'base64')).toString('utf8');

            /*
             * IMPORTANT: `data` object should follow Splunk event formatting specs prior to encoding.
             * Otherwise, it will throw a parsing error.
             * https://docs.splunk.com/Documentation/Splunk/8.0.3/Data/FormateventsforHTTPEventCollector
             */
            const obj = {
                sourcetype: "aws:firehose:json", // Required, will error
                event: JSON.parse(entry)
            }
            const payload = (Buffer.from(JSON.stringify(obj), 'utf8')).toString('base64');
            success++;
            return {
                recordId: record.recordId,
                result: 'Ok',
                data: payload,
            };
        } catch (e) {
            failure++
            console.error(e.message());
            return {
                recordId: record.recordId,
                result: 'ProcessingFailed'
            };
        }
    });
    console.log(`Processing completed.  Successful records ${success}. Failed records ${failure}.`);
    callback(null, {records: output});
}

【讨论】：