Elastic Stack —- Logstash

layout: post
title: Elastic Stack —— Logstash
date: 2018-04-08
tags: [“ELK”,”软件服务”]

一、Logstash简介

logstash是一个ETL工具，可以完成数据的抽取和装载的操作。讲数据抽取之后将数据文档化，并将其输出到指定的目标。

二、Logstash配置

1、Logstash的安装很简单，从官网下载最新的tar包，开箱即用。

2、简单配置Logstash

Logstash主要分三个配置段：input、stdout、filter，分别代表输入、输出、条件过滤，而每个代码块是支持多种插件的。

Logstash的配置格式框架

input {
 ...
}
 
filter{
 ...
}
 
output {
 ...
}

Logstash的简单配置：

vim /etc/logstash.conf
input {
    stdin {}     \\标准输入
}
 
output {
    stdout {
        codec => rubydebug  \\解码器，格式有"json" 和 "rubydebug"（用作调试） 
    }
}

启动logstash

./logstash -f /etc/logstash.conf
 
输入"aa"
 
{
    "@timestamp" => 2018-04-10T12:25:04.575Z,
       "message" => "aa",
      "@version" => "1",
          "host" => "test.middle.01"
}

3、基于”file”的输入配置

input {
    file {            #使用file插件，参数详见官网
        path => ["/var/log/httpd/access_log"]  
        start_position => "beginning"
    }
}
 
filter {              #filter用作中间层处理数据
    grok {            #该插件用作文本做模式匹配以后，添加键
        match => {
            "message" => "%{COMBINEDAPACHELOG}"
        }
        remove_field: "message"   #用于删除message
    }
}
output {
    stdout {
        codec => rubydebug
    }
}

grok切割message的匹配模式格式参见Github

4、时间戳的配置

标准输出”@timestamp”在生产环境中有时我们并不会使用logstash的时间戳作为时间标识，我们需要替换成日志产生时的时间标识，这时我们需要使用生产的时间戳来替换logstash的时间戳

filter {
        grok {
                match => {
                        "message" => "%{HTTPD_COMBINEDLOG}"
                }
                remove_field => "message"
        }
        date {
                match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
                remove_field => "timestamp"
        }
 
}

使用date标签的match使用日志的时间替换@timestamp的的时间，移除remove_field,格式匹配见官网

5、Filter的mutate、Geoip插件

filter {
        grok {
                match => {
                        "message" => "%{HTTPD_COMBINEDLOG}"
                }
        }
        date {
                match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
        }
        mutate {
                rename => {
                        "agent" => "user_agent"
                }
        }
        geoip {
                source => "clientip"
                target => "geoip"
                database => "/etc/logstash/maxmind/GeoLite2-City.mmdb"
        }
}

使用mutate的”rename”修改字段的名称
geoip的功能：

source 从分词中去除一个字段
target 指定geoip
database 指定geoip数据库

geoip库：ES支持geoip2，地址

注意：
1、输出的日志文件名必须以”logstash-“开头，方可将geoip.location的type自动设定为”geo_point”；
2、target => “geoip”

6、Redis插件（input）

input {
    redis {
        batch_count => 1
        data_type => "list"
        key => "logstash-list-%{+YYY.MM.dd}"
        host => ["192.168.0.2"]
        port => 6379
        threads => 5
    }
}

注意键值的类型，参见官网

7、elasticsearch组件（output）

output {
    elasticsearch {
        hosts => ["http://node1:9200/","http://node2:9200/","http://node3:9200/"]
        user => "ec18487808b6908009d3"
        password => "efcec6a1e0"
        index => "logstash-%{+YYYY.MM.dd}"
        document_type => "apache_logs"
    }
}

8、beats的输入

input {
    beats {
        port => 5044
    }
}

三、Beats的配置

1、filebeat（下载地址）

配置文件：filebeat.yml

开启tags，用于logstash辨识日志类型

开启document_type，用于logstash向elasticsearch输出时辨识索引类别

type配置log（log、stdin）

output可以使用三种：

1）elasticsearch
2）logstash
3）redis
4）其他输出类型参见filebeat.reference.yml文件

2、logstash在使用beats插件，在filter进行切割时通过以下配置来区分日志类型

if "nginx-accesslog" in [tags]

if "sys-messages" in [tags]

语法：

filter {
 
    if [type] == 'tomcat-accesslog' {
        grok {}
    }
 
    if [type] == 'httpd-accesslog' {
        grok {}
    }
 
}
 
if EXPRESSION {
...
} else if EXPRESSION {
...
} else {
...
}

3、output配置

使用``index => ``"logstash-%{type}-%{+YYYY.MM.dd}"来区分不同的索引