添加topic 和 agent config的监控

conf/etcd.go

@@ -45,18 +45,17 @@ func initConnect() *clientv3.Client {
 	return cli
 }
 
-// 获取当前所有的任务 (目前在初始化时使用)
-func GetAllConfFromEtcd() []EtcdValue {
+// 获取当前开启的Agent所有的任务 (目前在初始化时使用)
+func GetAllConfFromEtcd() ([]EtcdValue, error) {
 
+	configs := make([]EtcdValue, 0)
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	resp, err := cli.Get(ctx, configPath, clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
 	cancel()
 	if err != nil {
-		panic(fmt.Sprintf("get failed, err:%s \n", err))
+		return configs, err
 	}
 
-	configs := make([]EtcdValue, 0)
-
 	for _, etcdResult := range resp.Kvs {
 		// 根据系统中当前全部的节点名称, 确定节点状态
 		etcdKey := statusPath + string(etcdResult.Key[strings.LastIndex(string(etcdResult.Key), "/")+1:])
@@ -65,7 +64,7 @@ func GetAllConfFromEtcd() []EtcdValue {
 		resp, err := cli.Get(ctx, etcdKey)
 		cancel()
 		if err != nil {
-			panic(fmt.Sprintf("Get Etcd config failed, err:%s \n", err))
+			return configs, fmt.Errorf("get Etcd config failed, err:%s", err)
 		}
 
 		if len(resp.Kvs) != 0 {
@@ -77,34 +76,129 @@ func GetAllConfFromEtcd() []EtcdValue {
 
 		}
 
+	}
+
+	return configs, nil
+}
+
+func GetDelRevValueFromEtcd(key string, rev int64) (EtcdValue, error) {
+	var value EtcdValue
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	resp, err := cli.Get(ctx, key, clientv3.WithRev(rev))
+	cancel()
+	if err != nil {
+		log.Println(fmt.Sprintf("Get Etcd config failed, err:%s \n", err))
+	}
 
+	if len(resp.Kvs) == 0 {
+		return value, fmt.Errorf("config get error")
 	}
 
-	return configs
+	return resp.Kvs[0].Value, nil
 }
 
-// 加载所有的Topic主题配置信息
-func GetAllTopicFromEtcd() []EtcdValue {
+// 获取特定的配置
+func GetConfFromEtcd(name string) (EtcdValue, error) {
+
+	var value EtcdValue
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	resp, err := cli.Get(ctx, topicPath, clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
+	resp, err := cli.Get(ctx, statusPath+name)
+	cancel()
+	if err != nil {
+		log.Println(fmt.Sprintf("Get Etcd config failed, err:%s \n", err))
+	}
+
+	if len(resp.Kvs) == 0 {
+		return value, fmt.Errorf("status error")
+	}
+	status := string(resp.Kvs[0].Value)
+	if status != "1" {
+		return value, fmt.Errorf("status error")
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+	resp, err = cli.Get(ctx, configPath+name)
 	cancel()
+
 	if err != nil {
-		panic(fmt.Sprintf("get failed, err:%s \n", err))
+		return value, err
+	}
+
+	if len(resp.Kvs) == 0 {
+		return value, fmt.Errorf("config get error")
 	}
 
+	return resp.Kvs[0].Value, nil
+}
+
+// 加载所有的Topic主题配置信息
+func GetAllTopicFromEtcd() ([]EtcdValue, error) {
+
 	configs := make([]EtcdValue, 0)
 
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	resp, err := cli.Get(ctx, topicPath, clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
+	cancel()
+	if err != nil {
+		return configs, err
+	}
+
 	for _, etcdResult := range resp.Kvs {
 		configs = append(configs, etcdResult.Value)
 	}
 
-	return configs
+	return configs, nil
+}
+
+// 获取特定的Topic
+func GetTopicFromEtcd(name string) (EtcdValue, error) {
+
+	var value EtcdValue
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	resp, err := cli.Get(ctx, topicPath+name)
+	cancel()
+	if err != nil {
+		log.Println(fmt.Sprintf("Get Etcd config failed, err:%s \n", err))
+	}
+
+	if len(resp.Kvs) == 0 {
+		return value, fmt.Errorf("config get error")
+	}
+
+	return resp.Kvs[0].Value, nil
 }
 
-func WatchLogConfToEtcd() clientv3.WatchChan {
+func WatchLogTopicToEtcd() clientv3.WatchChan {
+
+	wch := cli.Watch(context.Background(), topicPath, clientv3.WithPrefix())
+
+	return wch
+}
+
+func WatchLogConfigToEtcd() clientv3.WatchChan {
 
 	wch := cli.Watch(context.Background(), configPath, clientv3.WithPrefix())
 
 	return wch
 }
+
+func CheckAgentActive(name string) bool {
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	resp, err := cli.Get(ctx, statusPath+name)
+	log.Println("looking for", statusPath+name)
+	cancel()
+	if err != nil {
+		log.Println(fmt.Sprintf("Get Etcd config failed, err:%s \n", err))
+	}
+
+	if len(resp.Kvs) == 0 {
+		return false
+	}
+	status := string(resp.Kvs[0].Value)
+	log.Println("it is"+status)
+	return status == "1"
+}

entity/matedata.go

@@ -41,7 +41,14 @@ func CloseMessageChan() {
 	close(messages)
 }
 
-func MatedateSender(ctx context.Context, esClient *elastic.Client) {
+func MatedateSender(ctx context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	// 初始化ES客户端
+	esClient, err := elastic.NewClient(elastic.SetSniff(false), elastic.SetURL(conf.APPConfig.Es.Address))
+	if err != nil {
+		panic(err)
+	}
 
 	wp := &ESWorkPool{
 		WorkerFunc: func(matedatas []*Matedata) bool {
@@ -53,8 +60,9 @@ func MatedateSender(ctx context.Context, esClient *elastic.Client) {
 			count := bulkRequest.NumberOfActions()
 			if count > 0 {
 				log.Printf("Send messages to Index: %d : \n", bulkRequest.NumberOfActions())
-				response, err := bulkRequest.Do(ctx)
-
+				timectx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+				response, err := bulkRequest.Do(timectx)
+				cancel()
 				if err != nil {
 					log.Println("Save Es Error:", err)
 					return false
@@ -73,11 +81,12 @@ func MatedateSender(ctx context.Context, esClient *elastic.Client) {
 			}
 			return true
 		},
-		MaxWorkerCount:        51,
+		MaxWorkerCount:        50,
 		MaxIdleWorkerDuration: 5 * time.Second,
 	}
-	wp.Start()
 
+	wp.Start()
+	defer wp.Stop()
 	var mateDatesItems []*Matedata
 
 	var mu sync.Mutex
@@ -106,8 +115,6 @@ func MatedateSender(ctx context.Context, esClient *elastic.Client) {
 			mu.Unlock()
 
 			wp.Serve(currentItems)
-
-			wp.Stop()
 			return
 		}
 	}

plugin/pipeline.go

@@ -9,7 +9,6 @@ import (
 	"github.com/y7ut/logtransfer/entity"
 )
 
-
 type Handler interface {
 	HandleFunc(*entity.Matedata) error
 	SetParams(string) error
@@ -24,6 +23,10 @@ type PipeLine struct {
 	pipe []*Handler
 }
 
+func (p *PipeLine) Length() int {
+	return len(p.pipe)
+}
+
 func (p *PipeLine) AppendPlugin(plugin Handler) {
 	p.pipe = append(p.pipe, &plugin)
 }

plugin/savees.go

@@ -2,6 +2,7 @@ package plugin
 
 import (
 	"fmt"
+	"log"
 
 	"github.com/y7ut/logtransfer/entity"
 )
@@ -10,7 +11,7 @@ import (
 type SaveES Plugin
 
 func (saveEs *SaveES) HandleFunc(m *entity.Matedata) error {
-	// log.Println("SaveES:")
+	log.Println("SaveES:")
 	m.Index = fmt.Sprintf("%s", (*saveEs.params)["index"])
 	m.Data["topic"] = m.Topic
 	m.Data["level"] = m.Level

source/customer.go

@@ -31,15 +31,14 @@ func (c Customer) Exit() {
 }
 
 // 结束信号监听
-func (c Customer) Listen() chan struct{} {
+func (c Customer) Listen() <-chan struct{} {
 	return c.done
 }
 
 // 初始化一个消费处理器
 func InitCustomer(topic *Topic) *Customer {
-	GroupID := topic.Name + "_group"
-	r := InitReader(topic.Name, GroupID)
-	log.Printf("Check Customer group of [%s] success!", GroupID)
+	r := InitReader(topic.Name)
+	log.Printf("Check Customer group of [%s] success!", topic.Name)
 	return &Customer{Topic: topic, Readers: r, done: make(chan struct{}), HandlePipeline: topic.PipeLine, Format: topic.Format}
 }
 
@@ -50,6 +49,12 @@ func RegisterManger(c *Customer) {
 	mu.Unlock()
 }
 
+func UnstallManger(topic string){
+	mu.Lock()
+	delete(CustomerManger, topic)
+	mu.Unlock()
+}
+
 // 根据topic快速获取消费处理器 目前用于关闭消费处理器
 func GetCustomer(topic string) (customer *Customer, ok bool) {
 	mu.Lock()
@@ -75,28 +80,36 @@ func ReadingMessage(ctx context.Context, c *Customer) {
 	readyToRead := make(chan *kafka.Reader)
 
 	go func(ctx context.Context, c *Customer) {
+		ctx, cancel := context.WithCancel(ctx)
+		defer cancel()
 		for {
 			select {
 			case <-c.Listen():
 				return
 			case <-ctx.Done():
 				c.Exit()
+				return
 			case reader := <-readyToRead:
-				defer reader.Close()
+				
 				go func(ctx context.Context, c *Customer) {
-					var errMessage strings.Builder
+					defer reader.Close()
 
+					var errMessage strings.Builder
 					var matedata entity.Matedata
+
 					for {
 						select {
-						case <-ctx.Done():
-							c.Exit()
+						case <- ctx.Done():
+							return
 						default:
 							m, err := reader.ReadMessage(ctx)
-
 							if err != nil {
-								// 退出
-								// c.Exit()
+								switch err {
+								case context.Canceled:
+									// 监听主上下文信号
+									log.Println("Closing Kafka Conection!")
+									return
+								default:
 									errMessage.Reset()
 									errMessage.WriteString("Reader Error")
 									errMessage.WriteString(err.Error())
@@ -105,6 +118,8 @@ func ReadingMessage(ctx context.Context, c *Customer) {
 									continue
 								}
 							
+							}
+
 							matedata, err = c.Format(string(reader.Config().Topic), string(m.Value))
 							if err != nil {
 								errMessage.Reset()
@@ -129,6 +144,5 @@ func ReadingMessage(ctx context.Context, c *Customer) {
 		log.Printf("Start Customer Group[%s][%d] success!", p.Config().GroupID, p.Config().Partition)
 
 		readyToRead <- p
-
 	}
 }

source/kafka.go

@@ -8,7 +8,9 @@ import (
 	"github.com/y7ut/logtransfer/conf"
 )
 
-func InitReader(topic string, groupId string) []*kafka.Reader {
+const GroupSuffix = "_group"
+
+func InitReader(topic string) []*kafka.Reader {
 	// // 先去创建一下这个分组
 	// make a writer that produces to topic-A, using the least-bytes distribution
 	var readers []*kafka.Reader
@@ -16,7 +18,7 @@ func InitReader(topic string, groupId string) []*kafka.Reader {
 		readers = append(readers, kafka.NewReader(kafka.ReaderConfig{
 			Brokers:   strings.Split(conf.APPConfig.Kafka.Address, ","),
 			Topic:     topic,
-			GroupID:   groupId,
+			GroupID:   topic+GroupSuffix,
 			// Partition: 0,
 			MinBytes:  10e3, // 10KB
 			MaxBytes:  10e6, // 10MB
@@ -33,15 +35,17 @@ func InitReader(topic string, groupId string) []*kafka.Reader {
 	return readers
 }
 
-func CreateCustomerGroup(topic string, groupId string) {
+func CreateCustomerGroup(topic string) error {
 	config := kafka.ConsumerGroupConfig{
-		ID:          groupId,
+		ID:          topic+GroupSuffix,
 		Brokers:     strings.Split(conf.APPConfig.Kafka.Address, ","),
 		Topics:      []string{topic},
 		StartOffset: kafka.LastOffset,
 	}
 	_, err := kafka.NewConsumerGroup(config)
+	log.Printf("Customer group [%s] created success!", topic+GroupSuffix)
 	if err != nil {
-		log.Println("create CustomerGroup error:", err)
+		return err
 	}
+	return nil
 }

source/topic.go

+package source
+
+import (
+	"log"
+
+	"github.com/y7ut/logtransfer/entity"
+	"github.com/y7ut/logtransfer/plugin"
+)
+
+func generateTopic(config TopicConfig) *Topic {
+
+	if config.PipelineConfig == nil {
+		log.Printf("get topic setting error:%s ", config.Label)
+	}
+
+	p := plugin.PipeLine{}
+
+	// log.Println("get config",  currentTopic.PipelineConfig)
+	for _, v := range config.PipelineConfig {
+		currentPlugin := plugin.RegistedPlugins[v.Name]
+		err := currentPlugin.SetParams(v.Params)
+		if err != nil {
+			log.Panicln("plugin encode params error:", err)
+		}
+		p.AppendPlugin(currentPlugin)
+	}
+	var formatMethod entity.Formater
+
+	switch config.Format {
+
+	case 1:
+		formatMethod = entity.DefaultJsonLog
+	case 2:
+		formatMethod = entity.FormatServiceWfLog
+	default:
+		formatMethod = entity.DefaultLog
+
+	}
+
+	return &Topic{Name: config.Name, Label: config.Label, PipeLine: &p, Format: formatMethod}
+}

transfer/kernel.go

@@ -6,10 +6,8 @@ import (
 	"log"
 	"os"
 	"os/signal"
-	"sync"
 	"syscall"
-
-	elastic "github.com/olivere/elastic/v7"
+	"time"
 
 	"github.com/y7ut/logtransfer/conf"
 	"github.com/y7ut/logtransfer/entity"
@@ -19,59 +17,40 @@ import (
 var (
 	Start = make(chan *source.Customer)
 	Close = make(chan string)
-	closeWg        sync.WaitGroup
 )
 
 // 核心启动
 func Run(confPath string) {
 	// 加载配置
 	conf.Init(confPath)
-	// 初始化ES客户端
-	esClient, err := elastic.NewClient(elastic.SetSniff(false), elastic.SetURL(conf.APPConfig.Es.Address))
-	if err != nil {
-		fmt.Println("connect es error", err)
-		panic(err)
-	}
 
 	// 做一个master的上下文
 	ctx, cancel := context.WithCancel(context.Background())
 
-	// 启动es消息发送器
-	// for i := 0; i < 20; i++ {
-	// 	go entity.MatedateSender(ctx, esClient)
-	// }
-
-	go entity.MatedateSender(ctx, esClient)
+	go entity.MatedateSender(ctx)
 
 	// 用于处理启动与关闭消费处理器的信号通知
-	go func() {
-		for {
-			select {
-			case customer := <-Start:
-				source.RegisterManger(customer)
-				go source.ReadingMessage(ctx, customer)
+	go CollectorRegister(ctx)
 
-			case closer := <-Close:
-				c, ok := source.GetCustomer(closer)
-				if !ok {
-					log.Printf(" Customer %s unstall Failed ", closer)
-
-				}
-				c.Exit()
-				closeWg.Done()
-			}
+	initTopic, err := source.ChooseTopic()
+	if err != nil {
+		panic(fmt.Sprintf("init topic fail: %s", err))
 	}
-	}()
 
-	// TODO: 动态的注册customer，目前仅支持初始化的时候来加载
-	for topic := range source.ChooseTopic() {
+	for topic := range initTopic {
 		currentCustomer := source.InitCustomer(topic)
 		Start <- currentCustomer
 	}
 
-	// TODO: 还要监听Topic的配置变更
-	// 目前是通过topic的name来注册所有的消费处理器
-	// 所以直接给对应的topic中的customer重启就可以杀了就可以了
+	go TopicWatcherHandle()
+
+	// 监听Agent Collector变更
+	go source.WatchConfigs()
+
+	// 还要监听Topic的配置变更
+	go source.WatchTopics()
+
+	// TODO: 监听Agent 启动状态变更
 
 	for sign := range sign() {
 		switch sign {
@@ -82,13 +61,13 @@ func Run(confPath string) {
 
 			for _, topic := range currentTopics {
 
-				closeWg.Add(1)
 				Close <- topic
 				log.Printf(" Customer %s unstalling...", topic)
-				closeWg.Wait()
+
 			}
-			entity.CloseMessageChan()
 			cancel()
+			entity.CloseMessageChan()
+			time.Sleep(1 * time.Second)
 			log.Printf(" Success unstall %d Transfer", len(currentTopics))
 			os.Exit(0)
 		}
@@ -97,6 +76,83 @@ func Run(confPath string) {
 
 }
 
+func CollectorRegister(ctx context.Context) {
+	for {
+		select {
+		case customer := <-Start:
+			source.RegisterManger(customer)
+			go source.ReadingMessage(ctx, customer)
+
+		case closer := <-Close:
+			c, ok := source.GetCustomer(closer)
+			if !ok {
+				log.Printf(" Customer %s unstall Failed ", closer)
+				break
+			}
+			source.UnstallManger(closer)
+			c.Exit()
+			// closeWg.Done()
+		}
+	}
+}
+
+// 监控topic的变动, 只处理更新, 若删除topic的话，不会触发配置的重新载入行为
+func TopicWatcherHandle() {
+	// restart
+	go func() {
+		for topic := range source.TopicChangeListener() {
+
+			var checkUsed bool
+
+			collectors, err := source.LoadCollectors()
+
+			if err != nil {
+				log.Println("Load Collector error:", err)
+				continue
+			}
+
+			// 检查是否使用
+			for _, item := range collectors {
+				if item.Topic == topic.Name {
+					checkUsed = true
+				}
+			}
+			if !checkUsed {
+				log.Println("Put topic but not used")
+				err := source.CreateCustomerGroup(topic.Name)
+				if err != nil {
+					log.Printf(" Create Topic Kafka customer group Failed : %s", err)
+					continue
+				}
+				continue
+			}
+
+			Close <- topic.Name
+			log.Printf(" Customer %s restart...", topic.Name)
+
+			currentCustomer := source.InitCustomer(topic)
+			Start <- currentCustomer
+		}
+	}()
+	// close
+	go func() {
+		for deleteTopic := range source.TopicDeleteListener() {
+			// closeWg.Add(1)
+
+			Close <- deleteTopic
+			log.Printf(" Customer %s deleting...", deleteTopic)
+			// closeWg.Wait()
+		}
+	}()
+	// start
+	go func() {
+		for topic := range source.TopicStartListener() {
+			currentCustomer := source.InitCustomer(topic)
+			Start <- currentCustomer
+		}
+	}()
+}
+
 func sign() <-chan os.Signal {
 	c := make(chan os.Signal, 2)
 	// 监听信号