目录

http://blog.csdn.net/fenglailea/article/details/52486471

老版本

http://blog.csdn.net/fenglailea/article/details/52472419

JAVA版本说明

必须是 java 1.8.0_73 以上
安装 Java 版本 ,根据教程下载相应的版本
http://blog.csdn.net/fenglailea/article/details/26006647

0. 介绍

Logstash是一款轻量级的日志搜集处理框架，可以方便的把分散的、多样化的日志搜集起来，并进行自定义的处理，然后传输到指定的位置，比如某个服务器或者文件。
http://kibana.logstash.es/content/logstash/
风来了.fox

1.下载和安装

下载地址
https://www.elastic.co/downloads/logstash
目前最新版 5.2.1
这里使用 TAR.GZ 源码安装，即方式一
方式一：TAR

wget https://artifacts.elastic.co/downloads/logstash/logstash-5.2.1.tar.gz
tar -zxvf logstash-5.2.1.tar.gz

测试是否安装成功

bin/logstash -e 'input { stdin { } } output { stdout {}}'

2.配置

创建配置目录
先进入 Logstash 根目录

mkdir -p etc
vim etc/www.lanmps.com.conf

etc/test.conf 文件内容

input {
  file {
    type => "nginx-access"
    path => ["/www/wwwLogs/www.lanmps.com/*.log"]
    start_position => "beginning"
  }
}

filter {
    grok {
        "message"=>"%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-)\" (%{HOSTNAME:domain}|-) %{NUMBER:response} (?:%{NUMBER:bytes}|-) (%{QS:referrer}) %{QS:agent} \"(%{WORD:x_forword}|-)\" (%{URIHOST:upstream_host}|-) (%{NUMBER:upstream_response}|-) (%{WORD:upstream_cache_status}|-) %{QS:upstream_content_type} (%{USERNAME:upstream_response_time}) > (%{USERNAME:response_time})"
　　　　#匹配模式 message是每段读进来的日志，IP、HTTPDATE、WORD、NOTSPACE、NUMBER都是patterns/grok-patterns中定义好的正则格式名称，对照上面的日志进行编写,冒号，(?:%{USER:ident}|-)这种形式是条件判断，相当于程序里面的二目运算。如果有双引号""或者[]号，需要在前面加\进行转义。
    }
    kv {
                source => "request"
                field_split => "&?"
                value_split => "="
        }
　　#再单独将取得的URL、request字段取出来进行key-value值匹配，需要kv插件。提供字段分隔符"&?"，值键分隔符"="，则会自动将字段和值采集出来。
    urldecode {
        all_fields => true
    }
　　#把所有字段进行urldecode（显示中文）
}

output {
  elasticsearch {
        hosts => ["10.1.5.66:9200"]
        index => "logstash-%{type}-%{+YYYY.MM.dd}"
        document_type => "%{type}"
  }
}

配置说明
http://kibana.logstash.es/content/logstash/plugins/input/file.html

2.2 Nginx 日志格式定义

log_format access '$remote_addr - $remote_user [$time_local] "$request" $http_host $status $body_bytes_sent "$http_referer" "$http_user_agent" "$http_x_forwarded_for" $upstream_addr $upstream_status $upstream_cache_status "$upstream_http_content_type" $upstream_response_time > $request_time';

3.启动和停止

3.1测试命令

测试logstash

bin/logstash -e 'input { stdin { } } output { stdout {codec=>rubydebug} }'

然后你会发现终端在等待你的输入。没问题，敲入 Hello World，回车，然后看看会返回什么结果！
出现如下结果

2017-02-23T08:34:25.661Z c-101 Hello World

3.2 测试配置文件是否 正确

bin/logstash -t -f etc/

3.3 启动

加载 etc文件夹下所有 *.conf 的文本文件，然后在自己内存里拼接成一个完整的大配置文件

bin/logstash -f etc/

后台运行

nohup bin/logstash -f etc/ &

3.4停止

查找进程 ID

ps -ef |grep logstash

KILL 他

kill -9  id

4.高级配置

http://kibana.logstash.es/content/logstash/get_start/full_config.html
http://kibana.logstash.es/content/logstash/plugins/output/elasticsearch.html

插件安装
http://www.jianshu.com/p/4fe495639a9a

5.插件

5.1 grok,useragent,urldecode和kv插件功能

./bin/logstash-plugin install kv
./bin/logstash-plugin install urldecode
./bin/logstash-plugin install grok
./bin/logstash-plugin install useragent

使用方式

input {
    file {
        path => "/home/vovo/access.log"  #指定日志目录或文件，也可以使用通配符*.log输入目录中的log文件。
        start_position => "beginning" 
    }
}
filter {
    grok {
        match => {
        "message"=>"%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-)\" (%{HOSTNAME:domain}|-) %{NUMBER:response} (?:%{NUMBER:bytes}|-) (%{QS:referrer}) %{QS:agent} \"(%{WORD:x_forword}|-)\" (%{URIHOST:upstream_host}|-) (%{NUMBER:upstream_response}|-) (%{WORD:upstream_cache_status}|-) %{QS:upstream_content_type} (%{USERNAME:upstream_response_time}) > (%{USERNAME:response_time})"
        }
    }
    kv {
                source => "request"
                field_split => "&?"
                value_split => "="
        }
　　#再单独将取得的URL、request字段取出来进行key-value值匹配，需要kv插件。提供字段分隔符"&?"，值键分隔符"="，则会自动将字段和值采集出来。
    urldecode {
        all_fields => true
    }
　　#把所有字段进行urldecode（显示中文）
}

5.2 MYSQL 插件 logstash-input-jdbc插件

logstash-input-jdbc插件是logstash 的一个个插件。
使用ruby语言开发。
下载插件过程中最大的坑是下载插件相关的依赖的时候下不动，因为国内网络的原因，访问不到亚马逊的服务器。
解决办法，改成国内的ruby仓库镜像。此镜像托管于Ruby China的阿里云服务器上 ：
https://gems.ruby-china.org/

5.3 如果没有安装 gem 的话 安装gem

sudo yum install gem

替换Ruby China

gem sources --add https://gems.ruby-china.org/ --remove https://rubygems.org/
gem sources -l

第二次(gem sources -l)输出

*** CURRENT SOURCES ***

https://gems.ruby-china.org/  

注意

# 请确保只有 gems.ruby-china.org
如果 还是显示 https://rubygems.org/ 进入 home的 .gemrc 文件
sudo vim ~/.gemrc 
手动删除 https://rubygems.org/

修改Gemfile的数据源地址

#这里先要切换到root用户下执行
su -
sudo gem install bundler
bundle config mirror.https://rubygems.org https://gems.ruby-china.org

安装

最好早晨9点左右安装，这个时候速度比较快

bin/logstash-plugin install logstash-input-jdbc

显示如下表示安装成功

Validating logstash-input-jdbc
Installing logstash-input-jdbc
Installation successful

logstash-input-jdbc官方参考文档
https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html
重点字段说明：
schedule：设置监听间隔。可以设置每隔多久监听一次什么的。具体参考官方文档。
statement_filepath： 执行的sql 文件路径+名称

mysql 的Java 驱动包

https://dev.mysql.com/downloads/connector/j/
mysql-connector-java-5.1.40-bin.jar

x.报错

x.1 Pipeline aborted due to error {:exception=>”LogStash::ConfigurationError”,

原因：这类错误都是 参数不存在或参数写的不对造成，
例如 elasticsearch 新版host已改为hosts,port 已删除

部分来源：
http://blog.csdn.net/yeyuma/article/details/50240595