etcd collect conf

etcd/etcd.go

@@ -1,6 +1,8 @@
 package etcd
 
 import (
+	"context"
+	"encoding/json"
 	"fmt"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"time"
@@ -10,6 +12,11 @@ var (
 	cli *clientv3.Client
 )
 
+type LogEntry struct {
+	Path  string `json:"path"`
+	Topic string `json:"topic"`
+}
+
 //初始化etcd
 
 func Init(addr string, timeout time.Duration) (err error) {
@@ -25,3 +32,23 @@ func Init(addr string, timeout time.Duration) (err error) {
 	fmt.Println("connect etcd server success")
 	return
 }
+
+// 从etcd中根据key获取配置项
+
+func GetConf(key string) (logEntryConf []*LogEntry, err error) {
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	resp, err := cli.Get(ctx, key)
+	cancel()
+	if err != nil {
+		fmt.Println("get etcd config failed,error:", err)
+	}
+	for _, ev := range resp.Kvs {
+		err = json.Unmarshal(ev.Value, &logEntryConf)
+		if err != nil {
+			fmt.Println("parse etcd config failed,error:", err)
+			return
+		}
+	}
+	return
+}

main.go

@@ -46,6 +46,19 @@ func main() {
 	if err != nil {
 		fmt.Printf("init etcd failed,error:%v\n", err)
 	}
+
+	//1.从etcd中获取日志收集项的配置信息
+	logEntryConf, err := etcd.GetConf("/xxx/")
+	if err != nil {
+		fmt.Printf("get etcd failed,error:%v\n", err)
+		return
+	}
+	fmt.Printf("get conf from etcd success,logEntryConf:%#v\n", logEntryConf)
+	for k, v := range logEntryConf {
+		fmt.Printf("index:%v,value:%v\n", k, v)
+	}
+	//2.派一个哨兵去监视日志收集项的变化（有变化急事通知我的LogAgent，实现热加载）
+
 	//2.打开日志文件准备收集日志
 	//err = taillog.Init(cfg.TailLog.FileName)
 	//if err != nil {

put_etcd/etcd_putvalue.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"go.etcd.io/etcd/client/v3"
+	"time"
+)
+
+func main() {
+	cli, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{"43.143.245.135:2379"},
+		DialTimeout: 5 * time.Second,
+	})
+	if err != nil {
+		// handle error!
+		fmt.Println("connect etcd server failed,error:", err)
+	}
+	defer cli.Close()
+	// put
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	value := `[{"path":"c:/tmp/nginx.log","topic":"web_log"},{"path":"d:/xxx/redis.log","topic":"redis_log"}]`
+	_, err = cli.Put(ctx, "/xxx/", value)
+	cancel()
+	if err != nil {
+		fmt.Println("put to etcd failed,error:", err)
+	}
+}