perf: 优化

source/collector.go

@@ -14,19 +14,22 @@ import (
 	"github.com/y7ut/logtransfer/plugin"
 )
 
+// 收藏家, 收藏家的目的就是采集收集目标目录的数据, 并投递到对应 Topic 的 Kafka topic 中
 type Collector struct {
 	Style string `json:"style"`
 	Path  string `json:"path"`
 	Topic string `json:"topic"`
 }
 
+// 日志主题
 type Topic struct {
-	Name     string
-	Label    string
-	PipeLine *plugin.PipeLine
-	Format   entity.Formater
+	Name     string           // 主题的名字 对应Kafka的topic名称
+	Label    string           // 主题的简介
+	PipeLine *plugin.PipeLine // 主题所拥有的插件列表，以管道的形式存在
+	Format   entity.Formater  // 流入这个主题的数据，所要进行元数据的处理格式
 }
 
+// 日志主题在Etcd中所存储的数据结构
 type TopicConfig struct {
 	Format         int                     `json:"format"`
 	Label          string                  `json:"label"`
@@ -34,19 +37,14 @@ type TopicConfig struct {
 	PipelineConfig []PipeLinePluginsConfig `json:"piepline"`
 }
 
+// 日志主题中的插件配置, 在Etcd中所存储的数据结构
 type PipeLinePluginsConfig struct {
 	Label  string `json:"label"`
 	Name   string `json:"name"`
 	Params string `json:"params"`
 }
 
-var watchTopicChannel = make(chan *Topic)
-
-var startTopicChannel = make(chan *Topic)
-
-var deleteTopicChannel = make(chan string)
-
-// 加载所有的可用的collector
+// 获取所有的收藏家
 func LoadCollectors() ([]Collector, error) {
 	collectors := make([]Collector, 0)
 	configs, err := conf.GetAllConfFromEtcd()
@@ -69,7 +67,7 @@ func LoadCollectors() ([]Collector, error) {
 	return collectors, nil
 }
 
-// 加载Agent所有的collector
+// 根据收藏家的名称获取对应的收藏家
 func LoadCollector(name string) ([]Collector, error) {
 	var collectors []Collector
 	config, err := conf.GetConfFromEtcd(name)
@@ -85,7 +83,7 @@ func LoadCollector(name string) ([]Collector, error) {
 	return collectors, nil
 }
 
-// 收集所有需要监听的topic
+// 确定所有需要监听的日志主题
 func ChooseTopic() (map[*Topic]bool, error) {
 	// 收集全部的agent的collector信息
 	ableTopics := make(map[*Topic]bool)
@@ -132,42 +130,30 @@ func loadTopics() (map[string]*Topic, error) {
 	return topics, nil
 }
 
-func TopicChangeListener() <-chan *Topic {
-	return watchTopicChannel
-}
-
-func TopicDeleteListener() <-chan string {
-	return deleteTopicChannel
-}
-
-func TopicStartListener() <-chan *Topic {
-	return startTopicChannel
-}
-
+// 监听 Etcd 中 Topic 的配置变更
 func WatchTopics() {
 	for confResp := range conf.WatchLogTopicToEtcd() {
 
 		for _, event := range confResp.Events {
 			switch event.Type {
-			case mvccpb.PUT:
-				// 有PUT操作才进行通知
-				var newTopicCondfig TopicConfig
+			case mvccpb.PUT: // 有PUT操作才进行通知
+				var newVersionTopicConfig TopicConfig
+
 				if len(confResp.Events) == 0 {
 					continue
 				}
 				changedConf := confResp.Events[0].Kv.Value
 
-				err := json.Unmarshal(changedConf, &newTopicCondfig)
-
+				err := json.Unmarshal(changedConf, &newVersionTopicConfig)
 				if err != nil {
 					log.Println("Unmarshal New Topic Config Error:", err)
 				}
-				log.Println("load New Topic success!")
 
-				watchTopicChannel <- generateTopic(newTopicCondfig)
+				log.Printf("reload New Topic[%s] success!", confResp.Events[0].Kv.Key)
+				watchTopicChannel <- generateTopic(newVersionTopicConfig)
+
+			case mvccpb.DELETE: // 获取旧版本数据 来进行对比
 
-			case mvccpb.DELETE:
-				// 获取旧版本数据 来进行对比
 				// 要清空这个register 中全部这个topic的customer， 不过也应该没有了，应该都会被config watcher 给捕获到
 				oldTopic, err := getHistoryTopicWithEvent(confResp)
 				if err != nil {
@@ -175,7 +161,7 @@ func WatchTopics() {
 					continue
 				}
 
-				log.Println("some Topic remove", oldTopic.Name)
+				log.Println ("some Topic remove", oldTopic.Name)
 			}
 
 			time.Sleep(2 * time.Second)

source/topic.go

@@ -7,6 +7,27 @@ import (
 	"github.com/y7ut/logtransfer/plugin"
 )
 
+// 用于监听Topic改变的 channel
+var watchTopicChannel = make(chan *Topic)
+
+// 用于监听Topic启动的 channel
+var startTopicChannel = make(chan *Topic)
+
+// 用于监听Topic关闭的 channel
+var deleteTopicChannel = make(chan string)
+
+func TopicChangeListener() <-chan *Topic {
+	return watchTopicChannel
+}
+
+func TopicDeleteListener() <-chan string {
+	return deleteTopicChannel
+}
+
+func TopicStartListener() <-chan *Topic {
+	return startTopicChannel
+}
+
 func generateTopic(config TopicConfig) *Topic {
 
 	if config.PipelineConfig == nil {
@@ -19,7 +40,8 @@ func generateTopic(config TopicConfig) *Topic {
 		currentPlugin := plugin.LoadRegistedPlugins(v.Name)
 		err := currentPlugin.SetParams(v.Params)
 		if err != nil {
-			log.Panicln("plugin encode params error:", err)
+			log.Printf("plugin encode params error: %s", err)
+			continue
 		}
 		p.AppendPlugin(currentPlugin)
 	}