背景

ELK为日志存储/查询系统，用于监控系统数据源和日志查询

目前日志系统主要解决方案为ELK，整条链路为采集(filebeat)、缓冲 (kafka)[可选]、过滤（logstash）、搜索（elasticsearch）、展示（kibana）

1、日志落地

1.1 终端日志

终端会上报数据到指定的日志服务器

1.2 nginx日志

nginx日志落地在日志文件夹下

2、日志采集

2.1 filebeat配置说明

filebeats部署9.59.x.x和9.131.x.x 上，通过配置paths路径采集日志文件，然后投递到logstash或者kafka、有大量日志写入需要写入kafka进行缓冲，如果没有大量日志可以直接投递到logstash

filebeat.spool_size: 5000
filebeat.idle_timeout: "5s"
filebeat.prospectors:- input_type: logpaths:- /usr/local/tnginx/logs/www.joox.com.access.log- /usr/local/tnginx/logs/www.joox.com_445.access.logfields_under_root: truefields:feature: nginx-webjooxdocument_type: webjoox_nginx
- input_type: logpaths:- /usr/local/tnginx/logs/api_joox.access.log- /usr/local/tnginx/logs/api_joox_445.access.logfields_under_root: truefields:feature: nginx-apijooxdocument_type: apijoox_nginx- input_type: logpaths:- /data/joox-web-ssr/logs/JOOX-WEB-SSR-ERROR.logfields_under_root: truefields:feature: nginx-pm2ssrdocument_type: pm2joox_nginx
#output.logstash:
#   #loadbalance: true
#   #hosts: ["10.228.175.10:5044","10.228.14.205:5044"]
#   hosts: ["10.228.180.102:5045"]
#   #hosts: ["10.228.139.36:5045"]output.kafka:hosts: ["kafka.iops.woa.com:9092"]topic: web_joox2required_acks: 1compression: none

2.2 启动filebeat

~]# cd /usr/local/elk/filebeat_2000701487_15/ && filebeat -e -c filebeat-app.yml

3、日志缓冲【可选】

在腾讯云消息队列Ckafka 实例 joox-iops 下 topic 可以找到 web_joox2的topic

消息队列 CKafka 技术原理 - 产品简介 - 文档中心 - 腾讯云

4、日志过滤

4.1 Logstash 负责日志输入、过滤、输出

Logstash 通过容器启动，在k8s跳板机【9.81.x.x】执行如下命令来理解logstash数据来源和去向。

1、因为filebeats配置文件中配置了投递地址为9.26.x.x:17007，所以先找到service，发现是nodeport类型的service，端口为17007

[root@k8s-jump ~]# kubectl get svc -n iops | grep jooxapp

logstash-jooxapp-fb NodePort 10.97.93.0 <none> 17007:17007/TCP  466d

2、找到名称为logstash-jooxapp-fb的deployment，查看创建yaml文件，发现logstash配置文件通过configmap来挂载使用，配置文件在容器路径为/usr/share/logstash/pipeline/indexer-kafka-named-k8s.conf

[root@k8s-jump ~]# kubectl get deploy logstash-jooxapp-fb -n iops -o yaml

spec:progressDeadlineSeconds: 600replicas: 3revisionHistoryLimit: 10selector:matchLabels:app: logstash-jooxapp-fbstrategy:rollingUpdate:maxSurge: 25%maxUnavailable: 25%type: RollingUpdatetemplate:metadata:creationTimestamp: nulllabels:app: logstash-jooxapp-fbspec:containers:- env:volumeMounts:- mountPath: /usr/share/logstash/pipelinename: vm-configdnsPolicy: ClusterFirstterminationGracePeriodSeconds: 30volumes:- configMap:defaultMode: 420items:- key: logstash-config-named-k8spath: indexer-kafka-named-k8s.confname: logstash-jooxapp-fbname: vm-config

3、查看logstash通过configmap挂载的indexer-kafka-named-k8s.conf配置文件，主要逻辑为通过读取filebeats的日志输入，然后投递到ES

