GO开发：用go写个日志监控系统

日志收集系统架构 1.项目背景

a. 每个系统都有日志，当系统出现问题时，需要通过日志解决问题

b. 当系统机器比较少时，登陆到服务器上查看即可满足

c. 当系统机器规模巨大，登陆到机器上查看几乎不现实

2.解决方案

a. 把机器上的日志实时收集，统一的存储到中心系统

b. 然后再对这些日志建立索引，通过搜索即可以找到对应日志

c. 通过提供界面友好的web界面，通过web即可以完成日志搜索

面临的问题

a. 实时日志量非常大，每天几十亿条

b. 日志准实时收集，延迟控制在分钟级别

c. 能够水平可扩展

ELK介绍

•官网https://www.elastic.co/cn/

• 中文指南https://www.gitbook.com/book/chenryn/elk-stack-guide-cn/details

• ELKStack （5.0版本之后）--> ElasticStack == （ELKStack + Beats）

• ELK Stack包含：ElasticSearch、Logstash、Kibana

• ElasticSearch是一个搜索引擎，用来搜索、分析、存储日志。它是分布式的，也就是说可以横向扩容，可以自动发现，索引自动分片，总之很强大。文档https://www.elastic.co/guide/cn/elasticsearch/guide/current/index.html

• Logstash用来采集日志，把日志解析为json格式交给ElasticSearch。

• Kibana是一个数据可视化组件，把处理后的结果通过web界面展示

• Beats在这里是一个轻量级日志采集器，其实Beats家族有5个成员

• 早期的ELK架构中使用Logstash收集、解析日志，但是Logstash对内存、cpu、io等资源消耗比较高。相比 Logstash，Beats所占系统的CPU和内存几乎可以忽略不计

• x-pack对ElasticStack提供了安全、警报、监控、报表、图表于一身的扩展包，是收费的

elk方案问题

a. 运维成本高，每增加一个日志收集，都需要手动修改配置

b. 监控缺失，无法准确获取logstash的状态

c. 无法做定制化开发以及维护

日志收集系统设计

Kafka消息队列

数据解耦

a. Log Agent，日志收集客户端，用来收集服务器上的日志

b. Kafka，高吞吐量的分布式队列，linkin开发，apache顶级开源项目

c. ES，elasticsearch，开源的搜索引擎，提供基于http restful的web接口

d. Hadoop，分布式计算框架，能够对大量数据进行分布式处理的平台

zookeeper

Zookeeper 作为一个分布式的服务框架，主要用来解决分布式集群中应用系统的一致性问题，它能提供基于类似于文件系统的目录节点树方式的数据存储， Zookeeper 作用主要是用来维护和监控存储的数据的状态变化，通过监控这些数据状态的变化，从而达到基于数据的集群管理

简单的说，zookeeper=文件系统+通知机制

a. 安装JDK，从oracle下载最新的SDK安装

b. 安装zookeeper3.3.6，下载地址：

1）mv conf/zoo_sample.cfg conf/zoo.cfg

2）编辑 conf/zoo.cfg，修改dataDir

# the directory where the snapshot is stored. dataDir=http://www.likecs.com/tmp/zookeeper/data # the port at which the clients will connect clientPort=2181 dataLogDir=http://www.likecs.com/tmp/zookeeper/log

3）vim /etc/profile

export PATH=$PATH:/usr/local/zookeeper/bin

source /etc/profile

运行:

[root@greg02 zookeeper]#zkServer.sh start JMX enabled by default Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg Starting zookeeper ... STARTED kafka

1.打开链接：

下载https://www.apache.org/dyn/closer.cgi?path=http://www.likecs.com/kafka/0.11.0.2/kafka_2.12-0.11.0.2.tgz

2.打开config目录下的server.properties， 修改log.dirs为D:\kafka_logs，修改advertised.host.name=服务器ip

3.启动kafka

[root@greg02 kafka]#kafka-server-start.sh config/server.properties kafka消费者开启 [root@greg02 kafka]#kafka-console-consumer.sh --topic nginx_log --zookeeper 127.0.0.1 2181 Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]. [2018-02-05 18:30:22,451] WARN Connected to an old server; r-o mode will be unavailable (org.apache.zookeeper.ClientCnxnSocket) [2018-02-05 18:30:22,597] WARN Connected to an old server; r-o mode will be unavailable (org.apache.zookeeper.ClientCnxnSocket)

go kafka

package main import ( "fmt" "time" "github.com/Shopify/sarama" ) func main() { config := sarama.NewConfig() config.Producer.RequiredAcks = sarama.WaitForAll config.Producer.Partitioner = sarama.NewRandomPartitioner config.Producer.Return.Successes = true client, err := sarama.NewSyncProducer([]string{"192.168.179.130:9092"}, config) if err != nil { fmt.Println("producer close, err:", err) return } defer client.Close() msg := &sarama.ProducerMessage{} msg.Topic = "nginx_log" msg.Value = sarama.StringEncoder("this is a good test, my message is good") pid, offset, err := client.SendMessage(msg) if err != nil { fmt.Println("send message failed,", err) return } fmt.Printf("pid:%v offset:%v\n", pid, offset) time.Sleep(10 * time.Millisecond) }