go语言并发实战——日志收集系统(十一)基于etcd来监视配置文件的变化

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: go语言并发实战——日志收集系统(十一)基于etcd来监视配置文件的变化

前言

在我们实际生产中,我们常常因为新的项目或者新的功能进而要对配置文件进行修改,但是在生产环境下我们不是每次配置文件发生变化都重启一次系统,这无疑是不切实际的,所以我们需要对配置文件进行实时监控,而今天我们所要展示的也就是如何基于etcd来监控配置文件的变化。

etcd对配置项监控的流程

需求分析

首先我们来看我们日志收集服务的主要工作流程:

func main() {
  //读取配置文件,获取配置信息
  filename := "G:\\goproject\\-goroutine-\\log-agent\\conf\\config.ini"
  ConfigObj := new(Config)
  err := ini.MapTo(ConfigObj, filename)
  if err != nil {
    logrus.Error("%s Load failed,err:", filename, err)
  }
  //初始化Kafka
  err = Kafka.InitKafka(ConfigObj.Kafakaddress.Addr, ConfigObj.Kafakaddress.MessageSize)
  if err != nil {
    logrus.Error("InitKafka failed, err:%v", err)
    return
  }
  logrus.Infof("InitKafka success")
  //初始化etcd
  err = etcd.Init(ConfigObj.Etcdaddress.Addr)
  if err != nil {
    logrus.Error("InitEtcd failed, err:%v", err)
    return
  }
  logrus.Infof("InitEtcd success")
  //拉取要收集日志文件的配置项
  err, collectEntryList := etcd.GetConf(ConfigObj.Etcdaddress.Key)
  if err != nil {
    logrus.Error("GetConf failed, err:%v", err)
    return
  }
  fmt.Println(collectEntryList)
  //初始化tail
  err = tailFile.InitTail(collectEntryList)
  if err != nil {
    logrus.Error("InitTail failed, err:%v", err)
    return
  }
  logrus.Infof("InitTail success")
  run()
}

在上述主要工作逻辑的基础上,现在我们需要etcd来实现对配置文件的实时监控,而这就需要我们在后态去运行一个监控程序来实时监控查看需要见监控的配置文件是否变化。并且将变化发送到tailFile模块中

实现Watch监控

所以这里我们对main.go进行一点简单的修改,添加一个后台程序 go etcd.WatchConf(ConfigObj.Etcdaddress.Key):

package main
import (
  "fmt"
  "github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
  "github.com/go-ini/ini"
  "log-agent/Kafka"
  "log-agent/etcd"
  "log-agent/tailFile"
)
type Config struct {
  Kafakaddress Kafkaddress `ini:"kafka"`
  LogFilePath  LogFilePath `ini:"collect"`
  Etcdaddress  EtcdAddress `ini:"etcd"`
}
type Kafkaddress struct {
  Addr        []string `ini:"address"`
  Topic       string   `ini:"topic"`
  MessageSize int64    `ini:"chan_size"`
}
type LogFilePath struct {
  Path string `ini:"logfile_path"`
}
type EtcdAddress struct {
  Addr []string `ini:"address"`
  Key  string   `ini:"collect_key"`
}
func run() {
  select {}
}
func main() {
  //读取配置文件,获取配置信息
  filename := "G:\\goproject\\-goroutine-\\log-agent\\conf\\config.ini"
  ConfigObj := new(Config)
  err := ini.MapTo(ConfigObj, filename)
  if err != nil {
    logrus.Error("%s Load failed,err:", filename, err)
  }
  //初始化Kafka
  err = Kafka.InitKafka(ConfigObj.Kafakaddress.Addr, ConfigObj.Kafakaddress.MessageSize)
  if err != nil {
    logrus.Error("InitKafka failed, err:%v", err)
    return
  }
  logrus.Infof("InitKafka success")
  //初始化etcd
  err = etcd.Init(ConfigObj.Etcdaddress.Addr)
  if err != nil {
    logrus.Error("InitEtcd failed, err:%v", err)
    return
  }
  logrus.Infof("InitEtcd success")
  //拉取要收集日志文件的配置项
  err, collectEntryList := etcd.GetConf(ConfigObj.Etcdaddress.Key)
  if err != nil {
    logrus.Error("GetConf failed, err:%v", err)
    return
  }
  fmt.Println(collectEntryList)
  go etcd.WatchConf(ConfigObj.Etcdaddress.Key)
  //初始化tail
  err = tailFile.InitTail(collectEntryList)
  if err != nil {
    logrus.Error("InitTail failed, err:%v", err)
    return
  }
  logrus.Infof("InitTail success")
  run()
}

我们在来看这个函数的具体逻辑:

func WatchConf(key string) {
  rch := client.Watch(context.Background(), key)
  var newConf []common.CollectEntry
  for wresp := range rch {
    logrus.Infof("get new conf fromn etcd")
    for _, ev := range wresp.Events {
      fmt.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
      err := json.Unmarshal(ev.Kv.Value, &newConf)
      if err != nil {
        logrus.Error("json unmarshal failed,err:%v", err)
        continue
      }
      tailFile.SendNewConf(newConf)
    }
  }
}

