printf重新实现put_Go实现海量日志收集系统(四)

2020.1.16

51Reboot 将在 2020.1.16日(今天) 21&＃xff1a;00 为您带来分享主题《大佬教你如何从 ES 初学者到 ES专家》

直播链接(提前报名)&＃xff1a;https://ke.qq.com/course/482014?taid&＃61;4309905192737502&tuin&＃61;31589b0e

51Reboot 课程信息最新一期 Golang 课程

开课时间&＃xff1a;2020.2.16

1、GO 并发

• goroutine

• channel

• 有缓存 chan 和无缓存 chan&＃xff0c;以及通信的一些细节

• 常见并发模式

• 锁与条件竞争

• 实战

 并发 Web 爬虫

2、Beego 实战

• Go Web 浅析

HTTP 协议 

net/http

• Gin/Beego 框架对比介绍

• 请求校验包的使用

• 中间件 middleware ORM

• 日志模块

3、堡垒机权限管理系统

• 架构设计

• HTTP 协议 加密传输

• 分布式 Agent/Server 开发

• Beego 可视化 Web 界面统管

• 思考&＃xff1a;如何对接服务树系统

4、分布式监控

• 架构设计

• GRPC 传输

• Etcd 服务注册发现

• Influxdb 存储

• Granfan 展示

运维自动化进阶课程

实战项目一&＃xff1a;SQL 自动化上线平台

• 手动 VS 自动的现状对比

• Mysql、Inception、SQLAdvisor 讲解

• 用户权限设计、执行流程梳理、平台登录双因子安全认证

• 敏感配置加解密实现、人员/数据库配置设计&＃xff0c;整体代码实现

实战项目二&＃xff1a;公有云管理平台

• 通过 API 对[Aws、阿里云、腾讯云、青云、百度云]管理

• 公有云账单、比价、资源管理

实战项目三&＃xff1a;任务管理系统

• Ansible 简介及常用场景分析

• Ansible API 二次开发入门

• 基于 Ansible Playbook API 快速实现任务管理系统

实战项目四&＃xff1a;工单系统

• 工作中流程规范的设计思路

• 典型工单系统的实现原理-状态机

实战项目五&＃xff1a;代码管理系统及发布平台

• 基于 Gitlab &＃43; Jenkins &＃43; DevOps 平台实现 CI/CD 的设计思路

• 结合运维平台流程规范实现持续集成与交付

• 结合 ELK 的日志分析平台&＃xff0c;实现代码上线运维无人值守

docker&＃43;k8s 课程

第一阶段&＃xff1a;Docker 基础与进阶

第二阶段&＃xff1a;Kubernetes 基础

第三阶段&＃xff1a;Pod 与生命周期管理

• Kubernetes 集群资源管理与调度管理

• Kubernetes 控制器和常用资源对象

第四阶段&＃xff1a;Kubernetes 服务发现

• 持久化存储

第五阶段&＃xff1a;Helm 包管理工具

第六阶段&＃xff1a;Kubernetes 集群网络

• Kubernetes 集群网络常用方案比较及选型建议

• Flannel 网络组件详解

• Flannel 网络组件配置及应用

• Flannel 生产环境应用经验

• Calico 网络组件详解

• Calico 网络组件配置及应用

• ......

第七阶段&＃xff1a;Kubernetes 集群监控

• Prometheus 介绍

• 部署 Prometheus

• 监控 Kubernetes 集群及应用

• NodeExporter 的安装使用

• Prometheus的自动发现

• Kubernetes常用资源对象监控

• Grafana的安装与使用

• Grafana的插件与监控

• .......

第八阶段&＃xff1a;日志收集

• 日志收集架构

• Elasticsearch 集群

• Kibana 可视化组件

• Fluentd 采集组件

• 生产环境采集日志方案详解

第九、十阶段&＃xff1a;DevOps

• 动态 Jenkins Slave

• Jenkins Pipeline

• Jenkins Blue Ocean

• Harbor 详解

• Gitlab 安装与使用

• Gitlab CI Runner

• Gitlab CI 示例