[root@k8s-jump ~] # kubectl exec -it logstash-jooxapp-fb-76bb48b96-zh2c2 /bin/bash -n iops

beac38d ] # cat /usr/share/logstash/pipeline/indexer-kafka-named-k8s.conf

filter {if [type] == "cmsjoox_nginx" {ruby {init => "@kname = ['remote_addr','unknow','remote_user','time_local','request','status','body_bytes_sent','http_referer','http_user_agent','http_x_forwarded_for','host','cookie_ssl_edition','upstream_addr','upstream_status','request_time','upstream_response_time','cookie_uin','cookie_luin','proxy_host']"code => "v = Hash[@kname.zip(event.get('message').split('|'))]new_event = LogStash::Event.new(v)new_event.remove('@timestamp')event.append(new_event)event.set('raw_request', event.get('request'))req = event.get('request').split(' ')[1];if req.index('?') == ''event.set('request',req);elseevent.set('request',req.split('?')[0]);endevent.set('upstream_response_time_ext',event.get('upstream_response_time'));event.set('request_time_ext',event.get('request_time'));"}geoip {source => "remote_addr"}date {match => ["time_local","dd/MMM/yyyy:HH:mm:ss Z"]target => "@timestamp"}mutate {convert => ["upstream_response_time_ext", "float"]convert => ["request_time_ext", "float"]                      }}}## logstash confinput {kafka{bootstrap_servers => ["kafka.iops.woa.com:9092"]client_id => "${NODE_NAME}"group_id => "joox_cms"auto_offset_reset => "latest"decorate_events => "true"topics => ["cmsjoox"]codec => json       #一定要加上这段，不然传输错误，${message}}}## logstash confoutput {if "parsefailure" in [tags] {stdout {codec => rubydebug}}else{elasticsearch {hosts => ["http://iLogstack.oa.com:19200/es-logstashing/"]index => "cmsjoox_%{+YYYY.MM.dd}"}}}

4、logstash-joox-cms 结构解析

logstash pipeline 包含两个必须的元素：input和output，和一个可选元素：filter。

从input读取事件源，（经过filter解析和处理之后），从output输出到目标存储库（elasticsearch或其他）

4、日志存储和搜索（ES）

4.1 分布式、RESTful 风格的搜索和数据分析引擎

1、logstash 投递到ES的数据保留7天，webjoox_nginx 的index格式为 logstash-webjoox_nginx_%{+YYYY.MM.dd}

4.2 定时清理ES索引

9.81.x.x 每5分钟会执行一次的清理脚本

*/5 9-23 * * * cd /data/iLogStack/es_plugins/ops && sh delete_indices.sh > /dev/null 2>&1

#!/bin/bash
##rossypli at 2018/03/25###################################
#删除早于7天的ES集群的索引
###################################function delete_indices() {Today=`date +"%Y.%m.%d %H:%M:%S"`#修复匹配%{ostype}问题indices=`echo "$1" | sed 's/%/*/g' | sed 's/{/*/g' | sed 's/}/*/g'`curl -XDELETE http://iLogStack.oa.com:19200/es-monitoring/$indicesToday_end=`date +"%Y.%m.%d %H:%M:%S"`echo "DELETE indices|$Today|$Today_end|$indices" >> log/DELETE.logecho "DELETE $indices SUCC"exit   ##因为删除大索引会影响ES稳定性，所以删除一个索引后退出，等下次调用再删除下一个
}
function filter_indices(){comp_date=`date -d "$delete_day day ago" +"%Y.%m.%d"`# V updated for closed indices on 20181130curl -XGET http://iLogStack.oa.com:19200/es-monitoring/_cat/indices |grep $indices|grep $comp_date| awk -F "$comp_date" '{print $1}' | awk -F " " '{print $NF}' | sort | uniq| while read LINEdo#调用索引删除函数delete_indices ${LINE}$comp_datedone
}
function filter_indices_default(){   ##默认删除为7天comp_date=`date -d "7 day ago" +"%Y.%m.%d"`#curl -XGET http://iLogStack.oa.com:19200/es-monitoring/_cat/indices |grep -Ev "$indices_set"|grep $comp_date| awk -F" " '{print $3}' | sort | uniq| while read LINEcurl -XGET http://iLogStack.oa.com:19200/es-monitoring/_cat/indices |grep -Ev "$indices_set"|grep $comp_date| awk -F "$comp_date" '{print $1}' | awk -F " " '{print $NF}' | sort | uniq| while read LINEdo#调用索引删除函数delete_indices ${LINE}$comp_date#break   ##因为删除大索引会影响ES稳定性，所以删除一个索引后退出，等下次调用再删除下一个done
}##如果索引有设置存储时长，按存储时长删除，如果没有设置，则按默认的7天进行删除
grep -v '#' indices.conf |while read line
doindices=$(echo $line |awk '{print $1}')delete_day=$(echo $line |awk '{print $2}')filter_indicesecho $indices >> indices_tmp
done
sed -i ':a;N;$!ba;s/\n/|/g' indices_tmp ##换行符替换为逗号
indices_set=$(cat indices_tmp)
rm -f indices_tmp
filter_indices_default

