printf 重新实现put_Go 实现海量日志收集系统(四)
51Reboot 将在 2020.1.16日(今天) 21:00 为您带来分享主题《大佬教你如何从 ES 初学者到 ES专家》
直播链接(提前报名):https://ke.qq.com/course/482014?taid=4309905192737502&tuin=31589b0e
51Reboot 课程信息最新一期 Golang 课程
开课时间:2020.2.16
1、GO 并发
goroutine
channel
有缓存 chan 和无缓存 chan,以及通信的一些细节
常见并发模式
锁与条件竞争
实战
并发 Web 爬虫
2、Beego 实战
Go Web 浅析
HTTP 协议
net/http
Gin/Beego 框架对比介绍
请求校验包的使用
中间件 middleware ORM
日志模块
3、堡垒机权限管理系统
架构设计
HTTP 协议 加密传输
分布式 Agent/Server 开发
Beego 可视化 Web 界面统管
思考:如何对接服务树系统
4、分布式监控
架构设计
GRPC 传输
Etcd 服务注册发现
Influxdb 存储
Granfan 展示
运维自动化进阶课程
实战项目一:SQL 自动化上线平台
手动 VS 自动的现状对比
Mysql、Inception、SQLAdvisor 讲解
用户权限设计、执行流程梳理、平台登录双因子安全认证
敏感配置加解密实现、人员/数据库配置设计,整体代码实现
实战项目二:公有云管理平台
通过 API 对[Aws、阿里云、腾讯云、青云、百度云]管理
公有云账单、比价、资源管理
实战项目三:任务管理系统
Ansible 简介及常用场景分析
Ansible API 二次开发入门
基于 Ansible Playbook API 快速实现任务管理系统
实战项目四:工单系统
工作中流程规范的设计思路
典型工单系统的实现原理-状态机
实战项目五:代码管理系统及发布平台
基于 Gitlab + Jenkins + DevOps 平台实现 CI/CD 的设计思路
结合运维平台流程规范实现持续集成与交付
结合 ELK 的日志分析平台,实现代码上线运维无人值守
docker+k8s 课程
第一阶段:Docker 基础与进阶
第二阶段:Kubernetes 基础
第三阶段:Pod 与生命周期管理
Kubernetes 集群资源管理与调度管理
Kubernetes 控制器和常用资源对象
第四阶段:Kubernetes 服务发现
持久化存储
第五阶段:Helm 包管理工具
第六阶段:Kubernetes 集群网络
Kubernetes 集群网络常用方案比较及选型建议
Flannel 网络组件详解
Flannel 网络组件配置及应用
Flannel 生产环境应用经验
Calico 网络组件详解
Calico 网络组件配置及应用
......
第七阶段:Kubernetes 集群监控
Prometheus 介绍
部署 Prometheus
监控 Kubernetes 集群及应用
NodeExporter 的安装使用
Prometheus的自动发现
Kubernetes常用资源对象监控
Grafana的安装与使用
Grafana的插件与监控
.......
第八阶段:日志收集
日志收集架构
Elasticsearch 集群
Kibana 可视化组件
Fluentd 采集组件
生产环境采集日志方案详解
第九、十阶段: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 实现海量日志收集系统(三)
到这一步,我的收集系统就已经完成很大一部分工作,我们重新看一下我们之前画的图:
我们已经完成前面的部分,剩下是要完成后半部分,将kafka中的数据扔到ElasticSearch,并且最终通过 kibana 展现出来
ElasticSearch官网地址这里介绍了非常详细的安装方法:https://www.elastic.co/downloads/elasticsearch但是其实这里是需要配置一些东西的,要不然直接启动是会悲剧的,在网上找了一个地址,如果出现类似的错误直接处理就行,我自己已经验证了:https://blog.csdn.net/liangzhao_jay/article/details/56840941如下图所示就表示已经安装完成:
通过 go 写一个简单的调用 ElasticSearch 的例子:
package mainimport ( "fmt" elastic "gopkg.in/olivere/elastic.v2")type Tweet struct{ User string Message string}func main(){ client,err := elastic.NewClient(elastic.SetSniff(false),elastic.SetURL("http://192.168.0.118:9200/")) if err != nil{ fmt.Println("connect es error",err) return } fmt.Println("conn es succ") tweet := Tweet{User:"olivere name",Message:"Take Five"} _, err = client.Index().Index("twitter").Type("tweet").Id("1").BodyJson(tweet).Do() if err != nil { panic(err) return } fmt.Println("insert succ")}
logtransfer
logtransfer 主要负责从 kafka队列中读取日志信息,并且添加到 ElasticSearch 中看那一下 logtransfer 目录结构如下:
├── conf│ └── app.conf├── es.go├── etcd.go├── ip.go├── kafka.go├── logs│ └── transfer.log└── main.go
conf : 存放配置文件es.go : 主要是连接 ElasticSearch 的部分以及用于将消息放到 ElasticSearch 中etcd.go : 主要用于做动态的配置更改,当我们需要将 kafka 中的哪些 topic 日志内容扔到 ElasticSearch 中ip.go : 用于获取当前服务器的 ip 地址kafka.go : 主要是 kafka 的处理逻辑,包括连接 kafka 以及从 kafka 中读日志内容main.go : 代码的入口函数
整体大代码框架,通过如图展示:
和之前的 logagent 中的代码有很多启示是可以复用的或者稍作更改,就可以了,其中 es 之心的,主要是连接 ElasticSearch 并将日志内容放进去 es.go 的代码内容为:
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 = elastic.NewClient(elastic.SetSniff(false),elastic.SetURL(addr)) if err != nil{ logs.Error("connect to es error:%v",err) return } logs.Debug("conn to es success") return}func reloadKafka(topicArray []string) { for _, topic := range topicArray{ kafkaMgr.AddTopic(topic) }}func reload(){ //GetLogConf() 从channel中获topic信息,而这部分信息是从etcd放进去的 for conf := range GetLogConf(){ var topicArray []string err := json.Unmarshal([]byte(conf),&topicArray) if err != nil { logs.Error("unmarshal failed,err:%v conf:%v",err,conf) continue } reloadKafka(topicArray) }}func Run(esThreadNum int) (err error) { go reload() for i:=0;i waitGroup.Add(1) go sendToEs() } waitGroup.Wait() return}type EsMessage struct { Message string}func sendToEs(){ // 从msgChan中读取日志内容并扔到elasticsearch中 for msg:= range GetMessage() { var esMsg EsMessage esMsg.Message = msg.line _,err := client.Index().Index(msg.topic).Type(msg.topic).BodyJson(esMsg).Do() if err != nil { logs.Error("send to es failed,err:%v",err) continue } logs.Debug("send to es success") } waitGroup.Done()}
最终我将 logagnet 以及logtransfer 部署到虚拟机上进行测试的效果是:
这样当我再次查日志的时候就可以不用登陆每台服务器去查日志,只需要通过页面根据关键字迅速看到相关日志,当然目前实现的功能还是有点粗糙,etcd 的更改程序,是自己写的发送程序,其实更好的解决方法是通过页面,让用户点来点去,来控制自己要收集哪些日志,以及自己要将哪些 topic 的日志从 kafka 中放到 ElasticSearch (本人是做后端开发,不擅长前端的开发,不过后面可以试着写个页面试试,估计会很丑哈哈)。
同时这里关于各个部分的安装并没有做过多的介绍,以及维护,当然我们的目标是是通过这些开源的软件以及包来实现我们想要的功能,后期的维护,肯定需要对各个组件部分都进行深入了解。
这里附赠一下那个 etcd 客户端代码:
package mainimport ( "github.com/coreos/etcd/clientv3" "time" "fmt" "golang.org/x/net/context")var logconf = `[ { "topic":"eslservice_log", "log_path":"/opt/pbx/log/eslservice.log", "service":"eslservice", "send_rate":50000 }]`var test111 = `[ { "topic":"test_log", "log_path":"D:/a.log", "service":"test", "send_rate":50000 }]`var transconf = `[ "eslservice_log"]`func main() { cli, err := clientv3.New(clientv3.Config{ Endpoints:[]string{"192.168.90.78:2371"}, DialTimeout:5*time.Second, }) if err != nil { fmt.Println("connect failed,err:",err) return } fmt.Println("connect success") defer cli.Close() ctx,cancel := context.WithTimeout(context.Background(),time.Second) //_,err = cli.Put(ctx,"/logagent/192.168.90.11/log_config",logconf) //_,err = cli.Put(ctx,"/logagent/192.168.90.61/log_config",test111) _, err = cli.Put(ctx,"/logtransfer/192.168.90.11/log_config",transconf) cancel() if err != nil { fmt.Println("put failed ,err:",err) return } ctx,cancel = context.WithTimeout(context.Background(),time.Second) resp,err := cli.Get(ctx,"/logtransfer/",clientv3.WithPrefix()) cancel() if err != nil { fmt.Println("get failed,err:",err) return } for _,ev:=range resp.Kvs{ fmt.Printf("%s:%s\n",ev.Key,ev.Value) }}
到目前为止基本的功能都已经实现了,当然了现在的代码结构还有的糙,后面会进行优化!整个项目中的代码:logagent 代码地址:https://github.com/pythonsite/logagentlogtransfer 代码地址:https://github.com/pythonsite/logtransfer
关注我们 ,一起成长吧!
printf 重新实现put_Go 实现海量日志收集系统(四)相关推荐
- python分布式日志收集系统_Go实现海量日志收集系统(一)
项目背景 每个系统都有日志,当系统出现问题时,需要通过日志解决问题 当系统机器比较少时,登陆到服务器上查看即可满足 当系统机器规模巨大,登陆到机器上查看几乎不现实 当然即使是机器规模不大,一个系统通常 ...
- Flume 海量日志收集利器
Flume 海量日志收集利器 关于日志收集 服务器日志收集 服务器日志是大数据系统中最主要的数据来源之一 服务器日志可能包含的信息 访问信息 系统信息 其他业务信息 基于服务器日志的应用 业务仪表盘: ...
- go语言日志收集系统
0.项目地址 完整项目的GitHub地址 https://github.com/taw19960426/learning-go-language/tree/main/go-log-collect 一. ...
- 系统调用日志收集系统
1,系统调用日志收集系统的意义. 系统调用是用户获取系统服务的唯一入口,因此对系统调用的安全调用直接关系到系统 的安全,假如有用户恶意不断地调用系统调用,将会导致系统负载增加,所以如果能收集到时谁 调 ...
- fluentd mysql_使用Fluentd + MongoDB构建实时日志收集系统
日志处理场景 日志量大 日志分散不易进行统一分析 难以添加有效监控 系统实现 Fluentd(td-agent) MongoDB Python Script(PyMongo module) Zabbi ...
- execve系统调用_张凯捷—系统调用分析(3) (基于最新Linux5.0版本系统调用日志收集系统)...
在上一篇文章<系统调用分析(2)>中介绍和分析了32位和64位的快速系统调用指令--sysenter/sysexit和syscall/sysret,以及内核对快速系统调用部分的相关代码,并 ...
- 基于Flume的美团日志收集系统(一)架构和设计
背景 美团的日志收集系统负责美团的所有业务日志的收集,并分别给Hadoop平台提供离线数据和Storm平台提供实时数据流.美团的日志收集系统基于Flume设计和搭建而成. <基于Flume的美团 ...
- 基于Flume的美团日志收集系统-----架构和设计
问题导读: 1.Flume-NG与Scribe对比,Flume-NG的优势在什么地方? 2.架构设计考虑需要考虑什么问题? 3.Agent死机该如何解决? 4.Collector死机是否会有影响? 5 ...
- Go语言学习之11 日志收集系统kafka库实战
本节主要内容: 1. 日志收集系统设计 2. 日志客户端开发 1. 项目背景 a. 每个系统都有日志,当系统出现问题时,需要通过日志解决问题 b. 当系统机器比较少时,登陆到服务器上查 ...
最新文章
- 【未完成】[Spark SQL_2] 在 IDEA 中编写 Spark SQL 程序
- 五分钟创建一个自己的NPM包
- Jquery EasyUI datagrid数据库分页
- [二叉树]二叉搜索树转换为双向链表(剑指Offer26)
- lxml处理xml时的字符编码问题
- SQL Server学习笔记
- Google Breakpad的使用
- Jmeter-Beanshell
- 对话 NuCypher:如何安全地存储、共享和管理链上私人数据
- 三只松鼠营收持续下滑:市值两天缩水28亿元,能否熬过漫漫转型路
- 帝国CMS Table '***.phome_ecms_news_data_' doesn't exist
- 阿里云ACP认证之云安全知识整理(考题占比 10%)
- 基于FPGA的数字时钟设计
- 【虚幻引擎UE】场景中物体闪烁抖动摩尔纹问题集合
- 百度地图 web 只显示地图
- PowerQuery操作分类3
- Flask——1.初识flask微框架
- 58、【backtrader股票策略】两资产的配对交易策略(pairs trading strategy)
- HTT超线程技术,CPU,core,package,die的概念
- bing每日壁纸客户端