• Kubernetes 开源管理平台

• 完整 devops 项目实例

END

来源

https://www.cnblogs.com/zhaof/p/8948516.html

接Go 实现海量日志收集系统(一)Go实现海量日志收集系统(二)Go 实现海量日志收集系统(三)

到这一步&＃xff0c;我的收集系统就已经完成很大一部分工作&＃xff0c;我们重新看一下我们之前画的图&＃xff1a;

我们已经完成前面的部分&＃xff0c;剩下是要完成后半部分&＃xff0c;将kafka中的数据扔到ElasticSearch&＃xff0c;并且最终通过 kibana 展现出来

ElasticSearch

官网地址这里介绍了非常详细的安装方法&＃xff1a;https://www.elastic.co/downloads/elasticsearch但是其实这里是需要配置一些东西的&＃xff0c;要不然直接启动是会悲剧的&＃xff0c;在网上找了一个地址&＃xff0c;如果出现类似的错误直接处理就行&＃xff0c;我自己已经验证了&＃xff1a;https://blog.csdn.net/liangzhao_jay/article/details/56840941如下图所示就表示已经安装完成&＃xff1a;

通过 go 写一个简单的调用 ElasticSearch 的例子&＃xff1a;

package mainimport ( "fmt" elastic "gopkg.in/olivere/elastic.v2")type Tweet struct{ User string Message string}func main(){ client,err :&＃61; elastic.NewClient(elastic.SetSniff(false),elastic.SetURL("http://192.168.0.118:9200/")) if err !&＃61; nil{ fmt.Println("connect es error",err) return } fmt.Println("conn es succ") tweet :&＃61; Tweet{User:"olivere name",Message:"Take Five"} _, err &＃61; client.Index().Index("twitter").Type("tweet").Id("1").BodyJson(tweet).Do() if err !&＃61; nil { panic(err) return } fmt.Println("insert succ")}logtransfer

logtransfer 主要负责从 kafka队列中读取日志信息&＃xff0c;并且添加到 ElasticSearch   中看那一下 logtransfer 目录结构如下&＃xff1a;

├── conf│ └── app.conf├── es.go├── etcd.go├── ip.go├── kafka.go├── logs│ └── transfer.log└── main.go

conf : 存放配置文件es.go : 主要是连接 ElasticSearch 的部分以及用于将消息放到 ElasticSearch 中etcd.go : 主要用于做动态的配置更改&＃xff0c;当我们需要将 kafka 中的哪些 topic 日志内容扔到 ElasticSearch 中ip.go : 用于获取当前服务器的 ip 地址kafka.go : 主要是 kafka 的处理逻辑&＃xff0c;包括连接 kafka 以及从 kafka 中读日志内容main.go : 代码的入口函数

整体大代码框架&＃xff0c;通过如图展示&＃xff1a;

和之前的 logagent 中的代码有很多启示是可以复用的或者稍作更改&＃xff0c;就可以了&＃xff0c;其中 es 之心的&＃xff0c;主要是连接 ElasticSearch 并将日志内容放进去 es.go 的代码内容为&＃xff1a;