5、日志展示（Kibana）

5.1 Kibana 负责数据可视化、数据分析

1、Kibana支持创建索引匹配，然后在kibana上选择创建的索引模版进行索引匹配即可

2、在kibana选择一个索引模版，进行搜索即可

基于ELK的日志系统最佳实践相关推荐

【Elasticsearch】Elasticsearch日志场景最佳实践
1.概述转载:Day 12 - Elasticsearch日志场景最佳实践相似文章:[Elasticsearch]Elasticsearch 最佳实践系列之分片恢复并发故障 Elasticsear ...
Java日志记录最佳实践
作者:GeekerLou www.jianshu.com/p/546e9aace657 一.日志简介 1.1 日志是什么(WHAT) 日志:记录程序的运行轨迹,方便查找关键信息,也方便快速定位解决问题 ...
Java日志管理最佳实践
原文出处:http://www.ibm.com/developerworks/cn/java/j-lo-practicelog/. 感谢原作者,感谢ibm网站,里面有好多的精华帖. 日志记录是应用程序 ...
日志管理最佳实践：成功的六要诀【解读版】
合适的日志管理工具能够大幅减轻管理企业系统日志数据的负担.但是,除非组织为这个工具投入必要的时间和精力,否则再好的工具也会很快变成一个差劲的工具.Diana Kelley为大家提供了6个确保成功的日志 ...
基于ITIL的SCOM监控最佳实践
1. 按照系统类别进行监控很多朋友在使用SCOM进行监控的时候,往往只是导入管理包,推送代理,并不会思考很多,那么在这种情况下,SCOM在进行监控的时候都是基于缺省的类对象进行监控,比如说Wind ...
ELK+kafka日志系统搭建-实战
日志主要包括系统日志.应用程序日志和安全日志.系统运维和开发人员可以通过日志了解服务器软硬件信息.检查配置过程中的错误及错误发生的原因.经常分析日志可以了解服务器的负荷,性能安全性,从而及时采取措施纠 ...
阿里云数据库快速搭建疫情分析系统最佳实践
简介:疫情降临,疫情态势分析和防控任务迫在眉睫,如果快速搭建高效的疫情态势分析系统是众多部门和单位的难题,阿里云RDS PG+Ganos解决方案可在极短时间内完成分析系统搭建,有效助力疫情防控. 直达 ...
jsp给前端注入值失败_基于 qiankun 的微前端最佳实践（图文并茂）应用间通信篇...
引言大家好~ 本文是基于 qiankun 的微前端最佳实践系列文章之应用间通信篇,本文将分享在 qiankun 中如何进行应用间通信. 在开始介绍 qiankun 的应用通信之前,我们需要先了解微 ...
干货：基于 Git Flow 的 Git 最佳实践（附加解决大家经常碰到的问题）
干货:基于 Git Flow 的 Git 最佳实践(附加解决大家经常碰到的问题) 参考文章: (1)干货:基于 Git Flow 的 Git 最佳实践(附加解决大家经常碰到的问题) (2)https: ...

基于ELK的日志系统最佳实践

背景