如何配置 FLUME-NG 根据日志发送者 IP 将日志写入不同的日志文件？

Question

我是 FLUME 的新手。刚开始使用它（flume-ng）。目前我有多个应用程序服务器在不同的服务器上运行。我想收集这些服务器的日志。我在独立的 LINUX 计算机上安装 FLUME。但我不知道如何配置 FLUME 以根据应用程序服务器 IP 地址将日志写入本地文件。或者如何根据我定义的不同类别来写日志？

Answer 1

Flume 有一个多路复用流的选项。这是 Flume-Ng 用户指南 (https://flume.apache.org/FlumeUserGuide.html) 中的一个示例

# list the sources, sinks and channels in the agent
agent_foo.sources = avro-AppSrv-source1
agent_foo.sinks = hdfs-Cluster1-sink1 avro-forward-sink2
agent_foo.channels = mem-channel-1 file-channel-2

# set channels for source
agent_foo.sources.avro-AppSrv-source1.channels = mem-channel-1 file-channel-2

# set channel for sinks
agent_foo.sinks.hdfs-Cluster1-sink1.channel = mem-channel-1
agent_foo.sinks.avro-forward-sink2.channel = file-channel-2

# channel selector configuration
agent_foo.sources.avro-AppSrv-source1.selector.type = multiplexing
agent_foo.sources.avro-AppSrv-source1.selector.header = State
agent_foo.sources.avro-AppSrv-source1.selector.mapping.CA = mem-channel-1
agent_foo.sources.avro-AppSrv-source1.selector.mapping.AZ = file-channel-2
agent_foo.sources.avro-AppSrv-source1.selector.mapping.NY = mem-channel-1 file-channel-2
agent_foo.sources.avro-AppSrv-source1.selector.default = mem-channel-1

Answer 2

您可以使用水槽通道选择器将事件简单地路由到不同的目的地。或者您可以将多个 Flume 代理链接在一起以实现复杂的路由功能。 但是链式水槽代理将变得有点难以维护（资源使用和水槽拓扑）。 你可以看看flume-ng router sink，它可能会提供你想要的一些功能。

首先，通过flume interceptor在事件头中添加特定字段

a1.sources = r1 r2
a1.channels = c1 c2
a1.sources.r1.channels =  c1
a1.sources.r1.type = seq
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = ip
a1.sources.r1.interceptors.i1.value = 192.168.0.12
a1.sources.r2.channels =  c2
a1.sources.r2.type = seq
a1.sources.r2.interceptors = i2
a1.sources.r2.interceptors.i2.type = static
a1.sources.r2.interceptors.i2.key = ip
a1.sources.r2.interceptors.i2.value = 192.168.0.25

然后，您可以像这样设置 avro-router sink：

agent.sinks.routerSink.type = com.datums.stream.AvroRouterSink
agent.sinks.routerSink.hostname = test_host
agent.sinks.routerSink.port = 34541
agent.sinks.routerSink.channel = memoryChannel

# Set sink name
agent.sinks.routerSink.component.name = AvroRouterSink

# Set header name for routing
agent.sinks.routerSink.condition = ip

# Set routing conditions
agent.sinks.routerSink.conditions = east,west
agent.sinks.routerSink.conditions.east.if = ^192.168.0.12
agent.sinks.routerSink.conditions.east.then.hostname = test_host
agent.sinks.routerSink.conditions.east.then.port = 34542
agent.sinks.routerSink.conditions.west.if = ^192.168.0.25
agent.sinks.routerSink.conditions.west.then.hostname = test_host
agent.sinks.routerSink.conditions.west.then.port = 34543

最后，还有一个代理用于接收事件并将它们写入不同的日志文件

agent.sources = src1 src2 src3
agent.channels = c1 c2 c3
agent.sinks = logger1 logger2 logger3

agent.sources.src1.type = avro
agent.sources.src1.bind = test_host
agent.sources.src1.port = 34531
agent.sources.src1.channel = c1

agent.sources.src2.type = avro
agent.sources.src2.bind = test_host
agent.sources.src2.port = 34532
agent.sources.src2.channel = c2

agent.sources.src3.type = avro
agent.sources.src3.bind = test_host
agent.sources.src3.port = 34533
agent.sources.src3.channel = c3

agent.sinks.logger1.type = logger
agent.sinks.logger1.channel = c1

agent.sinks.logger2.type = logger
agent.sinks.logger2.channel = c2

agent.sinks.logger3.type = logger
agent.sinks.logger3.channel = c3