package mainimport ( "gopkg.in/olivere/elastic.v2" "github.com/astaxie/beego/logs" "sync" "encoding/json")var waitGroup sync.WaitGroupvar client *elastic.Clientfunc initEs(addr string,) (err error){ client,err &＃61; elastic.NewClient(elastic.SetSniff(false),elastic.SetURL(addr)) if err !&＃61; nil{ logs.Error("connect to es error:%v",err) return } logs.Debug("conn to es success") return}func reloadKafka(topicArray []string) { for _, topic :&＃61; range topicArray{ kafkaMgr.AddTopic(topic) }}func reload(){ //GetLogConf() 从channel中获topic信息&＃xff0c;而这部分信息是从etcd放进去的 for conf :&＃61; range GetLogConf(){ var topicArray []string err :&＃61; json.Unmarshal([]byte(conf),&topicArray) if err !&＃61; nil { logs.Error("unmarshal failed,err:%v conf:%v",err,conf) continue } reloadKafka(topicArray) }}func Run(esThreadNum int) (err error) { go reload() for i:&＃61;0;i waitGroup.Add(1) go sendToEs() } waitGroup.Wait() return}type EsMessage struct { Message string}func sendToEs(){ // 从msgChan中读取日志内容并扔到elasticsearch中 for msg:&＃61; range GetMessage() { var esMsg EsMessage esMsg.Message &＃61; msg.line _,err :&＃61; client.Index().Index(msg.topic).Type(msg.topic).BodyJson(esMsg).Do() if err !&＃61; nil { logs.Error("send to es failed,err:%v",err) continue } logs.Debug("send to es success") } waitGroup.Done()}

最终我将 logagnet 以及logtransfer 部署到虚拟机上进行测试的效果是&＃xff1a;

这样当我再次查日志的时候就可以不用登陆每台服务器去查日志&＃xff0c;只需要通过页面根据关键字迅速看到相关日志&＃xff0c;当然目前实现的功能还是有点粗糙&＃xff0c;etcd 的更改程序&＃xff0c;是自己写的发送程序&＃xff0c;其实更好的解决方法是通过页面&＃xff0c;让用户点来点去&＃xff0c;来控制自己要收集哪些日志&＃xff0c;以及自己要将哪些 topic 的日志从 kafka 中放到 ElasticSearch (本人是做后端开发&＃xff0c;不擅长前端的开发&＃xff0c;不过后面可以试着写个页面试试&＃xff0c;估计会很丑哈哈)。

同时这里关于各个部分的安装并没有做过多的介绍&＃xff0c;以及维护&＃xff0c;当然我们的目标是是通过这些开源的软件以及包来实现我们想要的功能&＃xff0c;后期的维护&＃xff0c;肯定需要对各个组件部分都进行深入了解。

这里附赠一下那个 etcd 客户端代码&＃xff1a;

package mainimport ( "github.com/coreos/etcd/clientv3" "time" "fmt" "golang.org/x/net/context")var logconf &＃61; &＃96;[ { "topic":"eslservice_log", "log_path":"/opt/pbx/log/eslservice.log", "service":"eslservice", "send_rate":50000 }]&＃96;var test111 &＃61; &＃96;[ { "topic":"test_log", "log_path":"D:/a.log", "service":"test", "send_rate":50000 }]&＃96;var transconf &＃61; &＃96;[ "eslservice_log"]&＃96;func main() { cli, err :&＃61; clientv3.New(clientv3.Config{ Endpoints:[]string{"192.168.90.78:2371"}, DialTimeout:5*time.Second, }) if err !&＃61; nil { fmt.Println("connect failed,err:",err) return } fmt.Println("connect success") defer cli.Close() ctx,cancel :&＃61; context.WithTimeout(context.Background(),time.Second) //_,err &＃61; cli.Put(ctx,"/logagent/192.168.90.11/log_config",logconf) //_,err &＃61; cli.Put(ctx,"/logagent/192.168.90.61/log_config",test111) _, err &＃61; cli.Put(ctx,"/logtransfer/192.168.90.11/log_config",transconf) cancel() if err !&＃61; nil { fmt.Println("put failed ,err:",err) return } ctx,cancel &＃61; context.WithTimeout(context.Background(),time.Second) resp,err :&＃61; cli.Get(ctx,"/logtransfer/",clientv3.WithPrefix()) cancel() if err !&＃61; nil { fmt.Println("get failed,err:",err) return } for _,ev:&＃61;range resp.Kvs{ fmt.Printf("%s:%s\n",ev.Key,ev.Value) }}

到目前为止基本的功能都已经实现了&＃xff0c;当然了现在的代码结构还有的糙&＃xff0c;后面会进行优化&＃xff01;整个项目中的代码&＃xff1a;logagent 代码地址&＃xff1a;https://github.com/pythonsite/logagentlogtransfer 代码地址&＃xff1a;https://github.com/pythonsite/logtransfer

关注我们 &＃xff0c;一起成长吧&＃xff01;