Fluentd配置Kafka用户名和密码_fluent的用法和搭配[通俗易懂]

Fluentd

Fluentd 是完全开源免费的日志收集系统，支持在超过125种不同类型的系统上收集日志。

其系统示意图如下：

安装

第一步，RPM包安装

在RedHat/CentOS 6, 7 64位系统上可以用下列方式安装：

curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent3.sh | sh

第二步，启动

根据系统版本可以用下列两种方式之一启动：

systemctl start td-agent

/etc/init.d/td-agent start

第三步，验证

默认安装的话，配置文件在/etc/td-agent/td-agent.conf，从HTTP的8888端口接收日志并路由到stdout(/var/log/td-agent/td-agent.log)。可以用下列命令发起测试日志：

curl -X POST -d 'json={"json":"message"}' http://localhost:8888/debug.test

查看日志的最后一行即是刚刚发送的测试日志：

# tail /var/log/td-agent/td-agent.log
2018-09-14 16:42:14 +0800 [warn]: #0 [output_td] secondary type should be same with primary one primary="Fluent::Plugin::TreasureDataLogOutput" secondary="Fluent::Plugin::FileOutput"
2018-09-14 16:42:14 +0800 [info]: adding match pattern="debug.**" type="stdout"
2018-09-14 16:42:14 +0800 [info]: adding source type="forward"
2018-09-14 16:42:14 +0800 [info]: adding source type="http"
2018-09-14 16:42:14 +0800 [info]: adding source type="debug_agent"
2018-09-14 16:42:14 +0800 [info]: #0 starting fluentd worker pid=24064 ppid=24051 worker=0
2018-09-14 16:42:14 +0800 [info]: #0 [input_debug_agent] listening dRuby uri="druby://127.0.0.1:24230" object="Fluent::Engine" worker=0
2018-09-14 16:42:14 +0800 [info]: #0 [input_forward] listening port port=24224 bind="0.0.0.0"
2018-09-14 16:42:14 +0800 [info]: #0 fluentd worker is now running worker=0
2018-09-14 16:46:16.388094455 +0800 debug.test: {"json":"message"}

配置

配置文件用于让用户控制Fluentd输入(input)和输出(output)的行为：(1) 选择输入和输出插件，(2) 指定插件参数。配置文件要求必须存在以使Fluentd正常工作。

字符编码

Fluentd假设配置文件编码为UTF-8或ASCII。

指令列表

配置文件由下列指令组成：

source 指令确定输入源。
match 指令确定输出目的地。
filter 指令确定事件处理管道。
system 指令设置系统级配置。
label 指令将output和filter分组以进行内部路由。
@include 指令用于包括其它文件。

配置示例 Step by Step

1. source: 所有数据的来源

选择和配置Fluentd的输入源，需要用source指令。Fluentd的标准输入插件包括http和forward。http监听HTTP端口，并接受从HTTP来的日志消息。forward让fluentd监听TCP端口，并接受TCP包。当然，两者可以同时开启（可以添加任意多需要的源）。

# Receive events from 24224/tcp
# This is used by log forwarding and the fluent-cat command
<source>
  @type forward
  port 24224
</source>

# http://this.host:9880/myapp.access?json={"event":"data"}
<source>
  @type http
  port 9880
</source>

每个source指令必须包含 @type 参数。@type 参数指定了何种输入插件将被使用。

插播一下：Routing

source 将事件提交到Fluentd的路由引擎。每个事件由3部分实体组成：tag, time 和 record。tag 是由’.’分隔的字符串(如：myapp.access)，并且作为Fluentd内部路由的指示。time字段由 input 插件指定，并且必须得是Unix时间格式。record 是一个JSON对象。

2. match: 告诉 fluentd 该干什么！

match 指令查找所有具有匹配tag的事件，并且进行处理。match 指令最常见的用法是将事件输出到其它系统（因此，对应于match指令的插件被称为”output”插件）。Fluentd的标准输出插件包括 file 和 forward。现在来加入到配置文件中。

# Receive events from 24224/tcp
# This is used by log forwarding and the fluent-cat command
<source>
  @type forward
  port 24224
</source>

# http://this.host:9880/myapp.access?json={"event":"data"}
<source>
  @type http
  port 9880
</source>

# Match events tagged with "myapp.access" and
# store them to /var/log/fluent/access.%Y-%m-%d
# Of course, you can control how you partition your data
# with the time_slice_format option.
<match myapp.access>
  @type file
  path /var/log/fluent/access
</match>

每个 match 指令必须包括一个匹配模板和一个 @type 参数。只有tag 匹配到模板的事件会被送到输出目的地（在上述示例中，只有tag为”myapp.access”的事件会被匹配）。@type 参数指定了将要使用的 output 插件。

3. filter: 事件处理管道

filter 指令与 match 有相同的语法，但 filter 可以将处理串成管道。使用 filter 时事件流看起来会像这样：

Input -> filter 1 -> ... -> filter N -> Output

