Logstash官方文档中文版-建立一个Logstash管道

向右看齐 2022-05-18 06:42 365阅读 0赞

## **创建一个Logstash管道** ##

Logstash管道在大多数用例有一个或多个输入,过滤器,和输出插件。本节中的场景构建Logstash配置文件指定这些插件和讨论每个插件都做什么。

Logstash配置文件定义了你Logstash管道。当你开始一个Logstash实例,使用 **- f< path/to/file>**选项指定的配置文件,它定义了实例的管道。

Logstash 管道必要的两个元素，**input**和**output**，和一个可选的元素，**filter**。输入插件将数据源的数据输入，过滤器插件主要修改数据成为您指定数据，输出插件主要将数据写入到另一个目的源。

![Logstash pipeline][]

以下文字代表了一个基本管道配置文件的架构：

# The # character at the beginning of a line indicates a comment. Use
    # comments to describe your configuration.
    input {
    }
    # The filter part of this file is commented out to indicate that it is
    # optional.
    # filter {
    #
    # }
    output {
    }

这个构架没有任何功能，因为输入和输出没有任何有效的定义。本教程的示例构建的配置文件用来解决特定的用例。

将基本架构粘贴到你Logstash根目录下的一个名为 **first-pipeline.conf**的文件中。（没有的话创建一个）

### **解析成Elasticsearch Apache日志** ###

这个示例创建了一个Logstash管道需要Apache web日志作为输入,分析这些日志创建特定的命名字段的日志,并将解析后的数据写入一个Elasticsearch集群。

你可以下载这个例子中使用的样本数据集。解压缩这个文件。   
[点击下载官方样例][Link 1]

**配置Logstash文件输入**

开始使用你的Logstash管道，配置Logstash实例从你的输入管道读取一个文件。   
编辑**first-pipeline.conf**文件，添加下面的内容：

input {
        file {
            path => "/path/to/logstash-tutorial.log"
            start_position => beginning //1
            ignore_older => 0 //2
        }
    }

