记录下logstash的配置加解释

logstash.conf(全,加注释)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
input {
  beats {
    port => 5044
    #ssl => true
    #ssl_certificate => "/etc/logstash/logstash.crt"
    #ssl_key => "/etc/logstash/logstash.key"
# 1. SSL 详情可参考 
  }
}
# filter 模块主要是数据预处理，提取一些信息，方便 elasticsearch 好归类存储。
# 2. grok 正则捕获 
# 3. grok 插件语法介绍 
# 4. logstash 配置语法 
# 5. grok 内置 pattern 
filter {
    grok {  
      match => {"message" => "%{EXIM_DATE:timestamp}\|%{LOGLEVEL:log_level}\|%{INT:pid}\|%{GREEDYDATA}"}
# message 字段是 log 的内容，例如 2018-12-11 23:46:47.051|DEBUG|3491|helper.py:85|helper._save_to_cache|shop_session
# 在这里我们提取出了 timestamp log_level pid，grok 有内置定义好的 patterns: EXIM_DATE, EXIM_DATE, INT
# GREEDYDATA 贪婪数据，代表任意字符都可以匹配 
    }
# 我们在 filebeat 里面添加了这个字段[fields][function] 的话，那就会执行对应的 match 规则去匹配 path
# source 字段就是 log 的来源路径，例如 /var/log/nginx/feiyang233.club.access.log
# match 后我们就可以得到 path=feiyang233.club.access
    if [fields][function]=="nginx" {
        grok {         
        match => {"source" => "/var/log/nginx/%{GREEDYDATA:path}.log%{GREEDYDATA}"}  
            }
        } 
# 例如 ims 日志来源是 /var/log/ims_logic/debug.log
# match 后我们就可以得到 path=ims_logic
    else if [fields][function]=="ims" {
        grok {
        match => {"source" => "/var/log/%{GREEDYDATA:path}/%{GREEDYDATA}"}
            }
        }  

    else {
        grok {
        match => {"source" => "/var/log/app/%{GREEDYDATA:path}/%{GREEDYDATA}"}
            }         
        }
# filebeat 有定义 [fields][function] 时，我们就添加上这个字段，例如 QA
    if [fields][function] {
          mutate {
              add_field => {
                  "function" => "%{[fields][function]}"
                }
            }
        } 
# 因为线上的机器更多，线上的我默认不在 filebeat 添加 function，所以 else 我就添加上 live  
    else {
          mutate {
              add_field => {
                  "function" => "live"
                }
            }
        }
# 在之前 filter message 时，我们得到了 timestamp，这里我们修改一下格式，添加上时区。
    date {
      match => ["timestamp" , "yyyy-MM-dd HH:mm:ss Z"]
      target => "@timestamp"
      timezone => "Asia/Singapore"
    }
# 将之前获得的 path 替换其中的 / 替换为 - , 因为 elasticsearch index name 有要求
# 例如 feiyang/test  feiyang_test 
    mutate {
        gsub => ["path","/","-"]
        # 重命名某个字段的名称
        rename => {
          "old_field" => "new_field"
        }
        # 添加新的字段
        add_field => {"host_ip" => "%{[fields][host]}"}
        # remove_field 去掉一些多余的字段
        remove_field => ["tags","@version","offset","beat","fields","exim_year","exim_month","exim_day","exim_time","timestamp"]
    }

}
# 单节点 output 就在本机，也不需要 SSL, 但 index 的命名规则还是需要非常的注意
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "sg-%{function}-%{path}-%{+xxxx.ww}"
# sg-nginx-feiyang233.club.access-2019.13  ww 代表周数
  }
}