第一章 日志收集1.1 filebeat1.1.1 filebeat作用filebeat是一款部署在服务器上的日志收集工具, 用来实时收集指定服务的日志, 并对日志进行指定输出注意:filebeat的输出目标同一时间只能有一个filebeat只有第一次收集日志时, 才会从文件头开始读取filebeat收集的日志格式为json, 每一行日志都是一个事件1.1.2 安装filebeat下载地址: https://mirrors.aliyun.com/elasticstack/7.x/yum/7.17.29/?spma2c6h.25603864.0.0.7975505007UAnD安装filebeattar -zxvf ./filebeat-7.13.2-linux-x86_64.tar.gz -C /usr/localmv /usr/local/filebeat-7.13.2 /usr/local/filebeat配置filebeat的systemctl启动vim /usr/lib/systemd/system/filebeat.service[Unit]DescriptionFilebeat sends log files to Logstash or directly to Elasticsearch.Wantsnetwork-online.targetAfternetwork-online.target[Service]UserrootGrouprootEnvironment“BEAT_CONFIG_OPTS-c /usr/local/filebeat/filebeat.yml”ExecStart/usr/local/filebeat/filebeat $BEAT_CONFIG_OPTSRestartalways[Install]WantedBymulti-user.targetsystemctl daemon-reload注意:filebeat两种启动方式: 前台启动, 后台启动前台启动: 可以在控制台直接查看日志内容后台启动: nohup与systemctlnohup启动, 日志结果可以在nohup.out文件中查看systemctl启动, 无法查看输出日志filebeat的简单使用配置filebeat读取指定文件中的日志vim /usr/local/filebeat/filebeat.ymlinputs日志来源filebeat.inputs:type: log#开启filebeat的读取日志功能enable: true#日志来源paths:/var/log/httpd/access_log#配置调用filebeat的模块来收集日志filebeat.config.modules:#调用filebeat的所有模块来收集日志path: ${path.config}/modules.d/*.yml#filebeat的配置文件更新时, 该程序不自动加载更新后的配置reload.enabled: falsesetup.template.settings:index.number_of_shards: 1将filebeat读取到的日志输出到控制台output.console:pretty: truefilebeat对收集的日志进行处理processors:add_host_metadata:when.not.contains.tags: forwarded启动filebeat, 控制台观察日志的输出cd /usr/local/filebeat/usr/local/filebeat/filebeat或/usr/local/filebeat/filebeat -c /usr/local/filebeat/filebeat.yml1.1.3 filebeat的日志收集模块查看filebeat支持的收集模块/usr/local/filebeat/filebeat modules list#支持nginx, mysql, apache, redis等注意:enabled表示启用的模块disabled表示未启用的模块启用apache模块/usr/local/filebeat/filebeat modules enable apache#或vim /usr/local/filebeat/filebeat/modules.d/apache.ymlaccess:将false改为trueenabled: truevar.path: [“/usr/httpd/access_log1”,“/usr/httpd/access_log2”]error:enabled: true3. 执行测试/usr/local/filebeat/filebeatsystemctl stop httpdsystemctl start httpdhttp://192.168.255.132注意:若服务日志的存放路径不是默认路径, 在模块对应的配置文件中, 使用var.paths来指定即可var.paths指定的路径与默认配置不冲突, 若默认配置与var.paths中都存在日志, 那么filebeat都会收集filebeat常用output: console(输出到控制台), kafka(输出到kafka), logstash(输出到logstash), elasticsearsh(输出到elasticsearch)output只能指定一个输出方式1.1.4 filebeat的重读日志由于filebeat只会在第一次读日志时, 会从头开始读, 若需要再次重读日志, 删除/usr/local/filebeat/filebeat/data/*即可1.1.5 filebeat对日志的过滤与增强过滤含有core:notice的行vim /usr/local/filebeat/filebeat.ymlprocessors:#删除info事件drop_event:when:regexp:message: “core:notice”#运行/usr/local/filebeat/filebeat输出事件中添加指定内容vim /usr/local/filebeat/filebeat.ymlprocessors:#事件中添加json段落add_fields:#输出时添加json段落testtarget: test#test的json中包含字段id, namefields:name: zsid: 123456#运行/usr/local/filebeat/filebeat删除事件中的字段vim /usr/local/filebeat/filebeat.ymlprocessors:drop_fields#删除事件中host的json段落, 删除test段落中的name字段fields: [“host”,“test.name”]#若删除的字不存在, 忽略他ignore_missing: true#运行/usr/local/filebeat/filebeat1.2 logstash1.2.1 lostash作用logstash是一款部署在服务器上的日志收集工具, 用来实时收集指定服务的日志, 并对日志进行指定输出, 常用于对filebeat收集的日志进行过滤注意:logstash的三大组件: input, filter, output1.2.2 安装logstash#解压并重命名tar -zxvf ./logstash-7.13.2.tar.gz -C /usr/localmv /usr/local/logstash-7.13.2 /usr/local/logstash1.2.3 简单测试logstash#使用logstash的标准输入输出, 并启动logstash-e 等同于 -e input { stdin { type stdin } } output { stdout { codec rubydebug } }表示logstash从键盘接收输入, 使用rubydebug格式输入/usr/local/logstash/bin/logstash -e ‘’启动后, 使用键入内容进行测试1.2.4 配置logstash的管道输入输出vim /usr/local/logstash/config/first-pipeline.confinput {#使用标准输入stdin {}}output {#使用标准输出stdout {}}#运行测试, -f用于指定logstash启动时加载的管道文件/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/first-pipeline.conf#控制台中输入内容测试管道文件是否能被正常使用注意:logstash支持多输入, 多输出以及多种过滤器1.2.5 Grok过滤器vim /usr/local/logstash/config/two-pipeline.confinput {#采用文件形式的输入file {#文件路径path [“/var/log/httpd/access_log”, /var/log/httpd/error_log]#从文件第一行开始读start_position “beginning”}}#使用logstash的filter对日志内容进行过滤filter {#使用grok插件, 对httpd的日志进行格式化gork {#match表示匹配字段, 对message字段, 使用combinedapachelog模式, 对apache的访问与错误日志进行格式化match { “message” “%{COMBINEDAPACHELOG}” }}mutate {#对结构化的字段重命名rename {“timestamp” “access_time”}}mutate {#输出时不显示message与ident字段remove_field [“message”,“ident”]}}output {#过滤后执行标准输出stdout {}}#测试/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/two-pipeline.conf注意:grok的combinedapachelog模式, 是对apache服务产生的日志进行格式化1.2.6 filebeat输入logstash接收配置logstashvim /usr/local/logstash/config/filebeat.confinput {#beats插件用于接收其他程序的数据, 运行在5044端口beats {port 5044}}filter {grok {match { “message” “%{COMBINEDAPACHELOG}” }}}output {stdout {}}配置filebeat的apache模块vim /usr/local/filebeat/modules.d/apache.ymlfilebeat开启对apache的访问与错误日志的收集module: apacheaccess:enabled: trueerror:enabled: true配置filebeat的输出vim /usr/local/filebeat/filebeat.yml#output模块中修改output.logstash:#filebeat收集到的内容, 输出到logstash主机的5044的beats端口hosts: [“192.168.255.132:5044”]删除filebeat的缓存#由于filebeat对度过的内容不会从头开始读, 删除缓存, 重新读apache的日志文件rm -rf /usr/local/filebeat/data运行filebeat与logstash/usr/local/filebeat/filebeat/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/filebeat.conf第二章 日志存储2.1 ElasticSearch2.1.1 elasticsearch作用用于存储logstash与filebeat收集的日志, 并对存储的日志进行搜索与分析注意:elasticsearch支持的数据类型: 结构化数据, 非结构化数据, 数据字典, 地理空间数据elasticsearch的最小数据单元是文档文档是指json化后的一条数据elasticsearch会对每一个文档添加索引, 方便检索2.1.2 es集群es集群是将数据进行分布式存储, 具有高可用, 易扩展的特点2.1.2.1 es集群节点主节点: 负责管理集群的状态与分片的分配, 不会存储与查询数据数据节点: 负责数据主分片与副本分片的存储, 写入, 查询等工作2.1.2.2 主分片与副本分片主分片: 将数据均匀的拆分, 存储到不同数据节点中副本分片: 主分片的备份注意:主分片与其副本分片不能存储在同一数据节点, 避免数据丢失2.1.3 es集群部署环境filebeat: 192.168.255.130kibana: 192.168.255.131logstash: 192.168.255.132es:192.168.255.133192.168.255.134192.168.255.135kafka:192.168.255.141192.168.255.142192.168.255.144配置服务器#关闭防火墙以及selinuxsystemctl disable firewalldsed -ri ‘s/SELINUXenforcing/SELINUXdisabled/g’ /etc/sysconfig/selinuxsetenforce 0#配置域名解析cat /etc/hosts EOF192.168.255.130 filebeat192.168.255.132 logstash192.168.255.133 es1192.168.255.134 es2192.168.255.135 es3EOF安装并配置filebeattar -zxvf ./filebeat-7.13.2-linux-x86_64.tar.gz -C /usr/localmv /usr/local/filebeat-7.13.2-linux-x86_64 /usr/local/filebeat#开启apache模块mv /usr/local/filebeat/modules.d/apache.yml.disabled /usr/local/filebeat/modules.d/apache.ymlvim /usr/local/filebeat/modules.d/apache.ymlmodule: apacheaccess:enabled: trueerror:enabled: true#配置filebeat的输出vim /usr/local/filebeat/filebeat.ymlfilebeat.config.modules:path: ${path.config}/modules.d/*.ymloutput.logstash:hosts: [192.168.255.132:5044]安装并配置logstash#安装logstashtar -zxvf logstash-7.13.2-linux-x86_64.tar.gz -C /usr/localmv /usr/local/logstash-7.13.2 /usr/local/logstash#配置logstash的输入输出, 以及过滤器vim /usr/local/logstash/config/pipe.conf#输入监听本机5044端口input {beats {port 5044}}#过滤器filter {#采用logstash的格式化输出grok{match { “message” “%{COMBINEDAPACHELOG}” }}#过滤掉message字段mutate {remove_field [“message”]}}#输出到es集群output {elasticsearch {hosts [“es1:9200”, “es2:9200”, “es3:9200”]}}安装elasticsearchyum -y install elasticsearch-7.13.2-x86_64.rpmsystemctl enable elasticsearch配置elasticsearch#133配置节点1vim /etc/elasticsearch/elasticsearch.yml#集群名称cluster.name: es#节点名称node.name: es1#是否是数据节点node.data: true#本机任意ip的9200端口, 都可接收logstash的信息network.host: 0.0.0.0#es的工作端口http.port: 9200#配置集群节点discovery.seed_hosts:es1es2es3#集群中主节点的选举范围cluster.initial_master_nodes: [“es1”,“es2”,“es3”]#134配置节点2vim /etc/elasticsearch/elasticsearch.yml#集群名称cluster.name: es#节点名称node.name: es2#是否是数据节点node.data: true#本机任意ip的9200端口, 都可接收logstash的信息network.host: 0.0.0.0#es的工作端口http.port: 9200#配置集群节点discovery.seed_hosts:es1es2es3#集群中主节点的选举范围cluster.initial_master_nodes: [“es1”,“es2”,“es3”]#135配置节3vim /etc/elasticsearch/elasticsearch.yml#集群名称cluster.name: es#节点名称node.name: es3#是否是数据节点node.data: true#本机任意ip的9200端口, 都可接收logstash的信息network.host: 0.0.0.0#es的工作端口http.port: 9200#配置集群节点discovery.seed_hosts:es1es2es3#集群中主节点的选举范围cluster.initial_master_nodes: [“es1”,“es2”,“es3”]#启动服务systemctl start elasticsearch#互相检查集群主机健康状态绿色表示正常;黄色表示单节点部署,不具有备份能力;红色表示数据不可用curl -X GET “localhost:9200/_cat/health?v”检查集群节点的运行状态curl -X GET “localhost:9200/_cat/nodes?v”注意:es的9200端口对外提供被访问功能, 用于接收数据es的9300端口是集群中主机与主机之间的通信, 用于选举主节点主节点用*表示, 数据节点用-表示开启logstash与filebeat/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/pipe.conf/usr/local/filebeat/filebeat -c /usr/local/filebeat/filebeat.yml验证es集群是否创建索引curl -X GET “localhost:9200/_cat/indices”logstash配置指定索引vim /usr/local/logstash/config/pipe.confoutput {#httpd的访问日志if [event][dataset] “apache.access” {elasticsearch {hosts [“es1:9200”, “es2:9200”, “es3:9200”]#向es集群传输时创建http的访问日志索引index “%{[host][hostname]}-httpd-access-%{YYYY.MM.dd}”}}#http的错误日志else if [event][dataset] “apache.error” {elasticsearch {hosts [“es1:9200”, “es2:9200”, “es3:9200”]#向es集群传输时创建http的访问日志索引index “%{[host][hostname]}-httpd-error-%{YYYY.MM.dd}”}}}启动logstash/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/pipe.confes集群查看索引http://192.168.255.130systemctl stop httpdcurl -X GET “localhost:9200/_cat/indices”第三章 日志管理3.1 kibana3.1.1 作用用于对es集群中存储的日志进行可视化, 同时对日志进行分析和管理3.2 安装kibanatar -zxvf ./kibana-7.13.2-linux-x86_64.tar.gz -C /usr/localmv /usr/local/kibana-7.13.2 /usr/local/kibana3.3 将kibana部署在es集群上配置域名解析与kibana#延续之间es集群, 部署kibanacat /etc/hosts EOF192.168.255.130 filebeat192.168.255.131 kibana192.168.255.132 logstash192.168.255.133 es1192.168.255.134 es2192.168.255.135 es3EOF#配置kibanavim /usr/local/kibana/config/kibana.ymlserver.port: 5601server.host: “0.0.0.0”#连接es集群的地址与端口elasticsearch.hosts: [http://es1:9200,“http://es2:9200”,“http://es3:9200”]#配置kibana的日志存放路径logging.dest: /var/log/kibana/kibana.log#设置操作界面为中文il8n.locale: “zh-CN”2. 创建kibana的管理用户useradd kibanapasswd kibanamkdir /run/kibana /var/log/kibanachown -R kibana.kibana /run/kibana /var/log/kibana /usr/local/kibana3. 启动kibanasu - kibana#将kibana运行在后台nohup /usr/local/kibana/bin/kibana 4. 测试访问http://kinaba:56015. 创建索引[图片][图片][图片][图片][图片][图片]6. 查看数据[图片][图片]第四章 日志缓冲4.1 kafka4.1.1 作用用于数据缓冲队列, 临时存放数据, 避免高并发导致集群宕机, 具有削峰填谷的能力注意:kafka的运行依赖zookeeper4.1.2 特点高吞吐量: 每秒可以处理十万级的消息可扩展性: 支持热扩展可靠性: kafka的数据会被持久化到磁盘, 并且支持数据的备份容错性: 允许集群中的节点失败高并发: 支持数千用户的同时读写4.1.3 组件话题(topic): 逻辑概念, kafka将消息进行分类, 每一个类型就是一个topic生产者(producer): 消息的发送者, 将消息发送给kafka消费者(consumer): 消息的接收者, 读取kafka中的消息服务代理(broker): 处理生产者与消费者的请求, 一个集群有多个brokerpartition(区): topic的物理分片, 将一个topic进行多个分片存储,replication: 区的副本, 防止消息丢失, 每个分片可以有多个副本leader: 处理所有生产者的写入以及消费者的读取follower: 同步leader中的数据, 不处理请求, leader宕后, follower会自动作为leaderzookeeper: 对数据进行存储4.1.4 安装zookeeper#安装jdk, 支持zookeeper的运行yum -y install java-1.8.0-openjdk#配置域名解析cat /etc/hosts EOF192.168.255.141 kafka1192.168.255.142 kafka2192.168.255.144 kafka2EOF#安装kafkatar -zxvf kafka_2.12-2.8.0.tgz -C /usr/localmv /usr/local/kafka_2.12-2.8.0.tgz /usr/local/kafka4.1.5 配置kafka配置zookeepervim /usr/local/kafka/config/zookeeper.properties#指定数据的持久化路径dataDir/opt/data/zookeeper/data#指定日志存放路径dataLogDir/opt/data/zookeeper/logs#客户端与zookeepr的连接端口clientPort2181#zk服务器之间或客户端与zk服务器之间的心跳间隔2秒tickTime2000#follower与leader同步数据的初始化连接时间initLimit20#follower无法在指定时间内在leader上获得数据, 那么follower将会作为leadersyncLimit10#配置kafka集群, 2888是主从数据交换的端口, 3888从选举主的端口server.1192.168.255.141:2888:3888server.2192.168.255.142:2888:3888server.3192.168.255.143:2888:3888#创建日志与数据存放目录mkdir -p /opt/data/zookeeper/{logs,data}#192.168.255.141:1, 192.168.255.142:2, 192.168.255.143:3echo 1 /opt/data/zookeeper/data/myid2. 配置kafkavim /usr/local/kafka/config/server.properties#集群中broker的id, 不能相同broker.id1listenersPLAINTEXT://192.168.255.141:9092 #改#broker处理消息的线程数num.network.threads3#broker对数据持久化的IO线程数num.io.threads8#消息发送缓冲区socket.send.buffer.bytes102400#消息接收缓冲区socket.receive.buffer.bytes102400#请求的最大值socket.request.max.bytes104857600#日志路径log.dirs/opt/data/kafka/logs#分片数量num.partitions6num.recovery.threads.per.data.dir1offsets.topic.replication.factor2transaction.state.log.replication.factor1transaction.state.log.min.isr1log.retention.hours168log.segment.bytes536870912log.retention.check.interval.ms300000zookeeper.connect192.168.255.141:2181,192.168.255.142:2181,192.168.255.143:2181zookeeper.connection.timeout.ms6000group.initial.rebalance.delay.ms03. 启动zookeepernohup bin/zookeeper‐server‐start.sh config/zookeeper.properties 4. 启动kafkanohup bin/kafka‐server‐start.sh config/server.properties 5. 141创建topicbin/kafka‐topics.sh ‐‐create ‐‐zookeeper localhost:2181 ‐‐replication‐factor 1 ‐‐partitions 1 ‐‐topic testtopic6. 142查询topicbin/kafka‐topics.sh ‐‐zookeeper 192.168.255.142:2181 ‐‐list7. 配置logstash与kafkavim /usr/local/logstash/config/pipe.confinput {kafka {type “apache_log”codec “json”topics [“apache”]decorate_events truebootstrap_servers “192.168.255.141:9092, 192.168.255.142:9092, 192.168.255.143:9092”}}8. 配置filebeat与kafkavim /usr/local/filebeat/filebeat.ymloutput.kafka:hosts: [“kafka11:9092”, “kafka2:9092”, “kafka3:9092”]topic: ‘apache’partition.round_robin:reachable_only: falserequired_acks: 1compression: gzipmax_message_bytes: 10000009. 启动使用kibana验证