如何配置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-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接收器：

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