filebeat如何收集每天百万个日志文件

初步解决了filebeat收集日志，因为传递到logstash时生成的@timestamp一样(或者很接近)，导致kibana展示时顺序乱的问题
原理就是，用日志文件内容里面的时间来替换@timestamp的值，这样在kibana里面展示的时候按照这个时间排序就没有问题
（其实后来发现每条收集上去的日志都是有一个offset字段，使用offset字段排序也是能达到效果的）

1、logstash的配置如下
[root@localhost logstash-6.2.4]# cat config/logstash.conf
input{
beats {
port => "5044"
type => "xgdfin-test-filebeat"
}
}
filter {
grok {
patterns_dir => "/data/ELK/logstash-6.2.4/partterns/mypartterns.conf"
match => [ "message","%{SOALOG}" ]
}
date {
match => [ "logtime", "yyyyMMdd:HH:mm:ss.SSS" ]
locale => "en"
target => [ "@timestamp" ]
timezone => "Asia/Shanghai"
}
}
output{
elasticsearch {
hosts => ["10.18.6.38:9200"]
index => "logstash-%{type}-%{+YYYY.MM.dd}"
}
stdout {
codec => rubydebug
}
}
[root@localhost logstash-6.2.4]# cat partterns/mypartterns.conf
SOATIME (?:[0-9]{8}:[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{3})
LOGLEVEL ([A-a]lert|ALERT|[T|t]race|TRACE|[D|d]ebug|DEBUG|[N|n]otice|NOTICE|[I|i]nfo|INFO|[W|w]arn?(?:ing)?|WARN?(?:ING)?|[E|e]rr?(?:or)?|ERR?(?:OR)?|[C|c]rit?(?:ical)?|CRIT?(?:ICAL)?|[F|f]atal|FATAL|[S|s]evere|SEVERE|EMERG(?:ENCY)?|[Ee]merg(?:ency)?)
DATA .*?
SOALOG %{SOATIME:logtime} %{LOGLEVEL:loglevel} %{DATA:logmsg}
[root@localhost logstash-6.2.4]#

2、解析
(1)、先在filter里面是有grok插件中解析字段，使用正则表达式，将日志文件里面本身的日志时间解析出来，字段为logtime
(2)、然后继续在filter的date插件中将刚才解析出来的logtime字段提取，然后替换为@timestamp的值，这里需要注意的是 match => [ "logtime", "yyyyMMdd:HH:mm:ss.SSS" ] 后面的格式一定要和自己的日志内容里面的时间格式是匹配的，如下，我的日志内容里面时间是这样的
...
20181025:16:57:31.723 INFO execurting request：http://10.2.14.10:4440/uc/getEduInfo
20181025:16:57:31.723 INFO 请求参数：{"idNum":"******","userName":"余*"}
...

3、替换后结果
如下，除了因为时区问题相差8小时，@timestamp的时间已经被替换为日志内容里面的时间了

1. 推荐使用 filebeat 进行日志文件的采集，对于多个大的服务，你可以对 filebeat 进行拆分，不一定要在单个采集服务里面把全部日志都采集了，这样也不高效；
2. 关于日志的采集，推荐还是使用一条数据解析成一条日志，不是一行一条，因为日志与日志之间是有明确的分隔符的，filebeat 支持设置 multi_line 的分隔符，你根据你的日志进行设置（正则表达式），设置完成后，一条完整的日志会作为单个 message 进行传递到 logstash；
3. logstash 接收到 message 后，可以使用 gork 匹配到真实的日志时间，推荐将真实的日志时间存储两份，第一份转换为 UTC 时间后替换  @timestamp，第二份存储为另一个字段，例如叫 log_dtm；
4. 在 kibana 中进行时间选择的时候请选择 utc 时间的 @timestamp 字段即可，这样就没有时差了。

我觉得你应该优化日志生成逻辑，减少日志生成数，上百万个日志文件的收集对收集 agent(filebeat flume）来讲都是有压力的。为什么不能将所有的交易输出到一个日志文件里面呢？

第二点的正则你试试

\d{8}:\d{2}:\d{2}:\d{2}\.\d{3} [A-Z]{4,}

第三点通过 gork 的正则匹配到日期字段，使用 date 过滤器转换日期

filter {

  grok {

    match => {"message" => "(?<log_dtm>\S+) [A-Z]{4,}[\s\S]+"}

  }

  date{

    match=>["log_dtm","YYYYMMdd:HH:mm:ss.SSS", "ISO8601"]

    target=>"log_dtm"

  }

}