与之前有关etcd的文章中的操作例子不同,这里我们并没有定义上下文,主要是因为这里我们不确定什么时候终止这个程序,所以不使用上下文了。

发送新配置到tailFile中

在上面我们已经完成etcd的监控,现在我们需要把新的配置消息发送到tailFile,这里我们第一反应是写一个死循环一直独缺,但是这样其实不大方便,毕竟储蓄一直运行会占掉大量不必要消耗的资源,这里我们可以让双方使用管道来进行通信,平时管道处于阻塞状态,只有监测到新配置才会进行通信,这样会使资源得到最大化的利用,我们来看一看具体的代码实现:

  • 首先我们来定义一下用于通信的管道
var (
  confchan chan []common.CollectEntry
)
  • 然后我们要对管道进行初始化,并且读取管道中新的配置信息:
confchan = make(chan []common.CollectEntry)
  newConf := <-confchan
  logrus.Infof("get newconf from etcd", newConf)

最后,由于我们这里管道只用于etcd模块与tailFile模块之间的通信,所以这里我们就不暴露管道,而是选择暴露函数:

func SendNewConf(newConf []common.CollectEntry) {
  confchan <- newConf
}

结语

最后附上上述变化模块的代码:

  • main.go
package main
import (
  "fmt"
  "github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
  "github.com/go-ini/ini"
  "log-agent/Kafka"
  "log-agent/etcd"
  "log-agent/tailFile"
)
type Config struct {
  Kafakaddress Kafkaddress `ini:"kafka"`
  LogFilePath  LogFilePath `ini:"collect"`
  Etcdaddress  EtcdAddress `ini:"etcd"`
}
type Kafkaddress struct {
  Addr        []string `ini:"address"`
  Topic       string   `ini:"topic"`
  MessageSize int64    `ini:"chan_size"`
}
type LogFilePath struct {
  Path string `ini:"logfile_path"`
}
type EtcdAddress struct {
  Addr []string `ini:"address"`
  Key  string   `ini:"collect_key"`
}
func run() {
  select {}
}
func main() {
  //读取配置文件,获取配置信息
  filename := "G:\\goproject\\-goroutine-\\log-agent\\conf\\config.ini"
  ConfigObj := new(Config)
  err := ini.MapTo(ConfigObj, filename)
  if err != nil {
    logrus.Error("%s Load failed,err:", filename, err)
  }
  //初始化Kafka
  err = Kafka.InitKafka(ConfigObj.Kafakaddress.Addr, ConfigObj.Kafakaddress.MessageSize)
  if err != nil {
    logrus.Error("InitKafka failed, err:%v", err)
    return
  }
  logrus.Infof("InitKafka success")
  //初始化etcd
  err = etcd.Init(ConfigObj.Etcdaddress.Addr)
  if err != nil {
    logrus.Error("InitEtcd failed, err:%v", err)
    return
  }
  logrus.Infof("InitEtcd success")
  //拉取要收集日志文件的配置项
  err, collectEntryList := etcd.GetConf(ConfigObj.Etcdaddress.Key)
  if err != nil {
    logrus.Error("GetConf failed, err:%v", err)
    return
  }
  fmt.Println(collectEntryList)
  go etcd.WatchConf(ConfigObj.Etcdaddress.Key)
  //初始化tail
  err = tailFile.InitTail(collectEntryList)
  if err != nil {
    logrus.Error("InitTail failed, err:%v", err)
    return
  }
  logrus.Infof("InitTail success")
  run()
}
  • etcd.go
