实例3：动态管理、根据ip区分
当etcd发送变更的时候，会根据新配置来新增、删除文件监控，并实现通过部署机的ip来区分加载不同的etcd简直对
example.go
package main
import (
"fmt"
"sync"
"example.go/addrget"
"example.go/etcd"
"example.go/kafka"
"example.go/tailfile"
"github.com/sirupsen/logrus"
"gopkg.in/ini.v1"
)
type Kafka struct {
Address string `ini:"addr"`
Topic string `ini:"topic"`
Size int64 `ini:"chan_size"`
}
type Tsdb struct {
Address string `ini:"address"`
Configpath string `ini:"confpath"`
}
type Etcd struct {
Address []string `ini:"address"`
Collectkey string `ini:"collectkey"`
}
type Configure struct { //映射用结构
Kafka `ini:"kafka"`
Tsdb `ini:"tsdb"`
Etcd `ini:"etcd"`
}
func main() {
ip, err := addrget.GetIP() //获取本机ip
if err != nil {
logrus.Error("FAILED TO GET IP!!!!", err)
return
}
logrus.SetLevel(logrus.DebugLevel)
var configOBJ = new(Configure) //创建一个configure类型的指针并分配空间
cfg, err := ini.Load("./configure.ini")
if err != nil {
logrus.Error("failed to load configure:", err)
return
}
// err = ini.MapTo(configOBJ,cfg)
// err := ini.MapTo(configOBJ, "./configure.ini")
err = cfg.MapTo(configOBJ) //映射
if err != nil {
logrus.Error("failed to reflect:", err)
return
}
fmt.Printf("%#v\n", configOBJ)
err = kafka.Kafkainit([]string{configOBJ.Kafka.Address}, configOBJ.Kafka.Size) //类型转换，kafka初始化
if err != nil {
logrus.Error("kafka init failed:", err)
return
}
logrus.Debug("kafka init success")
err = etcd.Etcd_ini(configOBJ.Etcd.Address) //etcd初始化
if err != nil {
logrus.Error("failed to init etcd")
return
}
logrus.Info("etcd init success")
conf_key := fmt.Sprintf(configOBJ.Etcd.Collectkey, ip)
conf_list, err := etcd.Etcd_getvalue(conf_key) //通过etcd获取value
if err != nil {
logrus.Error("faild to get value from etcd ", err)
return
}
// fmt.Printf("list:%v", list)
fmt.Println(conf_list)
err = tailfile.Tailinit(conf_list) //tail初始化，根据etcd返回的json去加载对应要读取的日志文件
if err != nil {
logrus.WithFields(logrus.Fields{"filename": configOBJ.Configpath, "err": err}).Error("file loaded failed")
return
}
/**
// go tailfile.Tailgetnewconf()
**/
go send_to_kafka()
var wg sync.WaitGroup
wg.Add(1)
// go send_to_kafka()
go etcd.Etcd_watchdog(conf_key, &wg)
wg.Wait()
}
func send_to_kafka() { //给kafka发送数据
// defer wg.Done()
defer kafka.Kafka_client.Close()
for {
partition, offset, err := kafka.Kafka_client.SendMessage(kafka.Kafkaoutput())
if err != nil {
logrus.Error("kafka send failed")
} else {
logrus.WithFields(logrus.Fields{"partition": partition, "offset": offset}).Info("send message success")
}
}
}getaddr.go
package addrget
import (
"net"
"strings"
"github.com/sirupsen/logrus"
)
//获取本地ip
func GetIP() (string, error) {
conn, err := net.Dial("udp", "8.8.8.8:8888")
if err != nil {
logrus.Error("Fail to get IP")
return "CAN NOT GET IP", err
}
defer conn.Close()
ipaddr := conn.LocalAddr()
ip := strings.Split(ipaddr.String(), ":")[0]
return ip, err
}kafka.go
package kafka
import (
"github.com/Shopify/sarama"
"github.com/sirupsen/logrus"
)
var (
Kafka_client sarama.SyncProducer
send_chan chan *sarama.ProducerMessage
)
func Kafkainit(address []string, size int64) (err error) {
config := sarama.NewConfig() //配置
config.Producer.RequiredAcks = sarama.WaitForAll //ack应答模式为all
config.Producer.Partitioner = sarama.NewRandomPartitioner //partition的选择方式为随机
config.Producer.Return.Successes = true //Successes channel 返回值
send_chan = make(chan *sarama.ProducerMessage, size)
Kafka_client, err = sarama.NewSyncProducer(address, config) //使用参数创建一个SyncProducer
// defer kafka_client.Close()
if err != nil {
logrus.Error("broker connection has something wrong happened:", err)
return err
}
return err
}
func Kafkareceive(msg *sarama.ProducerMessage) {
send_chan <- msg
}
func Kafkaoutput() *sarama.ProducerMessage {
return <-send_chan
}
// func Send_kafka() {
// }tailread.go
package tailfile
import (
"context"
"fmt"
"example.go/kafka"
"github.com/Shopify/sarama"
"github.com/nxadm/tail"
"github.com/sirupsen/logrus"
)
var (
Tail_context *tail.Tail
// change_chan chan []global.Configure_list
//Context_listen chan *tail.Line
)
type File_reader struct {
Path string
Topic string
File_handle *tail.Tail
ctx context.Context
cancel context.CancelFunc
}
//监控对象配置初始化
func type_init(path string, topic string) *File_reader {
ctx, cancel := context.WithCancel(context.Background())
tobj := File_reader{
Path: path,
Topic: topic,
ctx: ctx,
cancel: cancel,
}
return &tobj
}
func (tobj *File_reader) run() { //从tail中接收读取到的信息并发送给kafka的channel
// defer wg.Done()
logrus.WithFields(logrus.Fields{"FIle": tobj.Path}).Info("Start to monitor file-----")
msg := &sarama.ProducerMessage{}
msg.Topic = tobj.Topic
for {
select {
case <-tobj.ctx.Done():
logrus.WithFields(logrus.Fields{"File": tobj.Path}).Info("STOP READ FILE:")
return
case line, ok := <-tobj.File_handle.Lines:
if len(line.Text) == 0 {
continue
}
if ok {
logrus.Info("get message")
msg.Value = sarama.StringEncoder(line.Text)
fmt.Println(msg.Value)
kafka.Kafkareceive(msg)
logrus.Info("send message")
} else {
logrus.Error("failed to get message")
continue
}
}
}
}
//监控对象初始化
func (tobj *File_reader) init() (err error) {
config := tail.Config{
ReOpen: true, //当出现轮转切割的时候，这两个参数会去追踪新的文件
Follow: true,
Location: &tail.SeekInfo{Offset: 0, Whence: 2}, //从哪开始
MustExist: false, //允许文件不存在
Poll: true, //轮询
}
tobj.File_handle, err = tail.TailFile(tobj.Path, config)
return err
}tailmanager.go
package tailfile
import (
"example.go/global"
"github.com/sirupsen/logrus"
)
//文件监控句柄管理对象
type tailobjmgr struct {
tailobj map[string]*File_reader
tailobjconf []global.Configure_list
chg_chan chan []global.Configure_list
}
var (
objmgr *tailobjmgr
)
//初始化函数
func Tailinit(flist []global.Configure_list) (err error) {
objmgr = &tailobjmgr{
tailobj: make(map[string]*File_reader),
tailobjconf: flist,
chg_chan: make(chan []global.Configure_list),
}
for _, item := range flist {
t_file := type_init(item.Path, item.Topic) //导入基本信息
err = t_file.init() //初始化
if err != nil {
logrus.WithFields(logrus.Fields{"File": item.Path, "topic": item.Topic}).Error("failed to init tailfile:", err)
continue
}
objmgr.tailobj[t_file.Path] = t_file //把创建的对象放进map里面登记
logrus.WithFields(logrus.Fields{"File": t_file.Path}).Info("Init tailfile successfully")
go t_file.run()
}
logrus.Info("All tailfile inited")
go objmgr.watch()
return err
}
//等待更新
func (mobj *tailobjmgr) watch() {
logrus.Info("tail's watchdog is waiting")
for {
listen := <-mobj.chg_chan //等待etcd给他发送新配置
logrus.WithFields(logrus.Fields{"newconf": listen}).Info("get new configure:")
for _, item := range listen {
if mobj.isExist(item) { //如果是存在的，则跳过
continue
}
t_file := type_init(item.Path, item.Topic) //不存在则初始化
err := t_file.init() //初始化
if err != nil {
logrus.WithFields(logrus.Fields{"File": item.Path, "topic": item.Topic}).Error("failed to init tailfile:", err)
continue
}
mobj.tailobj[t_file.Path] = t_file //把创建的对象放进map里面登记
logrus.WithFields(logrus.Fields{"File": t_file.Path}).Info("Init tailfile successfully")
go t_file.run()
}
//把存在于登记列表中，但是在新配置中不存在的任务给停掉
var flag bool
logrus.Info("start to check")
for key, item := range mobj.tailobj {
flag = false
for _, itx := range listen {
if key == itx.Path {
logrus.WithFields(logrus.Fields{"file": key}).Info("get file")
flag = true
}
}
if !flag {
logrus.WithFields(logrus.Fields{"file": key}).Info("File watch will stop")
item.cancel()
delete(mobj.tailobj, key) //一定要记得删除，不然被停止的文件下次加载会被当成已存在的而被跳过无法初始化
}
}
logrus.Info("check over")
}
}
func Sendnewconf(newconf []global.Configure_list) { //让etcd发送新配置过来
objmgr.chg_chan <- newconf
}
func (mobj *tailobjmgr) isExist(conf global.Configure_list) bool { //判断该项文件是否已在登记的目录里
_, ok := mobj.tailobj[conf.Path]
return ok
}etcd.go:注，暂不支持etcd del操作，会收不到信号
package etcd
import (
"context"
"encoding/json"
"fmt"
"sync"
"time"
"example.go/global"
"example.go/tailfile"
"github.com/coreos/etcd/clientv3"
"github.com/sirupsen/logrus"
)
var (
client *clientv3.Client
)
// type Configure_list struct {
// Path string `json:"path"`
// Topic string `json:"topic"`
// }
// type LogEntry struct {
// Path string `json:"path"`
// Topic string `json:"topic"`
// }
func Etcd_ini(address []string) (err error) {
client, err = clientv3.New(clientv3.Config{Endpoints: address, DialTimeout: 3 * time.Second})
if err != nil {
logrus.Error("Failed to connect etcd", err)
return err
}
return err
}
func Etcd_getvalue(key string) (list []global.Configure_list, err error) { // 通过键获取值,返回json格式
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
get, err := client.Get(ctx, key)
if err != nil {
logrus.WithFields(logrus.Fields{key: key}).Error("failed to get key value", err)
return
}
if len(get.Kvs) == 0 {
logrus.WithFields(logrus.Fields{key: key}).Error("empty value!", err)
return
}
stu := get.Kvs[0]
err = json.Unmarshal(stu.Value, &list)
if err != nil {
logrus.Error("failed to unmarshal ", err)
return
}
return list, err
}
//监控日志配置项是否发生变化
func Etcd_watchdog(key string, wg *sync.WaitGroup) {
defer wg.Done()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
watchdog := client.Watch(ctx, key) //WatchChan-->WatchResponse-->[]Events-->type/Kv-->key/value
var newlist []global.Configure_list
logrus.Info("watchdog stand by ")
for re := range watchdog {
logrus.Info("etcd watchdog catchs new message!")
for _, ev := range re.Events {
fmt.Printf("Operation type is:%s\nkey:%s\nvalue:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
//原本此处设计动作为del时会清空，但是实测channel接收不到del这个信号，暂弃,如需清空请传入"[{\"path\": \"\",\"topic\":\"\"}]"或随便传入一个不存在的地址
// if ev.Type == clientv3.EventTypeDelete { //storagepb.Event_EventType DELETE = 1
// logrus.Warn("RECEIVED DELETE OPERATION")
// tailfile.Sendnewconf(newlist)
// logrus.Warn("YOU DELETE ALL CONFIGURES")
// return
// }
err := json.Unmarshal(ev.Kv.Value, &newlist)
if err != nil {
logrus.Error("failed to unmarshal now conf!")
continue
}
logrus.Info("trying to send new configure")
tailfile.Sendnewconf(newlist)
logrus.Info("send new configure successfully")
}
}
}
// td@tddeMacBook-Air ~ % etcdctl put testk1 "[{\"path\": \"/Users/td/Documents/blog/prod.log\",\"topic\":\"prod\"},{\"path\": \"/Users/td/Documents/blog/readme.log\",\"topic\":\"readme\"}]"
// OK
// td@tddeMacBook-Air ~ % etcdctl get testk1
// testk1
// [{"path": "/Users/td/Documents/blog/prod.log","topic":"prod"},{"path": "/Users/td/Documents/blog/readme.log","topic":"readme"}]configure.ini
[kafka]
addr = "127.0.0.1:9092"
topic = "test1"
chan_size = 10000
[tsdb]
address = 127.0.0.1
confpath = "./readme.log"
[etcd]
address = "127.0.0.1:2379"
collectkey = "etcd_key_%s_testk1"
评论已关闭