13

Elastic Stack-03-Logstash-7.10.0-入门

 3 years ago
source link: https://blog.csdn.net/qq_22648091/article/details/111827543
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
neoserver,ios ssh client

参考官方网站 Logstash

二、测试安装

首先，让我们通过运行最基本的Logstash管道来测试Logstash安装。

Logstash管道具有两个必需元素input和output，以及一个可选元素filter。输入插件使用来自源的数据，过滤器插件根据您的指定修改数据，输出插件将数据写入目标。

在这里插入图片描述

测试Logstash安装，请运行最基本的Logstash管道。
例如,可以执行如下命令使 Logstash 程序运行于前台

进入 Logstash 的安装主目录下执行：

bin/logstash -e ''

-e 选项用于设置 Logstash 处理数据的输入和输出

-e '' 这里使用了空字符串等同于使用 -e input { stdin { type => stdin } } output { stdout { codec => rubydebug } }

input { stdin { type => stdin } } 表示 Logstash 需要处理的数据来源来自于标准输入设备（键盘）

output { stdout { codec => rubydebug } } 表示 Logstash 把处理好的数据输出到标准输出设备（屏幕，也就是终端）

稍等片刻，当看到屏幕上输出如下字样，即可尝试使用键盘输入 hello 字样

[2020-11-29T11:47:57,606][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

注意，你的时间肯定不和这里的一样，这里是我当时启动时候的时间。

输出 hello 机会立刻看到终端输出配格式化后的数据信息

在这里插入图片描述

  • message 字段对应的值是 Logstash 接收到的一行完整的数据
  • @version 是版本信息，可以用于建立索引使用（后面会讲）
  • @timestamp 处理次数据的时间戳，可以用于建立索引和搜索
  • type 就是之前 input 中设置的值，这个值可以任意修改，但是，type 是内置的变量，不能修改，用于建立索引和条件判断等
  • hosts 表示从那个主机过来的数据

修改 type 的值为 nginx 的示例


 bin/logstash -e "input { stdin { type => nginx } } output { stdout { codec => rubydebug } }"

在这里插入图片描述

三、配置输入和输出

在现实世界中，Logstash管道要复杂一些：它通常具有一个或多个输入，过滤器和输出插件。

在本部分中，您将创建一个Logstash管道，该管道使用标准输入来获取Apache Web日志作为输入，解析这些日志以从日志中创建特定的命名字段，然后将解析的数据输出到标准输出（屏幕上）。并且，我们这次无需在命令行上定义管道配置，而是在配置文件中定义管道。

创建任意一个文件，并写入如下内容，作为 Logstash 的管道配置文件

first-pipeline.conf

input { 
    stdin { } 
} 

output { 
    stdout {} 
}

运行如下命令可以测试配置文件

bin/logstash -f first-pipeline.conf --config.test_and_exit

-f 用于指定管道配置文件。

运行如下命令启动 Logstatsh

bin/logstash -f first-pipeline.conf --config.reload.automatic

--config.reload.automatic 会在你修改管道配置文件后自动加载，而不必重新启动 Logstash。

启动后复制如下内容到命令行中，并按下回车键

83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/imageskibana-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"

将会看到如下输出

{
      "@version" => "1",
       "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/imageskibana-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\"",
          "host" => "0.0.0.0",
    "@timestamp" => 2020-12-27T14:55:54.076Z
}

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

现在，您有了一个工作管道，该管道从Filebeat中读取日志行。
但是，您会注意到日志消息的格式不是理想的。
您想解析日志消息,以便能从日志中创建特定的命名字段。

为此，您应该使用grok 过滤器插件。

使用grok过滤器插件，您可以将非结构化日志数据解析为结构化和可查询的内容。

grok 过滤插件，会根据你感兴趣的内容分配字段名称，并把这些内容和对应的字段名称进行绑定。

grok 如何知道哪些内容是你感兴趣的呢？它是通过自己预定义的模式来识别感兴趣的字段的。这个可以通过给其配置不同的模式来实现。

这里使用的模式是 %{COMBINEDAPACHELOG}

{COMBINEDAPACHELOG 使用以下模式从Apache日志中构造行：

原信息对应新的字段名称IP 地址clientip用户 IDident用户认证信息auth时间戳timestampHTTP 请求方法verb请求的 URLrequestHTTP 版本httpversion响应码response响应体大小bytes跳转来源referrer客户端代理（浏览器）agent

关于 grok 更多的用法请参考 grok 参考文档

并且这里要想实现修改配置文件之后自动加载它，不能配置 input 为 stdin。
所以， 这里我们使用了 file

修改好的管道配置文件如下：

input {
  file {
    path => "/usr/local/logstash/access_log"
    start_position => "beginning"
  }
}

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

output {
  stdout {
     codec => rubydebug
  }
}

match => { "message" => "%{COMBINEDAPACHELOG}"} 的意思是：
当匹配到 “message” 字段时，用户模式 “COMBINEDAPACHELOG}” 进行字段映射。

配置完成后，再次进行验证

编辑示例日志文件，并写如如下内容

83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/imageskibana-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"

重新启动 logstash

 bin/logstash -f first-pipeline.conf --config.reload.automatic

下面是输出内容

{
          "ident" => "-",
          "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\"",
        "request" => "/presentations/logstash-monitorama-2013/imageskibana-search.png",
     "@timestamp" => 2020-12-28T13:55:03.369Z,
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
           "auth" => "-",
        "message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/imageskibana-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\"",
           "host" => "0.0.0.0",
       "clientip" => "83.149.9.216",
    "httpversion" => "1.1",
       "response" => "200",
      "timestamp" => "04/Jan/2015:05:13:42 +0000",
           "verb" => "GET",
          "bytes" => "203023",
       "@version" => "1"
}

你会发现原来的非结构化数据，变为结构化的数据了。

细心的你一定发现原来的 message 字段仍然存在，假如你不需要它，可以使用 grok 中提供的常用选项之一： remove_filed 来移除这个字段。
事实上 remove_field 可以移除任意的字段，它可以接收的值是一个数组。

修改后管道配置文件如下：

input {
  file {
    path => "/usr/local/logstash/access_log"
    start_position => "beginning"
  }
}

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
        remove_field => [ "message" ]
    }
}

