2023年7月17日发(作者：)

ELKelasticsearchkibana⽇志排序之⽇志⼆级排序背景：之前搭建ELK时候经常听开发⼈员反馈说⽇志的数据和服务器的⽇志顺序不⼀致， 看⽇志给他们带来许多烦恼问题分析：kibana向es(elasticsearch)发送请求的时候默认排序为@timestamp字段，然⽽@timestamp字段的精度是毫秒， 也就是说如果同⼀毫秒内输出多条⽇志则在kibana展⽰的⽇志就会出现和服务器⽇志不⼀致的问题。解决⽅案：1 调整⽇志收集把时间精度调整到微妙，当时确实也尝试过，好像不太容易实现(如果读者有该解决⽅案可以留⾔) 2 让kibana展⽰⽇志除了@timestamp 排序外在加⼀个⼆级排序⽅案实施：最终我选⽤了⽅案⼆， ⽅案说明： 我的采集器为filebeat， filebeat采集⽇志的时候会增加⼀个属性叫做offset字段， offset字段标识所读取⽇志的偏移量（⾏数）, ⽅案⼆思路：让kibana请求es的时候除了基于@timestamp外在基于offset排序，那么如果让kibana基于offset做⼆次排序呢， 实现⽅式为：拦截kibana 发送到es的请求(_msearch)需改请求内容添加基于offset排序。如何拦截kibana请求报⽂呢？以下是我的实现⽅式可供参考：实现⽅式⼀：我们⽣产环境的kibana是⽤到了ownhome插件 ，该插件其实就是⼀个代理服务，通过修改ownhome代码实现添加offset排序，修改如下：1. 打开own_home源码⽂件：vim {kibana_home}/plugins/own_home/server/proxy/modify_2. 添加如下脚本到该⽂件的 " function replaceRequestBody(body) { 下3. 重启kibana4. 搜索页⾯验证 // Replace in mget request body function replaceRequestBody(body) {// add default secondary sort for kibana, sort by offset field, add by zhangyuming

try{ if ( === '/_msearch'){ let lines = ('n'); for(var i = 0; i < ; i++){ if(lines[i].length > 2){ var tmp = (lines[i]); if( typeof(tmp) == "object" && Property("sort") && tmp["sort"].length == 1 && typeof(tmp["sort"][0]) == "object" && tmp["sort"][0].hasOwnProperty("@timestamp") && tmp["sort"][0]["@timestamp"]["order"] == "desc" ){ (['plugin:own-home', 'debug'], '################ : ' + i + ' : '+ lines[i]); var offsetSort = ("{"offset":{"order":"desc","unmapped_type":"boolean"}}"); tmp["sort"][1] = offsetSort; lines[i] = ify(tmp); body = ("n"); (['plugin:own-home', 'debug'], '################ : ' + body); } } } } } catch(err){ (['plugin:own-home', 'debug'], 'body add sort fail ' + body + ' : ' + err); }// end default sort for kibana if (!th('_mget')) { return new Buffer(body); }

实现⽅式⼆：我们测试环境没有⽤到ownhome插件，后来⾃⼰想了想写了代理⽤于修改kibana的报⽂的⼯具elsaticproxy。使⽤⽅式：1. 下载elasticproxy⼯具

2. 启动 ./elasticproxy -elastic_home you_es_host:es_port 默认服务端⼝为88993. 修改kibana的es指向指向elasticproxy的8899端⼝4. 验证排序项⽬说明：⽬前该项⽬只实现了kibana的⼆级排序， 当然你也可以基于该项⽬实现其他代理功能例如⽇志脱敏等代理服务， 项⽬的开发⽅式我稍后会完善到github上欢迎⼤家使⽤，或者PR