Logstash工作原理

Logstash事件处理有三个阶段：inputs → filters → outputs。是一个接收，处理，转发日志的工具。支持系统日志，webserver日志，错误日志，应用日志，总之包括所有可以抛出来的日志类型。
image.png

1.安装logstash

  1. [root@master ~]# wget https://artifacts.elastic.co/downloads/logstash/logstash-7.2.0.tar.gz
  2. [root@master ~]# tar zxvf logstash-7.2.0.tar.gz
  3. [root@master ~]# mv logstash-7.2.0 /usr/local/logstash
  4. [root@master ~]# vim /usr/local/logstash/config/test.conf
  5. input {
  6.         beats {
  7.                 port => 5044
  8.         }
  9. }
  10. ###########################################################################################################################
  12. filter {
  13.         if [fields][appname] =~ ".*access-log" {
  14.                 grok {
  15.                         match => {  "message" => ["(%{TIMESTAMP_ISO8601:Request_time}|-)\;(%{IPORHOST:Client_ip}|-)\;(%{DATA:User_name}|-)\;(%{WORD:Request_id}|-)\;(%{PATH:Request_uri}|-)\;(%{INT:Response_time}|-)\;(%{INT:Status_code}|-)\;(%{NOTSPACE:Status_message}|-)"] }
  16.                 }
  17.                 geoip {
  18.                         source => "Client_ip"
  19.                 }
  20.                 mutate {
  21.                         convert => [ "[geoip][coordinates]", "float"]
  22.                 }
  23.         }
  27.         if [fields][appname] =~ ".*info-log" {
  28.                 grok {
  29.                         match => {  "message" => ["(%{TIMESTAMP_ISO8601:Request_time})  (%{WORD:INFO}) (%{GREEDYDATA:Message})"] }
  30.                 }
  31.                 if [message] == ";" {
  32.                 drop {}
  33.                 }
  34.         }
  37.         if [fields][appname] =~ ".*error-log" {
  38.                 grok {
  39.                         match => {  "message" => ["(%{TIMESTAMP_ISO8601:Request_time}) (%{WORD:INFO}) (%{INT:Code}) --- \[(%{GREEDYDATA:Noname})\] (%{GREEDYDATA:Method}) \:(%{GREEDYDATA:Message})"] }
  40.                 }
  41.         }
  44.         if [fields][appname] =~ ".*bmw-info-log" {
  45.                 grok {
  46.                         match => {  "message" => ["(\[%{DATA:Request_time}\](%{GREEDYDATA:Message}))"] }
  47.                 }
  48.         }
  50. #       if [fields][appname] =~ ".*billing-log" {
  51. #               grok {
  52. #                       match => {  "message" => ["(%{TIMESTAMP_ISO8601:Request_time} \[ %{DATA:Schedul} \] - \[ %{DATA:Level} \] \[ %{DATA:Localtion} \] - %{GREEDYDATA:Message})"] }
  53. #               }
  54. #        }
  55.         date {
  56.                 match => [ "Request_time" , "MMM  d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601" ]
  57.         }
  58. }
  59. ##########################################################################################################################
  61. output {
  62.         if [fields][appname] == "online-bmw-info-log"{
  63.                 elasticsearch {
  64.                         hosts => ["ESIP:9200"]
  65.                         user => "elastic"
  66.                         password => "situ1234"
  67.                         index => "bmw-info-log-%{+YYYY.MM.dd}"
  68.                 }
  69.         }
  70.         if [fields][appname] == "staging-bmw-access-log" {
  71.                 elasticsearch {
  72.                         hosts => ["ESIP:9200"]
  73.                         user => "elastic"
  74.                         password => "situ1234"
  75.                         index => "bmw-access-log-%{+YYYY.MM.dd}"
  76.                 }
  77.         }
  78.         if [fields][appname] == "staging-bmw-beta-access-log" {
  79.                 elasticsearch {
  80.                         hosts => ["ESIP:9200"]
  81.                         user => "elastic"
  82.                         password => "situ1234"
  83.                         index => "bmw-beta-access-log-%{+YYYY.MM.dd}"
  84.                 }
  85.         }
  86. }
  87. #########################################################################################################################

2.准备启动服务

  1. # 以下为测试方法
  2. [root@master ~]# bin/logstash -e 'input { stdin { } } output { stdout {} }'
  3. # 手动输入hello world它也会输出hello world

启动logstash

  1. root@master ~]# nohup bin/logstash -f /usr/local/logstash/config/test.conf &  
  2. [root@master ~]# netstat -ntplu
  3. tcp6       0      0 :::5044                 :::*                    LISTEN      14069/java
  4. tcp6       0      0 127.0.0.1:9600          :::*                    LISTEN      14069/java
  6. # 通过下面操作可以查看logstash和elasticsearch是否建立连接，刷新会看到数据变化。
  7. [root@master ~]# curl http://192.168.3.128:9200/_cat/indices
  8. yellow open filebeat-2019.07.16 PCHNvs83TvOuSd1iDU4S-A 5 1 4131 0 841.7kb 841.7kb
  9. yellow open .kibana             qXyuI2ubT3-bBwyknxoyhA 1 1    2 0    21kb    21kb