ELK日志收集之多文件提取文件名和日志时间

需求：多个设备的日志同时保存在一台服务器上，日志文件的文件名是设备的ID，需要将多个文件提取文件名作为最终的筛选字段，同时提取日志中的时候日期时间替换系统的@timestamp

filebeat配置：

filebeat.inputs:- type: logenabled: truepaths:- /opt/data/*.logtags: ["test-android-log"]fields:log_source: my_log_sourcefields_under_root: trueprocessors:- dissect:tokenizer: "/opt/data/%{filename}.log"field: "log.file.path"target_prefix: "file"output:logstash:hosts: ["192.168.0.102:5044"]

logstash配置：

input {beats {port => 5044}
}filter {if [file][filename] {mutate {add_field => { "device_no" => "%{[file][filename]}" }}}grok {match => { "message" => "%{MONTHNUM:month}-%{MONTHDAY:day} %{TIME:time} %{GREEDYDATA:log_message}" }add_field => { "timestamp" => "%{month}-%{day} %{time}" }}date {match => ["timestamp", "MM-dd HH:mm:ss.SSS"]target => "@timestamp"}mutate {remove_field => [ "timestamp", "month", "day", "time" ]}
}output {if "test-android-log" in [tags] {elasticsearch {hosts => ["192.168.0.101:9200"]index => "test-android_log_t2014"}}stdout { codec => rubydebug }
}

使用kibana的开发工具获取一下对应index的结果看下是否有想要的字段传过来

GET /test-android_log_t2014/_search
{"size": 1,"_source": ["device_no"]
}

我这边想要的是device_no，查看见过如下表示获取成功：

{"took": 1,"timed_out": false,"_shards": {"total": 1,"successful": 1,"skipped": 0,"failed": 0},"hits": {"total": {"value": 5392,"relation": "eq"},"max_score": 1,"hits": [{"_index": "test-android_log_t2014","_id": "vohlxZABk6v1MxO1ydv2","_score": 1,"_source": {"device_no": "20240718173333"}}]}
}

以上便完成了多个设备日志上传以及设备日志筛选，欢迎大家指正。