现在添加一个标准的 record_transformer filter 到刚刚写的 match 示例中。

# http://this.host:9880/myapp.access?json={"event":"data"}
<source>
  @type http
  port 9880
</source>

<filter myapp.access>
  @type record_transformer
  <record>
    host_param "#{Socket.gethostname}"
  </record>
</filter>

<match myapp.access>
  @type file
  path /var/log/fluent/access
</match>

收到事件 {“event”: “data”} 后，先扔到filter: record_transformer中，record_transformer 在事件中添加了”host_param” 字段并且继续向后传，{“event”:“data”,“host_param”:“webserver1”}，事件最终到达 file 输出插件。

match 模板

通配符和展开

下列 match 模板可以用于<match> 和<filter>中的tag：

* 匹配单个tag部分。
- 例如：模板 a.* 匹配 a.b，但不能匹配 a 与 a.b.c
** 匹配0个或多个tag 部分
- 例如：a.** 匹配a, a.b 和 a.b.c
{X,Y,Z} 匹配X, Y, 或 Z, 这里 X, Y 和 Z都是匹配模板
- 例如：模板 {a,b} 匹配 a 和 b，但不匹配 c
- 这种方法可以用于组合 * 和 ** 模板。例如： a.{b,c}.* 和 a.{b,c.**}
当多个模板同时在单个tag中列出时（由一个或多个空格分隔），将匹配所列出的任意一个模板。例如：
- 模板 <match a b> 匹配 a 和 b
- 模板 <match a.** b.*> 匹配 a, a.b, a.b.c (匹配到前一个模板)和 b.d (匹配到后一个模板)

匹配顺序

Fluentd 尝试以模板在配置文件中出现的顺序进行匹配。所以看下面的例子：

# ** matches all tags. Bad :(
<match **>
  @type blackhole_plugin
</match>

<match myapp.access>
  @type file
  path /var/log/fluent/access
</match>

上面例子中，myapp.access 永远都匹配不到。更宽松的匹配模板应定义在更严格的匹配模板之后。像这样：

<match myapp.access>
  @type file
  path /var/log/fluent/access
</match>

# Capture all unmatched tags. Good :)
<match **>
  @type blackhole_plugin
</match>

当然，如果有两个同样的模板，后一个match将永远无法匹配。如果你希望将事件发送到多个输出，应考虑使用 out_copy 插件。

常见的陷阱是将<filter>放到<match>之后。由于前面说的原因，事件将无法到达filter，也就无法按预期的方式工作。

COPY 输出插件

out_copy 是Fluentd的核心插件，无须另外安装，out_copy 将事件复制多份，分别送到多个输出插件中去。直接来个例子：

<match pattern>
  @type copy
  <store>
    @type file
    path /var/log/fluent/myapp1
    ...
  </store>
  <store>
    ...
  </store>
  <store>
    ...
  </store>
</match>

下面的例子将事件存到本地文件夹/var/log/fluent/myapp下，同时发送到Elasticsearch实例的fluentd.test 集合中。具体请参见out_file和out_elasticsearch插件的细节说明。

<match docker.*>
  @type copy
  <store>
    @type file
    path /var/log/fluent/myapp
    compress gzip
    <format>
      localtime false
    </format>
    <buffer time>
      timekey_wait 10m
      timekey 86400
      timekey_use_utc true
      path /var/log/fluent/myapp
    </buffer>
    <inject>
      time_format %Y%m%dT%H%M%S%z
      localtime false
    </inject>
  </store>
  <store>
    @type elasticsearch
    host 192.168.198.46
    port 9200
    index_name fluentd
    type_name test
  </store>
</match>

启动一个docker，并将log-driver指向fluentd，即可将日志输出到fluentd，再转发至elasticsearch，同时本地留有日志备份。

docker run --rm --log-driver fluentd --log-opt fluentd-address=192.168.198.46:24224 --log-opt tag=docker.test alpine date

out_copy 参数

@type – 此值必须为”copy”
deep_copy – 深度复制，默认为false，out_copy在各store插件间共享事件记录。当此值为true时，out_copy为每一store插件复制一份事件记录。
<store> 字段 – 指定存储目的地。格式与<match>指令一致。此字段必须至少出现一次。
ignore_error – 忽略错误。当某一store出现错误时，这个参数将作用于其余store，例如：
```
<match app.**>
  @type copy
  <store>
    @type plugin1
  </store>
  <store>
    @type plugin2
  </store>
</match>
```
若plugin1出现错误时，plugin2将不会被执行。若希望忽略不重要的store出现的错误时，可以指定该store的ignore_error参数。
```
<match app.**>
  @type copy
  <store ignore_error>
    @type plugin1
  </store>
  <store>
    @type plugin2
  </store>
</match>
```

今天的文章Fluentd配置Kafka用户名和密码_fluent的用法和搭配[通俗易懂]分享到此就结束了，感谢您的阅读。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至举报，一经查实，本站将立刻删除。
如需转载请保留出处：https://bianchenghao.cn/79114.html