output {
  stdout { codec => rubydebug }
}

再次测试，你会发现 message 不见了：

使用如下命令向示例日志文件中输入新的一行内容


echo '83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/imageskibana-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"' >> /usr/local/logstash/access_log

{
          "ident" => "-",
     "@timestamp" => 2020-12-28T15:02:47.938Z,
    "httpversion" => "1.1",
       "@version" => "1",
           "auth" => "-",
          "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\"",
        "request" => "/presentations/logstash-monitorama-2013/imageskibana-search.png",
      "timestamp" => "04/Jan/2015:05:13:42 +0000",
          "bytes" => "203023",
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
          "country_name" => "Russia",
           "postal_code" => "144700",
              "timezone" => "Europe/Moscow",
         "country_code2" => "RU"
    },
           "verb" => "GET",
       "clientip" => "83.149.9.216",
           "host" => "0.0.0.0",
       "response" => "200"
}

3 使用Geoip过滤器插件增强数据编辑

除解析日志数据以进行更好的搜索外，筛选器插件还可以从现有数据中获取补充信息。例如，geoip 插件可以通过查找到IP地址，并从自己自带的数据库中找到地址对应的地理位置信息，然后将该位置信息添加到日志中。

该geoip插件配置要求您指定包含IP地址来查找源字段的名称。在此示例中，该clientip字段包含IP地址。

    geoip {
        source => "clientip"
    }

由于过滤器是按顺序求值的，因此请确保该geoip部分位于grok配置文件的该部分之后，并且grok和geoip部分都嵌套在该filter部分中。

完成后的管道配置文件如下：


input {
  file {
    path => "/usr/local/logstash/access_log"
    start_position => "beginning"
  }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    stdout { codec => rubydebug }
}

再次输入之前的日志内容，就会看到如下输出

{
          "bytes" => "203023",
          "geoip" => {
             "city_name" => "Moscow",
        "continent_code" => "EU",
          "country_name" => "Russia",
           "region_name" => "Moscow",
              "timezone" => "Europe/Moscow",
             "longitude" => 37.6172,
           "region_code" => "MOW",
         "country_code3" => "RU",
                    "ip" => "83.149.9.216",
              "location" => {
            "lat" => 55.7527,
            "lon" => 37.6172
        },
              "latitude" => 55.7527,
           "postal_code" => "144700",
         "country_code2" => "RU"
    },
           "host" => "0.0.0.0",
     "@timestamp" => 2020-12-28T14:44:23.900Z,
           "verb" => "GET",
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
          "ident" => "-",
       "@version" => "1",
           "auth" => "-",
       "response" => "200",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2015:05:13:42 +0000",
        "request" => "/presentations/logstash-monitorama-2013/imageskibana-search.png",
          "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\"",
       "clientip" => "83.149.9.216"
}

详情请参考 grok 和 geoip


report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK