Logstash:处理多个input

Posted fat_girl_spring

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Logstash:处理多个input相关的知识,希望对你有一定的参考价值。

我们知道Logstash的架构如下:

  它的整个pipleline分为三个部分:
    input插件:提取数据。 这可以来自日志文件,TCP或UDP侦听器,若干协议特定插件(如syslog或IRC)之一,甚至是排队系统(如Redis,AQMP或Kafka)。 此阶段使用围绕事件来源的元数据标记传入事件。
    filter 插件:插件转换并丰富数据
    output插件: 将已处理的事件加载到其他内容中,例如ElasticSearch或其他文档数据库,或排队系统,如Redis,AQMP或Kafka。 它还可以配置为与API通信。 也可以将像PagerDuty这样的东西连接到Logstash输出。

这里的input可以支持多个input,同时多个worker可以处理filter及output:

  在今天的介绍中,我们来介绍一下如何使用多个input。

应用文件

为了说明问题的方便,我把所需要用到的文件都传到github地址https://github.com/liu-xiao-guo/logstash_multi-input。我们可以通过如下的方式来下载这些文件:

git clone https://github.com/liu-xiao-guo/logstash_multi-input

Logstash配置文件

Logstash的配置文件如下:

multi-input.conf

input {
  file {
    path => "/Users/liuxg/data/multi-input/apache.log"
      start_position => "beginning"
    sincedb_path => "/dev/null"
    # ignore_older => 100000
    type => "apache"
  }
}
 
input {
  file {
    path => "/Users/liuxg/data/multi-input/apache-daily-access.log"
      start_position => "beginning"
    sincedb_path => "/dev/null"
    type => "daily"
  }
}
 
filter {
      grok {
        match => {
              "message" => \'%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent}\'
        }
      }
 
    if[type] == "apache" {
        mutate {
              add_tag => ["apache"]
          }
    }
 
    if [type] == "daily" {
        mutate {
            add_tag => ["daily"]
        }
    } 
}
 
 
output {
    stdout {
        codec => rubydebug
    }
 
    if "apache" in [tags] {
          elasticsearch {
            index => "apache_log"
            template => "/Users/liuxg/data/apache_template.json"
            template_name => "apache_elastic_example"
            template_overwrite => true
      }    
    }
 
    if "daily" in [tags] {
          elasticsearch {
            index => "apache_daily"
            template => "/Users/liuxg/data/apache_template.json"
            template_name => "apache_elastic_example"
            template_overwrite => true
      }    
    }    
}

为了说明问题的方便,我们使用了两个input。它们分别对应不同的log文件。对于这两个input,我们也使用了不同的type来表示:apache和daily。尽管它们的格式是一样的,它们共同使用同样的一个grok filter,但是我们还是想分别对它们进行处理。为此,我们添加了一个tag。我们也可以添加一个field来进行区别。在output的部分,我们根据在filter部分设置的tag来对它们输出到不同的index里。

运行Logstash

$ pwd
/Users/liuxg/elastic/logstash-7.3.0
bogon:logstash-7.3.0 liuxg$ sudo ./bin/logstash -f ~/data/multi-input/multi-input.conf

当你们运行这个例子的时候,你们需要根据自己存放multi-input.conf文件的位置改变而改变上面的命令。

运行的结果如下:

  根据显示的结果可以看出来daily的事件最早被处理及输出。接着apache的数据才开始处理。在实际的应用中,我们可能有不同的数据源,比如来自其它beats的监听某个端口的数据。

我们可以在Kibana中看到我们最终的index数据:

转载自:https://blog.csdn.net/UbuntuTouch/java/article/details/100980709

以上是关于Logstash:处理多个input的主要内容,如果未能解决你的问题,请参考以下文章

Logstash:多个配置文件(conf)

Logstash:多个配置文件(conf)

原创大叔经验分享(82)logstash一个实例运行多个配置文件

ELK 学习笔记之 Logstash基本语法

logstash-input-jdbc 同时同步多个表的情况

logstash-input-jdbc配置说明