Logstash 基础学习

Logstash简介

• 具备实时数据传输能力的管道

  支持多种数据源输入；

  支持多种guolvq；

  支持多种数据输出目的地；

  -在ELK中作为日志收集器

Logstash安装

首先对安装包进行解压，安装到指定的路径：

tar -zxvf logstash-6.2.2.tar.gz -C /opt/
           

安装成功后，重新命名，方便调用，进入到logstash目录下，进行测试，看是否安装成功：

测试程序，多等待一会，最后出现提示输入，可以输入hello验证一下，出现下图示例，表示成功

Logstash工作

Logstash对任何时间处理分为三个阶段：

输入：input，过滤器：filter，输出：output

语法结构：

过滤器不是必须存在的

# 输入
input {
  ...   
}
# 过滤器
filter {
  ...
}
# 输出
output {
  ...
}
           

数据类型：

array：数组。示例：path => [ “/var/log/messages”, “/var/log/*.log” ]

boolean：布尔值。

codec：表示编码名称。codec => “json”

hash：表示键值对，多个键值对用空格分隔。示例：

match => {

“field1” => “value1”

“field2” => “value2” … }

string :字符串

path：一个代表有效的操作系统路径。my_path => “/tmp/logstash”

stdin:

output中表示的是输出格式

input{
    stdin{}

}
}

output{
        stdout{
                codec => "rubydebug"
        }
        elasticsearch{
        hosts => [ "cent02:9200" ]
}
}
           

在控制台输入得到的结果：

./bin/logstash -f tmp/test.conf

           

file:

对文件的读取：

实现对文件的读取，可以先导入一个json文件，文件的路径path是必须有的，

sincedb_path表示日志文件

input{
        file{
         path => "/opt/logstash/tmp/stu.json"
        start_position => "beginning"
        sincedb_path => "dev/null"
        codec => "json"
}
}

output{
        stdout{
                codec => "rubydebug"
        }
        elasticsearch{
        hosts => [ "cent02:9200" ]
}
}

           

在这里可以直接实现对文件的读取

./bin/logstash -f tmp/test.conf

           

过滤器和输出

grok:使用正则表达式解析任意文本和结构

number是内置匹配规则，duration是字段名，

mutate过滤器可以重命名、删除、替换和修改字段，

if [duration] <100是进行输出的限制

input{
        stdin{}
}

filter{
        grok{
                match=>{"message"=>"%{NUMBER:duration} %{IP:client}"}
}
        mutate{
                convert => ["duration","integer"]
}
}
output{
        if [duration] <100{
        stdout{
                codec => "rubydebug"
        }
        }
}

           

./bin/logstash -f tmp/grok_test.conf

           

当我们输入100 10.10.10.1时，100不满足output的条件，无法输出，而90 10.20.30.10则可以正常输出