使用 Kinesis 代理写入亚马逊 Data Firehose

Amazon Kinesis 代理是一款独立的 Java 软件应用程序，可作为参考实现，展示如何收集数据并将其发送到 Firehose。代理会持续监控一组文件，并将新数据发送到您的 Firehose 传输流。该代理显示了如何处理文件轮换、检查点操作和失败时重试。它展示了如何以可靠、及时和简单的方式交付数据。它还显示了如何发布 CloudWatch 指标以更好地监控流媒体过程并对其进行故障排除。要了解更多信息，请访问 awslabs/ amazon-kinesis-agent。

默认情况下，会基于换行符 ('\n') 分析每个文件中的记录。但是，也可以将代理配置为分析多行记录（请参阅 代理配置设置）。

您可以在基于 Linux 的服务器环境（如 Web 服务器、日志服务器和数据库服务器）上安装此代理。安装代理后，通过指定要监控的文件和数据的 Firehose 流来对其进行配置。配置代理后，它会持久地从文件中收集数据，并可靠地将其发送到 Firehose 数据流。

先决条件

凭证

使用以下方法之一管理您的 Amazon 凭证：

自定义凭证提供程序

您可以创建自定义凭证提供程序，并在以下配置设置中为 Kinesis 代理提供其类名和 jar 路径：userDefinedCredentialsProvider.classname 和 userDefinedCredentialsProvider.location。有关这两个配置设置的说明，请参阅代理配置设置。

要创建自定义凭证提供程序，请定义一个实现 AmazonCredentialsProvider 接口的类，如下例所示。

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;

public class YourClassName implements AWSCredentialsProvider {
    public YourClassName() {
    }

    public AWSCredentials getCredentials() {
        return new BasicAWSCredentials("key1", "key2");
    }

    public void refresh() {
    }
}

您的类必须有一个不带参数的构造函数。

Amazon 会定期调用刷新方法以获取更新的凭证。如果希望凭证提供程序在其整个生命周期内提供不同的凭证，请在此方法中包含用于刷新凭证的代码。或者，如果您需要提供静态（不更改）凭证的凭证提供程序，则可以将此方法保留为空。

下载并安装代理

首先，连接到您的实例。有关更多信息，请参阅适用于 Linux 实例的 Amazon EC2 用户指南中的连接到您的实例。如果您在连接时遇到问题，请参阅《适用于 Linux 实例的 Amazon EC2 用户指南》中的实例连接问题排查。

接下来，请使用以下方法之一安装代理。

配置并启动代理

现在，代理作为系统服务在后台运行。它会持续监视指定的文件并将数据发送到指定的 Firehose 流。代理活动记录在 /var/log/aws-kinesis-agent/aws-kinesis-agent.log 中。

代理配置设置

代理支持两个必需的配置设置，即 filePattern 和 deliveryStream，以及可用于其他功能的可选配置设置。您可以在 /etc/aws-kinesis/agent.json 中指定必需配置设置和可选配置设置。

只要您更改配置文件，就必须使用以下命令停止再启动代理：

sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start

或者，您也可以使用以下命令：

sudo service aws-kinesis-agent restart

一般的设置配置如下。

流配置设置如下。

监控多个文件目录并写入多个流

通过指定多个流程配置设置，您可以配置代理以监控多个文件目录并将数据发送到多个流。在以下配置示例中，代理监控两个文件目录，并将数据分别发送到 Kinesis 数据流和 Firehose 流。您可以为 Kinesis Data Streams 和 Amazon Data Firehose 指定不同的终端节点，这样您的数据流和 Firehose 流就不必位于同一个区域。

{
    "cloudwatch.emitMetrics": true,
    "kinesis.endpoint": "https://your/kinesis/endpoint", 
    "firehose.endpoint": "https://your/firehose/endpoint", 
    "flows": [
        {
            "filePattern": "/tmp/app1.log*", 
            "kinesisStream": "yourkinesisstream"
        }, 
        {
            "filePattern": "/tmp/app2.log*",
            "deliveryStream": "yourfirehosedeliverystream" 
        }
    ] 
}

有关在 Amazon Kinesis Data Streams 中使用代理的更多详细信息，请参阅使用 Kinesis 代理写入 Amazon Kinesis Data Streams。

