logstash详解

介绍
Logstash：数据处理引擎，可以处理每秒几万条的日志；它支持动态的从各种数据源搜集数据，并对数据进行过滤、分析、丰富、统一格式等操作，然后存储到 ES

官网地址：https://www.elastic.co/cn/products/logstash

如图：

详解：
下面我们将文本数据同步到ES来做例子，其中版本为：logstash-5.2.1，elasticsearch-6.2.2

analysis.conf 配置文件内容为

input {
    file {
        type => "lualog"
        path=> [ "C:/JWD/logs/*" ] #同步文件的IP
        start_position => beginning
        stat_interval => 1 #设置多长时间检测文件是否修改 默认是1s
        #tags => "test" #添加标签
        #设置多长时间扫描目录，发现新文件
        discover_interval => 15
        # 设置监听间隔 各字段含义（从左至右）分、时、天、月、年，全为*默认含义为每分钟都更新
    }
    file {
        type => "lsdclog"
        path=> [ "C:/JWD/server/Debug/Logs/*" ]  #同步文件的IP
        start_position => beginning
        stat_interval => 1 #设置多长时间检测文件是否修改 默认是1s
        #tags => "test" #添加标签
        #设置多长时间扫描目录，发现新文件
        discover_interval => 15
        # 设置监听间隔 各字段含义（从左至右）分、时、天、月、年，全为*默认含义为每分钟都更新
    }
}
 
filter {
   json{
        source => "message"
    }
    date {
        match => ["addTime", "yyyy.MM.dd HH.mm.ss","UNIX_MS"] #addTime是我日志中的一个时间字段！
        target => "@timestamp"
    }
    ruby {
        code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
    }
    ruby {
        code => "event.set('@timestamp',event.get('timestamp'))"
    }
    mutate{
        remove_field => ["message","timestamp"]
    }
}
output {
     if [type] == "lualog"{
        elasticsearch {
            action => "index"
            hosts => ["127.0.0.1:9200"] 
            index => "ls_%{+YYYYMMdd}_lua_log"
            document_type => "ls_%{+YYYYMMdd}_lua_log"
        }
     }
     if [type] == "lsdclog"{
        elasticsearch {
            action => "index"
            hosts =>["127.0.0.1:9200"] 
            index => "ls_%{+YYYYMMdd}_lsdc_log"
            document_type => "ls_%{+YYYYMMdd}_lsdc_log"
        }
     }
     stdout {
       codec => json
     }
}

1、将C:/JWD/logs/ 目录下的所有文件同步到本机ES中，索引名为ls_%{+YYYYMMdd}_lua_log，{+YYYYMMdd}为当日的日期

2、将C:/JWD/server/Debug/Logs/ 目录下的所有文件同步到本机ES中，索引名为ls_%{+YYYYMMdd}_lsdc_log，{+YYYYMMdd}为当日的日期

3、filter表示只过滤json数据

注意：
1、这边有个logstash自带的坑，就是logstas同步日志的时候，索引是统计8点到第二天8点影响，所以要改索引建立为凌晨建立时间，如何修改，请看以上配置

2、要同步多个路径下的文件的时候，可以在input中file的 path=> [ “C:/JWD/logs/", "C:/JWD/server/Debug/Logs/”] 配置多个路劲（input中各个参数的意义，请参考：https://yq.aliyun.com/articles/152043?spm=a2c4e.11153940.blogcont154341.23.792e51fekIVnjR）

3、output的时候，如果在hosts =>[“127.0.0.1:9200”,“10.40.0.205:9200”] 配置多个路径的时候，不是在两台服务上各备份一份数据，而是把一份数据存到两台服务上，所有每台服务上的数据都是不完整的

启动代码：logstash -f analysis.conf （其中analysis.conf为配置文件的文件名，在bin目录下）