package etcd
import (
  "encoding/json"
  "fmt"
  "github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
  clientv3 "go.etcd.io/etcd/client/v3"
  "golang.org/x/net/context"
  "log-agent/common"
  "log-agent/tailFile"
  "time"
)
var client *clientv3.Client
func Init(address []string) (err error) {
  client, err = clientv3.New(clientv3.Config{
    Endpoints:   address,
    DialTimeout: 5 * time.Second,
  })
  if err != nil {
    logrus.Error("etcd client connect failed,err:%v", err)
    return
  }
  return
}
func GetConf(key string) (err error, collectEntryList []common.CollectEntry) {
  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  response, err := client.Get(ctx, key)
  cancel()
  if err != nil {
    logrus.Error("get conf from etcd failed,err:%v", err)
    return
  }
  if len(response.Kvs) == 0 {
    logrus.Warningf("get len:0 conf from etcd failed,err:%v", err)
    return
  }
  fmt.Println(response.Kvs[0].Value)                             //此时还是json字符串
  err = json.Unmarshal(response.Kvs[0].Value, &collectEntryList) //把值反序列化到collectEntryList
  if err != nil {
    logrus.Error("json unmarshal failed,err:%v", err)
    return
  }
  return
}
func WatchConf(key string) {
  rch := client.Watch(context.Background(), key)
  var newConf []common.CollectEntry
  for wresp := range rch {
    logrus.Infof("get new conf fromn etcd")
    for _, ev := range wresp.Events {
      fmt.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
      err := json.Unmarshal(ev.Kv.Value, &newConf)
      if err != nil {
        logrus.Error("json unmarshal failed,err:%v", err)
        continue
      }
      tailFile.SendNewConf(newConf)
    }
  }
  • tailFile.go
package tailFile
import (
  "github.com/Shopify/sarama"
  "github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
  "github.com/hpcloud/tail"
  "log-agent/Kafka"
  "log-agent/common"
  "strings"
  "time"
)
type tailTask struct {
  path    string
  topic   string
  TailObj *tail.Tail
}
var (
  confchan chan []common.CollectEntry
)
func NewTailTask(path, topic string) (tt *tailTask) {
  tt = &tailTask{
    path:  path,
    topic: topic,
  }
  return tt
}
func (task *tailTask) Init() (err error) {
  config := tail.Config{
    Follow:    true,
    ReOpen:    true,
    MustExist: true,
    Poll:      true,
    Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
  }
  task.TailObj, err = tail.TailFile(task.path, config)
  if err != nil {
    logrus.Error("tail create tailObj for path:%s,err:%v", task.path, err)
    return
  }
  return
}
func InitTail(collectEntryList []common.CollectEntry) (err error) {
  for _, entry := range collectEntryList {
    tt := NewTailTask(entry.Path, entry.Topic)
    err = tt.Init()
    if err != nil {
      logrus.Error("tail create tailObj for path:%s,err:%v", entry.Path, err)
      continue
    }
    go tt.run()
  }
  //初始化新配置的管道
  confchan = make(chan []common.CollectEntry)
  newConf := <-confchan
  logrus.Infof("get newconf from etcd", newConf)
  return
}
func (t *tailTask) run() {
  for {
    line, ok := <-t.TailObj.Lines
    if !ok {
      logrus.Warn("tailFile.TailObj.Lines channel closed,path:%s\n", t.path)
      time.Sleep(2 * time.Second)
      continue
    }
    if len(strings.Trim(line.Text, "\r")) == 0 {
      continue
    }
    msg := &sarama.ProducerMessage{}
    msg.Topic = t.topic
    msg.Value = sarama.StringEncoder(line.Text)
    Kafka.MesChan(msg)
  }
}
func SendNewConf(newConf []common.CollectEntry) {
  confchan <- newConf
}
相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
28天前
|
Go
go的并发初体验、加锁、异步
go的并发初体验、加锁、异步
18 0
支付系统40------定时查单-订单未创建,支付宝登陆前在支付宝端创建还是没有创建,不知道,之所以打印警告日志,是因为创建的时候更容易看到它
支付系统40------定时查单-订单未创建,支付宝登陆前在支付宝端创建还是没有创建,不知道,之所以打印警告日志,是因为创建的时候更容易看到它
|
2天前
|
存储 Java
有没有什么办法测试日志内容大小对系统性能的影响
有没有什么办法测试日志内容大小对系统性能的影响
|
11天前
|
安全 Go
Go语言map并发安全,互斥锁和读写锁谁更优?
Go并发编程中,`sync.Mutex`提供独占访问,适合读写操作均衡或写操作频繁的场景;`sync.RWMutex`允许多个读取者并行,适用于读多写少的情况。明智选择锁可提升程序性能和稳定性。示例展示了如何在操作map时使用这两种锁。
20 0
|
11天前
|
安全 Go 开发者
Go语言map并发安全使用的正确姿势
在Go并发编程中,由于普通map不是线程安全的,多goroutine访问可能导致数据竞态。为保证安全,可使用`sync.Mutex`封装map或使用从Go 1.9开始提供的`sync.Map`。前者通过加锁手动同步,后者内置并发控制,适用于多goroutine共享。选择哪种取决于具体场景和性能需求。
13 0
|
11天前
|
存储 安全 Java
Go语言中的map为什么默认不是并发安全的?
Go语言的map默认不保证并发安全,以优化性能和简洁性。官方建议在需要时使用`sync.Mutex`保证安全。从Go 1.6起,并发读写map会导致程序崩溃,鼓励开发者显式处理并发问题。这样做的哲学是让代码更清晰,并避免不必要的性能开销。
16 0
|
18天前
|
XML Java 数据格式
支付系统----微信支付20---创建案例项目--集成Mybatis-plus的补充,target下只有接口的编译文件,xml文件了,添加日志的写法
支付系统----微信支付20---创建案例项目--集成Mybatis-plus的补充,target下只有接口的编译文件,xml文件了,添加日志的写法
|
20天前
|
SQL 数据库
支付系统32-------支付宝支付-----支付成功异步通知----更新订单状态记录支付日志
支付系统32-------支付宝支付-----支付成功异步通知----更新订单状态记录支付日志
|
23天前
|
JavaScript
vue : 无法加载文件 D:\module\npm_module\npm_modules\vue.ps1,因为在此系统上禁止运行脚本。有关详细信息,请参阅 https:/go.microsoft.c
vue : 无法加载文件 D:\module\npm_module\npm_modules\vue.ps1,因为在此系统上禁止运行脚本。有关详细信息,请参阅 https:/go.microsoft.c