前文中已经完成了文件的监控,kafka信息读写,今天主要完成配置文件的读写以及热更新。并且规划一下系统的整体结构,然后将之前的功能串起来形成一套完整的日志采集系统。
前情提要
上一节我们完成了如下目标
1 完成kafka消息读写
2 借助tailf实现文件监控,并模拟测试实时写文件以及文件备份时功能无误。
本节目标
1 编写系统结构,在主函数中加载配置
2 管理配置文件,实现热更新
实现文件管理,支持热更新
golang中vipper库提供了配置文件的读取和监控功能,我们可以监控配置文件从而实现热更新。
先规划下日志采集系统的目录结构
logcatchsys为项目根目录,其下logcatchsys文件夹中main.go为系统启动的主函数,该文件加载配置,根据配置启动协程,监控指定目录的日志文件,当配置更新时,main做热更新,如果路径从配置中删除,则中止对应的监控协程。如果有新的路径添加到配置文件,则启动协程监控,如果路径有修改,则中止原路径协程,启动新的协程监听修改后的路径。
logconfig为配置存放的路径,logconfig.go主要负责配置的管理,包括监控。
var onceLogConf sync.Once type ConfigData struct { ConfigKey string ConfigValue string ConfigCancel context.CancelFunc }
在logconfig.go中定义了once操作的变量onceLogConf,该变量保证监控配置的协程退出后只执行一次析构。
ConfigData结构体存储了配置文件中路径的信息,ConfigKey表示路径名,ConfigValue表示路径值,ConfigCancel存储上下文的CancelFunc,因为一个路径对应一个日志文件,监控日志文件就要开启协程,我是通过context管理监控日志的协程的。
在config.yaml中记录的路径信息如下:
configpath: logdir1: "../logdir1/log.txt" logdir2: "../logdir2/log.txt" logdir3: "../logdir3/log.txt" logdir5: "../logdir3/log.txt"
logdir1对应ConfigKey
../logdir1/log.txt对应ConfigValue
接下来在logconfig.go中我实现了配置文件的加载
func ReadConfig(v *viper.Viper) (interface{}, bool) { //设置读取的配置文件 v.SetConfigName("config") //添加读取的配置文件路径 _, filename, _, _ := runtime.Caller(0) fmt.Println(filename) fmt.Println(path.Dir(filename)) v.AddConfigPath(path.Dir(filename)) //设置配置文件类型 v.SetConfigType("yaml") if err := v.ReadInConfig(); err != nil { fmt.Printf("err:%s\n", err) return nil, false } configPaths := v.Get("configpath") if configPaths == nil { return nil, false } return configPaths, true }
以及配置文件的监听
func WatchConfig(ctx context.Context, v *viper.Viper, pathChan chan interface{}) { defer func() { onceLogConf.Do(func() { fmt.Println("watch config goroutine exit") if err := recover(); err != nil { fmt.Println("watch config goroutine panic ", err) } close(pathChan) }) }() //设置监听回调函数 v.OnConfigChange(func(e fsnotify.Event) { //fmt.Printf("config is change :%s \n", e.String()) configPaths := v.Get("configpath") if configPaths == nil { return } pathChan <- configPaths }) //开始监听 v.WatchConfig() //信道不会主动关闭,可以主动调用cancel关闭 <-ctx.Done() }
当配置文件config.yaml有变动时,OnConfigChange传入的匿名函数会触发,从而将configpath节点的value传入chan中,这样main函数可以从外部获取最新的配置文件。
ctx为上下文,当main函数执行上下文中止时,监控配置的协程会自动退出。
根据配置变动,实现热更新
在main.go中,定义了mainOnce控制主协程资源析构,并且通过ConstructMgr全局函数构造configMgr这样的map记录最新的配置信息。
var mainOnce sync.Once var configMgr map[string]*logconfig.ConfigData func ConstructMgr(configPaths interface{}) { configDatas := configPaths.(map[string]interface{}) for conkey, confval := range configDatas { configData := new(logconfig.ConfigData) configData.ConfigKey = conkey configData.ConfigValue = confval.(string) _, cancel := context.WithCancel(context.Background()) configData.ConfigCancel = cancel configMgr[conkey] = configData } }
接下来实现主函数
func main() { v := viper.New() configPaths, confres := logconfig.ReadConfig(v) if configPaths == nil || !confres { fmt.Println("read config failed") return } configMgr = make(map[string]*logconfig.ConfigData) ConstructMgr(configPaths) ctx, cancel := context.WithCancel(context.Background()) pathChan := make(chan interface{}) go logconfig.WatchConfig(ctx, v, pathChan) defer func() { mainOnce.Do(func() { if err := recover(); err != nil { fmt.Println("main goroutine panic ", err) // 这里的err其实就是panic传入的内容 } cancel() }) }()
在主函数中读取配置,并且将配置的路径信息存储在configMgr中。接着启动了一个协程用来监控配置文件,并且我实现了主协程的资源回收。
我们在main中继续添加接受监控协程的数据逻辑。
for { select { case pathData, ok := <-pathChan: if !ok { return } fmt.Println("main goroutine receive pathData") fmt.Println(pathData) pathDataNew := pathData.(map[string]interface{}) for oldkey, oldval := range configMgr { _, ok := pathDataNew[oldkey] if ok { continue } oldval.ConfigCancel() delete(configMgr, oldkey) } for conkey, conval := range pathDataNew { oldval, ok := configMgr[conkey] if !ok { configData := new(logconfig.ConfigData) configData.ConfigKey = conkey configData.ConfigValue = conval.(string) _, cancel := context.WithCancel(context.Background()) configData.ConfigCancel = cancel configMgr[conkey] = configData continue } if oldval.ConfigValue != conval.(string) { oldval.ConfigValue = conval.(string) oldval.ConfigCancel() _, cancel := context.WithCancel(context.Background()) oldval.ConfigCancel = cancel continue } } for mgrkey, mgrval := range configMgr { fmt.Println(mgrkey) fmt.Println(mgrval) } } } }
主协程接受数据后对比新旧数据,将旧的配置中被删除的路径剔除,增加和修改新的路径。
日志监控留给之后处理,这里打印下更新后的配置信息。
整体运行下main函数,然后我们手动修改config.yaml,将logdir4修改为logdir5,可以看到如下信息
证明我们的热更新处理逻辑没有问题。下一节基于现有的逻辑,添加日志文件的监控处理,启动多个协程管理日志文件。
源码下载
https://github.com/secondtonone1/golang-/tree/master/logcatchsys
我的公众号
原文地址:https://www.cnblogs.com/secondtonone1/p/11971309.html