ELK_logstash

logstash.conf

input–>filter–>output

input:从哪里收集日志

filter:对日志进行过滤

output:输出哪里

input {
  beats {
    port => 5044
    host => "0.0.0.0"
  }
}
}
output {
  elasticsearch {   # 将日志发送给elasticsearch
    hosts => ["elasticsearch:9200"]   
    manage_template => false   # 数据模板
    index => "leqian-%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
  }   # 索引
  # stdout { codec => rubydebug }  # 假如有问题，可以打开此行进行调试
}

输 入插件(input)

文件输入

input {
file {
path => "/path/to/logsfiles"
}
}

标准输入

input {
stdin {
}
}

filebeat

input {
  beats {
    port => 5044
    host => "0.0.0.0"
  }
}

输出(output)

数据输出到屏幕终端

output {

stdout{codec => rubydebug}}

数据存储到 elasticsearch 集群中

output {
  elasticsearch { 
    hosts => ["elasticsearch:9200"]  # elasticsearch集群主机
    manage_template => false  
    index => "leqian-%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"   # 创建的索引
  }
  # stdout { codec => rubydebug }  # 假如有问题，可以打开此行进行调试
}

属性的数据类型

数组

python => ["value1","value2"]

布尔值

true 和 false

periodic_flush => false

输出的时候设置，把输出数据编码成JSON 格式

codec => "json"

以 "key" => "value"的形式表示，多个 用空格分开
ipHash

match => {

"key1" => "value1"  "key1" => "value2"
}

字符串

value => "Welcome   to  ELK"

字段引用

logstash 的数组也 支持倒序下标

"the longitude is %{[geoip][location][0]}"

Logstash 条件语句句

if (val == 0) {

#  一些处理理语句句

}else if(val == "str") {

#  一些处理理语句句

}else {

# 其他的处理理语句句

}
>相等运算符: ==, !=, <, >, <=, >=

正则表达式: =~, !~
包含: in, not in

逻辑运算符: and(与), or(或), nand( 非与), xor( 非或)

if "_grokparsefailure" not in [tags] {

} else if [status] !~ /^2\d\d/ or ( [url] == "/noc.gif" nand [g eoip][city] != "beijing" )
{

} else {

}

grok过滤器 插件

filter

filter {
    grok {  # 
        match => { "message" => "%{COMBINEDAPACHELOG}"}   #
    }
    geoip { # 根据ip获取地理信息
        source => "clientip"  
    }
    date{}
}

假如你在这之前已经运 行行了了 logstash 和 filebeat 。要想 生效现在的过滤配置，您需要强制Filebeat从头开始读取 日志 文件。
不必重新启动Logstash来接收更更改，但是需要删除 filebeat 下的注册表 文件registry，此 文件 一般在安装家目录下的 data 目录下。
由于Filebeat存储在注册表中收集的每个 文件的状态，因此删除注册表 文件会强制Filebeat读取从头开始捕获的所有 文件。
接下来，使 用以下命令重新启动Filebeat即可

索引名称中使 用的 日期是基于UTC， 而不不是Logstash运 行行的时区

列出集群中的所有节点

curl '172.16.153.129:9200/_cat/nodes?v'

列出集群健康状态

curl '172.16.153.129:9200/_cluster/health?pretty=true'

logstash输出信息

Logstash startup completed

Hello PackPUb   # 输 入的信息

{   # 以下为输出信息

"message" => "Hello PackPUb",

"@version" => "1",

"@timestamp" => "2017-10-22T19:57:50.698Z", # 事件被索引的时间

"host" => "0.0.0.0"

}

message => 完整的输 入信息或者事件。
@timestamp => 事件被索引的时间。
如果使 用了了 日期过滤插件，也可以是 message 中的某个指定事件时间的字段
host => 表示事件发 生的主机

生成测试数据

test-dot.conf

input{
  generator {
        count => 10000000
        message => '{"key1":"value1","key2":[1,2],"key3":{"subkey1":"subvalue1"}}'
        codec => json
    }
}
output{
    stdout {
        codec => dots
    }
}

root用户下载
yum -y install epel-*
yum -y install pv

利用pv测试数据

启动   logstash -f test-dot.conf | pv -abt >/dev/null

这里单位是 B/s，但是因为一个 event 就输出一个 .，也就是 1B。所以 12.5kiB/s 就相当于是 12.5k event/s。

image.png

grok

测试grok
github过滤正则文档

Jan  1 06:25:43 mailserver14 postfix/cleanup[21403]: BEF25A72965: message-id=<20130101142543.5828399CCAF@mailserver14.example.com>

filter{
grok {
patterns_dir => ["./patterns"]
match => { 
"message" => "%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{GREEDYDATA:syslog_message}" 
}

}
}

timestamp: Jan 1 06:25:43
logsource: mailserver14
program: postfix/cleanup
pid: 21403
queue_id: BEF25A72965
syslog_message: message-id=20130101142543.5828399CCAF@mailserver14.example.com

?

贪婪模式与非贪婪模式
对于字符串“oooo”，“o+?”将匹配单个“o”，而“o+”将匹配所有“o”。

源字符串：<div>test1</div>bb<div>test2</div>cc

正则表达式一：<div>.*</div>

匹配结果一：<div>test1</div>bb<div>test2</div>

正则表达式二：<div>.*?</div>

匹配结果二：<div>test1</div>（这里指的是一次匹配结果，所以没包括<div>test2</div>）

date
官方文档

input{ stdin{} }
filter{
    grok {
        match => {
                "message" => "(?<time>.*)"
        }
    }
    date {
        match => ["time","MMM dd yyyy HH:mm:ss"]
     }

}

output{ stdout{ codec => rubydebug }  }

将时间戳修改为日志内的时间,进行查询过滤

对于非格式语法，您需要在值周围放置单引号字符。例如，如果您要解析ISO8601时间，“ 2015-01-01T01：12：23”，则小“ T”不是有效的时间格式，而您想说的是“字面意义上的T”，则格式为这个：“ yyyy-MM-dd'T'HH：mm：ss”

时间字段

yyyy
全年数字。范例：2015。
y
两位数的年份。示例：15==>2015年。
M
最小位数月份。示例：1一月和12十二月。
MM
两位数的月份。如果需要，请补零。示例：01一月和12十二月
MMM
缩写的月份文本。例如：一Jan月份。注意：使用的语言取决于您的语言环境。请参阅有关locale如何更改语言的设置。
MMMM
整月文字，例如：January。注意：使用的语言取决于您的语言环境。
d
最小位数的一天。示例：1每月的1号。
dd
两位数的日期，如有必要，请补零。示例：01每月的1号。
Z
时区偏移的结构为HHmm（与Zulu / UTC的小时和分钟偏移）。范例：-0700。
ZZ
时区偏移量的结构为HH：mm（小时和分钟偏移量之间的冒号）。范例：-07:00。
E，EE，EEE
星期几的缩写。例如： Mon，Tue，Wed，Thu，Fri，Sat，Sun。注意：此操作的实际语言将取决于您的区域设置。