使用代理预处理数据

代理可以预处理从受监控文件中解析的记录，然后再将其发送到您的 Firehose 直播中。通过将 dataProcessingOptions 配置设置添加到您的文件流可以启用此功能。可以添加一个或多个处理选项，这些选项将按指定的顺序执行。

代理支持以下处理选项。由于代理是开源的，您可以进一步开发和扩展其处理选项。您可以从 Kinesis 代理下载代理。

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG"
}

64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291

{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG",
    "customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}

{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}

{ 
    "flows": [
        {
            "filePattern": "/tmp/app.log*", 
            "deliveryStream": "my-delivery-stream",
            "dataProcessingOptions": [
                {
                    "optionName": "LOGTOJSON",
                    "logFormat": "COMMONAPACHELOG"
                }
            ]
        }
    ] 
}

{ 
    "flows": [
        {
            "filePattern": "/tmp/app.log*", 
            "deliveryStream": "my-delivery-stream",
            "multiLineStartPattern": "\\[SEQUENCE=",
            "dataProcessingOptions": [
                {
                    "optionName": "SINGLELINE"
                },
                {
                    "optionName": "CSVTOJSON",
                    "customFieldNames": [ "field1", "field2", "field3" ],
                    "delimiter": "\\t"
                }
            ]
        }
    ] 
}

{
    "optionName": "LOGTOJSON",
    "logFormat": "COMMONAPACHELOG",
    "matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
    "customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}

123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200

{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}

代理 CLI 命令

系统启动时自动启动代理：

sudo chkconfig aws-kinesis-agent on

检查代理的状态：

sudo service aws-kinesis-agent status

停止代理：

sudo service aws-kinesis-agent stop

从此位置读取代理的日志文件：

/var/log/aws-kinesis-agent/aws-kinesis-agent.log

卸载代理：

sudo yum remove aws-kinesis-agent

常见问题解答

有没有适用于 Windows 的 Kinesis 代理？

适用于 Windows 的 Kinesis 代理与适用于 Linux 平台的 Kinesis 代理是不同的软件。

为什么 Kinesis 代理速度变慢且/或 RecordSendErrors 增加？

这通常是由于 Kinesis 节流造成的。查看 Kinesis Data Streams 的WriteProvisionedThroughputExceeded指标或 Firehose 直播ThrottledRecords的指标。这些指标从 0 开始的任何增加都表示需要增加流限制。有关更多信息，请参阅 Kinesis 数据流限制和 Firehose 数据流。

排除节流后，查看 Kinesis 代理是否配置为跟踪大量小文件。Kinesis 代理跟踪新文件时会有延迟，因此 Kinesis 代理应跟踪少量大文件。尝试将您的日志文件合并为大文件。

为什么我会收到 java.lang.OutOfMemoryError 异常？

Kinesis 代理没有足够的内存来处理当前的工作负载。尝试增加 /usr/bin/start-aws-kinesis-agent 中的 JAVA_START_HEAP 和 JAVA_MAX_HEAP，并重新启动代理。

为什么我会收到 IllegalStateException : connection pool shut down 异常？

Kinesis 代理没有足够的连接来处理当前的工作负载。尝试在 /etc/aws-kinesis/agent.json 的常规代理配置设置中增加 maxConnections 和 maxSendingThreads。这些字段的默认值是可用运行时系统处理器的 12 倍。有关高级代理配置设置的更多信息，请参见 AgentConfiguration.java。

如何调试 Kinesis 代理的其他问题？

可在 /etc/aws-kinesis/log4j.xml 中启用 DEBUG 级别日志。

我应该如何配置 Kinesis 代理？

maxBufferSizeBytes 越小，Kinesis 代理发送数据的频率就越高。这可能是好事，因为这减少了记录的传输时间，但也增加了每秒对 Kinesis 的请求。

为什么 Kinesis 代理会发送重复记录？

这是由于文件跟踪中的错误配置造成的。确保每个 fileFlow’s filePattern 只匹配一个文件。如果正在使用的 logrotate 模式处于 copytruncate 模式，也可能发生这种情况。尝试将模式更改为默认模式或创建模式以避免重复。有关处理重复记录的更多信息，请参阅处理重复记录。