Flume代理：将主机添加到消息，然后发布到kafka主题

2 回答

• 1

  如果您正在使用 exec 源，则不会阻止您运行智能命令以将主机名添加到日志文件内容前面 .

  注意：如果命令使用管道之类的东西，你还需要像这样指定shell：

  tier1.sources.source1.type = exec
  tier1.sources.source1.shell = /bin/sh -c
  tier1.sources.source1.command =  tail -F /var/log/auth.log | sed --unbuffered "s/^/$(hostname) /"
  

  消息看起来像这样：

  frb.hi.inet 2015-11-17 08:39:39.432 INFO [...]
  

  ... frb.hi.inet 我们主人的名字 .

  回复于 2024-04-26T05:37:03+08:00
• 2

  您可以创建一个自定义TCP源，它读取客户端地址并将其添加到标头中 .

  @Override
      public void configure(Context context) {
          port = context.getInteger("port");
          buffer = context.getInteger("buffer");
  
          try{
              serverSocket = new ServerSocket(port);
              logger.info("FlumeTCP source initialized");
          }catch(Exception e) {
              logger.error("FlumeTCP source failed to initialize");
          }
      }
  
  @Override
      public void start() {
          try {
              clientSocket = serverSocket.accept();
              receiveBuffer = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
              logger.info("Connection established with client : " + clientSocket.getRemoteSocketAddress());
              final ChannelProcessor channel = getChannelProcessor();
              final Map<String, String> headers = new HashMap<String, String>();
              headers.put("hostname", clientSocket.getRemoteSocketAddress().toString());
              String line = "";
              List<Event> events = new ArrayList<Event>();
  
              while ((line = receiveBuffer.readLine()) != null) {
                  Event event = EventBuilder.withBody(
                          line, Charset.defaultCharset(),headers);
  
                  logger.info("Event created");
                  events.add(event);
                  if (events.size() == buffer) {
                      channel.processEventBatch(events);
                  }
              }
          } catch (Exception e) {
  
          }
          super.start();
      }
  

  flume-conf.properties可以配置为：

  # Licensed to the Apache Software Foundation (ASF) under one
  # or more contributor license agreements.  See the NOTICE file
  # distributed with this work for additional information
  # regarding copyright ownership.  The ASF licenses this file
  # to you under the Apache License, Version 2.0 (the
  # "License"); you may not use this file except in compliance
  # with the License.  You may obtain a copy of the License at
  #
  #  http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing,
  # software distributed under the License is distributed on an
  # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  # KIND, either express or implied.  See the License for the
  # specific language governing permissions and limitations
  # under the License.
  
  
  # The configuration file needs to define the sources, 
  # the channels and the sinks.
  # Sources, channels and sinks are defined per agent, 
  # in this case called 'agent'
  
  agent.sources = CustomTcpSource
  agent.channels = memoryChannel
  agent.sinks = loggerSink
  
  # For each one of the sources, the type is defined
  agent.sources.CustomTcpSource.type = com.vishnu.flume.source.CustomFlumeTCPSource
  agent.sources.CustomTcpSource.port = 4443
  agent.sources.CustomTcpSource.buffer = 1
  
  
  # The channel can be defined as follows.
  agent.sources.CustomTcpSource.channels = memoryChannel
  
  # Each sink's type must be defined
  agent.sinks.loggerSink.type = logger
  
  #Specify the channel the sink should use
  agent.sinks.loggerSink.channel = memoryChannel
  
  # Each channel's type is defined.
  agent.channels.memoryChannel.type = memory
  
  # Other config values specific to each type of channel(sink or source)
  # can be defined as well
  # In this case, it specifies the capacity of the memory channel
  agent.channels.memoryChannel.capacity = 100
  

  我发送了一条测试消息来测试它，它看起来像：

  Event: { headers:{hostname=/127.0.0.1:50999} body: 74 65 73 74 20 6D 65 73 73 61 67 65             test message }
  

  我在github上传了这个项目

  回复于 2024-04-26T05:37:03+08:00