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
}