1.  文件输入插件的默认行为是监控新信息的文件,在某种程度上类似于UNIX tail - f命令。为了改变这种默认行为和处理整个文件,我们需要指定Logstash开始处理文件的起始位置。
2.  文件输入插件默认行为忽略最后修改时间大于86400s的文件。因此，为了改变这种默认行为并且处理个别的文件（时间可以大于一天），我们需要指定不要忽略这些旧文件。
    
    在你的文件系统中，用你的实际路径**logstash-tutorial.log**替换**/path/to/**。

**用Grok过滤器插件解析Web日志**

Grok过滤器插件是Logstash默认可以使用的插件之一。关于如何管理使用Logstash插件，可以查看插件管理章节。

因为Grok过滤器插件是根据传入数据然后寻找的模式，因此你需要配置如何找到你需要的数据部分。一个在web 服务器当中的日志样例看起来就像下面这样：

83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
    HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
    Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

IP地址在第一行很容易就能看到，时间戳在一个方括号中，在本教程中，使用**%\{COMBINEDAPACHELOG\}** 的Grok模式，从Apache Log当中使用下面的字段来标识COMBINEDAPACHELOG：

<table> 
 <thead> 
  <tr> 
   <th>Information</th> 
   <th>Field Name</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>IP Address</td> 
   <td>clientip</td> 
  </tr> 
  <tr> 
   <td>User ID</td> 
   <td>ident</td> 
  </tr> 
  <tr> 
   <td>User Authentication</td> 
   <td>auth</td> 
  </tr> 
  <tr> 
   <td>timestamp</td> 
   <td>timestamp</td> 
  </tr> 
  <tr> 
   <td>HTTP Verb</td> 
   <td>verb</td> 
  </tr> 
  <tr> 
   <td>Request body</td> 
   <td>request</td> 
  </tr> 
  <tr> 
   <td>HTTP Version</td> 
   <td>httpversion</td> 
  </tr> 
  <tr> 
   <td>HTTP Status Code</td> 
   <td>response</td> 
  </tr> 
  <tr> 
   <td>Bytes served</td> 
   <td>bytes</td> 
  </tr> 
  <tr> 
   <td>Referrer URL</td> 
   <td>referrer</td> 
  </tr> 
  <tr> 
   <td>User agent</td> 
   <td>agent</td> 
  </tr> 
 </tbody> 
</table>

编辑**first-pipeline.conf** 文件，添加下面的内容：

filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
    }

处理后，样例线有下面的JSON表示：

{
    "clientip" : "83.149.9.216",
    "ident" : ,
    "auth" : ,
    "timestamp" : "04/Jan/2015:05:13:42 +0000",
    "verb" : "GET",
    "request" : "/presentations/logstash-monitorama-2013/images/kibana-search.png",
    "httpversion" : "HTTP/1.1",
    "response" : "200",
    "bytes" : "203023",
    "referrer" : "http://semicomplete.com/presentations/logstash-monitorama-2013/",
    "agent" : "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"
    }

**索引解析数据到Elasticsearch**

现在web日志分解成特定的字段,Logstash管道可以索引的数据变成Elasticsearch集群。编辑**first-pipeline.conf**文件，在**input**后面添加下面的内容：

output {
        elasticsearch {
        }
    }

在这个配置中,Elasticsearch Logstash使用http协议连接。上面的示例假定Logstash和Elasticsearch运行在相同的实例。您可以指定一个远程Elasticsearch实例使用主机配置host = >“es-machine:9092”。

**优化你的数据和Geoip过滤器插件**

除了把解析的日志数据能够更好的搜索，过滤器插件可以从已存在的数据当中推出一些补充信息。作为一个例子,geoip插件查找IP地址,获取地理位置信息的地址,并添加位置信息的日志。配置Logstash实例使用geoip过滤器插件通过在first-pipeline.conf 文件中添加下面的内容到**filter**部分:

geoip {
        source => "clientip"
    }

geoip插件配置需要的数据已经定义为单独的字段。需要确保**geoip**部分配置在**grok**部分的后面。

指定一个包含IP地址的文件来查找，在本教程中，名字是：clientip。

**测试初始管道**

到现在，你的第一条管道。配置文件输入,过滤器,和输出部分正确配置,和看起来像这样:

input {
        file {
            path => "/Users/palecur/logstash-1.5.2/logstash-tutorial-dataset"
            start_position => beginning
        }
    }
    filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
        geoip {
            source => "clientip"
        }
    }
    output {
        elasticsearch {}
        stdout {}
    }

来验证你的配置，运行一下命令：

bin/logstash -f first-pipeline.conf --configtest

—configtest选项解析您的配置文件和报告任何错误。当配置文件通过配置测试,开始Logstash使用下面的命令:

bin/logstash -f first-pipeline.conf

试着测试查询Elasticsearch基于Grok过滤器插件创建的字段:

curl -XGET 'localhost:9200/logstash-$DATE/_search?q=response=200'

取代$Date用当前日期,YYYY.MM.DD格式。   
因为我们的样品只有一个200 HTTP响应，我们得到很多返回：

{"took":2,
    "timed_out":false,
    "_shards":{"total":5,
      "successful":5,
      "failed":0},
    "hits":{"total":1,
      "max_score":1.5351382,
      "hits":[{"_index":"logstash-2015.07.30",
        "_type":"logs",
        "_id":"AU7gqOky1um3U6ZomFaF",
        "_score":1.5351382,
        "_source":{"message":"83.149.9.216 - - [04/Jan/2015:05:13:45 +0000] \"GET /presentations/logstash-monitorama-2013/images/frontend-response-codes.png HTTP/1.1\" 200 52878 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
          "@version":"1",
          "@timestamp":"2015-07-30T20:30:41.265Z",
          "host":"localhost",
          "path":"/path/to/logstash-tutorial-dataset",
          "clientip":"83.149.9.216",
          "ident":"-",
          "auth":"-",
          "timestamp":"04/Jan/2015:05:13:45 +0000",
          "verb":"GET",
          "request":"/presentations/logstash-monitorama-2013/images/frontend-response-codes.png",
          "httpversion":"1.1",
          "response":"200",
          "bytes":"52878",
          "referrer":"\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
          "agent":"\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\""
          }
        }]
      }
    }

尝试另一个寻找：从IP地址查找地理位置信息：

curl -XGET 'localhost:9200/logstash-$DATE/_search?q=geoip.city_name=Buffalo'

取代$Date用当前日期,YYYY.MM.DD格式。

只有一个日志来自Buffalo，所以我们只得到了一个响应：

{"took":3,
    "timed_out":false,
    "_shards":{
      "total":5,
      "successful":5,
      "failed":0},
    "hits":{"total":1,
      "max_score":1.03399,
      "hits":[{"_index":"logstash-2015.07.31",
        "_type":"logs",
        "_id":"AU7mK3CVSiMeBsJ0b_EP",
        "_score":1.03399,
        "_source":{
          "message":"108.174.55.234 - - [04/Jan/2015:05:27:45 +0000] \"GET /?flav=rss20 HTTP/1.1\" 200 29941 \"-\" \"-\"",
          "@version":"1",
          "@timestamp":"2015-07-31T22:11:22.347Z",
          "host":"localhost",
          "path":"/path/to/logstash-tutorial-dataset",
          "clientip":"108.174.55.234",
          "ident":"-",
          "auth":"-",
          "timestamp":"04/Jan/2015:05:27:45 +0000",
          "verb":"GET",
          "request":"/?flav=rss20",
          "httpversion":"1.1",
          "response":"200",
          "bytes":"29941",
          "referrer":"\"-\"",
          "agent":"\"-\"",
          "geoip":{
            "ip":"108.174.55.234",
            "country_code2":"US",
            "country_code3":"USA",
            "country_name":"United States",
            "continent_code":"NA",
            "region_name":"NY",
            "city_name":"Buffalo",
            "postal_code":"14221",
            "latitude":42.9864,
            "longitude":-78.7279,
            "dma_code":514,
            "area_code":716,
            "timezone":"America/New_York",
            "real_region_name":"New York",
            "location":[-78.7279,42.9864]
          }
        }
      }]
     }
    }

### **多个输入和输出插件** ###

你需要管理的信息往往来自几个不同来源,并且用例的数据可能需要多个目的地。你的Logstash管道可以使用多个输入和输出插件来处理这些需求。

这个示例创建一个Logstash管道,从Twitter和Filebeat客户机输入,然后将信息发送给一个Elasticsearch集群以及直接写信息到文件中。

**从Twitter消息来源读取数据**

添加一个Twitter消息来源，你需要下面几条信息：

*  consumer\_key：它唯一的标识了你的Twitter app。
 *  consumer\_secret：它是你的Twitter的密码。
 *  keywords：一个或多个关键字在你的输入来源中。
 *  oauth\_token：标识了使用这个App的Twitter账户。
 *  oauth\_token\_secret：作为Twitter账户的密码。
 *  （这里其实笔者并不太了解，查询之后oauth这个是账号和密码，然后consumer\_key是从Twitter申请的开发号。两个是不一样的账号）
    
    在**first-pipeline.conf**文件中的input部分，添加下面的信息：

twitter {
        consumer_key =>
        consumer_secret =>
        keywords =>
        oauth_token =>
        oauth_token_secret =>
    }

**Fileabeat客户端**

Filebeat客户端是一个轻量级的资源友好的从服务器搜日志并且发送到Logstash实例进行处理的工具。Filebeat客户机使用的安全比协议与Logstash实例进行数据交换。Lumberjack协议是一个可靠和低延迟的协议。Filebeat使用计算机的主机资源托管源数据,并且Beats输入插件最小化的Logstash实例资源。

> 在一个典型的用例,Filebeat从机器运行在不同的机器上运行Logstash实例。对于本教程,Logstash和Filebeat是运行在同一台机器上。

默认Logstash配置包括资源友好的Beats输入插件，在数据源上安装Filebeat,从Filebeat产品页面下载适当的包。、

创建一个配置文件Filebeat类似于下面的例子:

filebeat:
      prospectors:
        -
          paths:
            - "/path/to/sample-log" //Filebeat执行的路径
          fields:
            type: syslog
    output:
      logstash:
        hosts: ["localhost:5043"]
      tls:
        certificate: /path/to/ssl-certificate.crt //Logstash实例的SSL证书路径
        certificate_key: /path/to/ssl-certificate.key
        certificate_authorities: /path/to/ssl-certificate.crt
        timeout: 15

保存配置文件 **filebeat.yml**。

配置你的Logstash实例来使用Filebeat输入插件：在**First-pipeline.conf**文件中的**input**部分添加下面的内容：

beats {
        port => "5043"
        ssl => true
        ssl_certificate => "/path/to/ssl-cert" //Logstash使用的验证Filebeat的SSL证书
        ssl_key => "/path/to/ssl-key" //SSL证书的路径
    }

**将Logstash数据写入文件**

你可以直接配置Logstash管道通过输出插件来将数据写入文件。

配置你的Logstash实例，使用**file**输出插件，在**first-pipeline.conf**文件的output部分添加下面内容：

file {
        path => /path/to/target/file
    }

**写入多个Elasticsearch节点**

写入多个Elasticsearch节点减轻给定Elasticsearch节点上的资源需求,并且当特定节点不可用时，提供冗余节点进入集群节点。

配置你的Logstash实例来写入多个Elasticsearch节点，编辑**first-pipeline.conf**文件，在output部分添加：

output {
        elasticsearch {
            hosts => ["IP Address 1:port1", "IP Address 2:port2", "IP Address 3"]
        }
    }

使用三个非主节点的IP地址在Elasticsearch集群主机线。使用三个非主节点的IP地址在Elasticsearch集群主机线。当主机参数列表多个IP地址,Logstash负载平衡请求在地址的列表。还要注意,Elasticsearch的默认端口是9200,可以省略在上面的配置。

**测试管道**

到现在，你的配置文件**first-pipeline.conf**文件看起来是下面的样子：

input {
        twitter {
            consumer_key =>
            consumer_secret =>
            keywords =>
            oauth_token =>
            oauth_token_secret =>
        }
        beats {
            port => "5043"
            ssl => true
            ssl_certificate => "/path/to/ssl-cert"
            ssl_key => "/path/to/ssl-key"
        }
    }
    output {
        elasticsearch {
            hosts => ["IP Address 1:port1", "IP Address 2:port2", "IP Address 3"]
        }
        file {
            path => /path/to/target/file
        }
    }

Logstash使用Twitter的数据配置,从Filebeat接收数据,索引此信息以Elasticsearch集群以及三个节点写入一个文件。

在数据源上,Filebeat运行以下命令:

sudo ./filebeat -e -c filebeat.yml -d "publish"

Filebeat将尝试连接在端口5403上。直到Logstash始于一个活跃的beats插件，在这个端口将不会有任何答案，所以你看到任何关于这个端口的错误消息都是正常的。

来验证您的配置,运行以下命令:

bin/logstash -f first-pipeline.conf --configtest

—configtest选项解析您的配置文件并且报告任何的错误。当配置文件通过配置测试,开始Logstash使用下面的命令:

bin/logstash -f first-pipeline.conf

使用grep工具搜索目标文件存在,验证信息:

grep Mozilla /path/to/target/file

运行一个Elasticsearch语句，来寻找Elasticsearch集群中相同的信息：

curl -XGET 'localhost:9200/logstash-2015.07.30/_search?q=agent=Mozilla'

[原文引用][Link 2]

[Logstash pipeline]: /images/20220518/4a2be6e542f148dfbfcbba288fc4d666.png
[Link 1]: https://download.elastic.co/demos/logstash/gettingstarted/logstash-tutorial.log.gz
[Link 2]: https://blog.csdn.net/DiamondXiao/article/details/52056731