日志收集系統系列（三）之LogAgent

一、什么是LogAhent

類似於在linux下通過tail的方法讀日志文件，將讀取的內容發給kafka，這里的tailf是可以動態變化的，當配置文件發生變化時，可以通知我們程序自動增加需要增加的配置文件。tailf去獲取相應的日志並發給kafka producer。主要包含kafka、tailf和configlog。LogAgent工作流程

• 讀日志--tailf第三方庫

• 在kafka寫日志 --sarama第三方庫

• 讀取配置文件 -- "gopkg.in/ini.v1"

二、tail

1. 下載

go get github.com/hpcloud/tail

2. 使用示例

package main

import (
	"fmt"
	"github.com/hpcloud/tail"
	"time"
)

func main() {
	fileName := "./my.log"
	config := tail.Config{
		ReOpen:    true,                                 // 重新打開
		Follow:    true,                                 // 是否跟隨
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, // 從文件的哪個地方開始讀
		MustExist: false,                                // 文件不存在不報錯
		Poll:      true,
	}
	tails, err := tail.TailFile(fileName, config)
	if err != nil {
		fmt.Println("tail file failed, err:", err)
		return
	}
	var (
		line *tail.Line
		ok   bool
	)
	for {
		line, ok = <-tails.Lines//遍歷chan，讀取日志內容
		if !ok {
			fmt.Printf("tail file close reopen, filename:%s\n", tails.Filename)
			time.Sleep(time.Second)
			continue
		}
		fmt.Println("line:", line.Text)
	}
}

• 使用說明

  1. 首先初始化配置結構體config

  2. 調用TailFile函數，並傳入文件路徑和config，返回有個tail的結構體，tail結構體的Lines字段封裝了拿到的信息

  3. 遍歷tail.Lnes字段，取出信息（注意這里要循環的取，因為tail可以實現實時監控）

• 運行

   

三、sarama

1. 下載

go get github.com/Shopify/sarama

2. 生產消費示例

package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

func main() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 發送完數據需要leader和follow都確認
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新選出一個partition
	config.Producer.Return.Successes = true                   // 成功交付的消息將在success channel返回

	// 構造一個消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = "web_log"
	msg.Value = sarama.StringEncoder("this is a test log")
	// 連接kafka
	client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	fmt.Println("kafka 連接成功！")
	defer client.Close()
	// 發送消息
	pid, offset, err := client.SendMessage(msg)
	if err != nil {
		fmt.Println("send msg failed, err:", err)
		return
	}
	fmt.Printf("pid:%v offset:%v\n", pid, offset)
	print("發送成功！")
}

• 運行

• kafka文件

3. 消費示例

package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

// kafka consumer

func main() {
	consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
	if err != nil {
		fmt.Printf("fail to start consumer, err:%v\n", err)
		return
	}
	partitionList, err := consumer.Partitions("web_log") // 根據topic取到所有的分區
	if err != nil {
		fmt.Printf("fail to get list of partition:err%v\n", err)
		return
	}
	fmt.Println("分區: ", partitionList)
	for partition := range partitionList { // 遍歷所有的分區
		// 針對每個分區創建一個對應的分區消費者
		pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
			return
		}
		defer pc.AsyncClose()
		// 異步從每個分區消費信息
		go func(sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%s Value:%s\n", msg.Partition, msg.Offset, msg.Key, msg.Value)
			}
		}(pc)
	}
	select {}
}

四、簡易版LogAgent

1. 項目結構

│  go.mod
│  go.sum
│  main.go
│  my.log
├─conf
│      config.go
│      config.ini
│
├─kafka
│      kafka.go
│
└─taillog
       taillog.go

2. 配置文件

[kafka]
address=127.0.0.1:9092
topic=web_log

[taillog]
filename=./my.log

　config/config.go

package conf

type Config struct {
	Kafka Kafka `ini:"kafka"`
	TailLog TailLog `ini:"taillog"`
}


type Kafka struct {
	Address string `ini:"address"`
	Topic string `ini:"topic"`
}

type TailLog struct {
	FileName string `ini:"filename"`
}

　kafka/kafka.go

package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
)
// 專門往kafka寫日志的模塊

var (
	client sarama.SyncProducer	// 聲明一個全局的連接kafka的生產者client
)

// init初始化client
func Init(addrs []string) (err error) {
	config := sarama.NewConfig()

	config.Producer.RequiredAcks = sarama.WaitForAll          // 發送完數據需要leader和follow都確認
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新選出⼀個partition
	config.Producer.Return.Successes = true                   // 成功交付的消息將在success channel返回

	// 連接kafka
	client, err = sarama.NewSyncProducer(addrs, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	return
}

func SendToKafka(topic, data string)  {
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Value = sarama.StringEncoder(data)
	// 發送到kafka
	pid, offset, err := client.SendMessage(msg)
	if err != nil{
		fmt.Println("sned mage failed, err:", err)
	}
	fmt.Printf("pid:%v offset:%v\n", pid, offset)
	fmt.Println("發送成功")
}

　taillog/taillog.go

package taillog

import (
	"fmt"
	"github.com/hpcloud/tail"
)

// 專門收集日志的模塊

var (
	tailObj *tail.Tail
	logChan chan string
)

func Init(filename string) (err error) {
	config := tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true}
	tailObj, err = tail.TailFile(filename, config)
	if err != nil {
		fmt.Println("tail file failed, err:", err)
		return
	}
	return
}

func ReadChan() <-chan *tail.Line {
	return tailObj.Lines
}

   main.go

package main

import (
	"fmt"
	"gopkg.in/ini.v1"
	"logagent/conf"
	"logagent/kafka"
	"logagent/taillog"
	"strings"
	"time"
)

var config  = new(conf.Config)

// logAgent 入口程序

func main() {
	// 0. 加載配置文件
	//cfg, err := ini.Load("./conf/config.ini")
	//address := cfg.Section("kafka").Key("address").String()
	//topic := cfg.Section("kafka").Key("topic").String()
	//path := cfg.Section("taillog").Key("path").String()
	err := ini.MapTo(config, "./conf/config.ini")
	if err != nil {
		fmt.Printf("Fail to read file: %v", err)
		return
	}
	fmt.Println(config)
	// 1. 初始化kafka連接
	err = kafka.Init(strings.Split(config.Kafka.Address, ";"))
	if err != nil {
		fmt.Println("init kafka failed, err:%v\n", err)
		return
	}
	fmt.Println("init kafka success.")
	// 2. 打開日志文件准備收集日志
	err = taillog.Init(config.TailLog.FileName)
	if err != nil {
		fmt.Printf("Init taillog failed,err:%v\n", err)
		return
	}
	fmt.Println("init taillog success.")
	run()
}

func run() {
	// 1. 讀取日志
	for {
		select {
		case line := <-taillog.ReadChan():
			// 2. 發送到kafka
			kafka.SendToKafka(config.Kafka.Topic, line.Text)
		default:
			time.Sleep(time.Second)
		}
	}
}

　kafka終端打開消費者程序，進入kafka安裝目錄　

bin\windows\kafka-console-consumer.bat --bootstrap-server=127.0.0.1:9092 --topic=web_log --from-beginning

五、etcd

etcd是使用Go語言開發的一個開源、高可用的分布式key-value存儲，可以用於配置共享和服務注冊和發現，類似的項目有Zookeeper和consul，提供Restful`的接口，使用簡單，基於raft算法的強一致性，高可用的服務存儲目錄。

5.1 特點

• 完全復制：集群中的每個節點都可以使用完整的存檔

• 高可用性：Etcd可用於避免硬件的單點故障或網絡問題

• 一致性：每次讀取都會返回跨多主機的最新寫入

• 簡單：包括一個定義良好、面向用戶的API（gRPC）

• 快速：每秒10000次寫入的基准速度

• 可靠：使用Raft算法實現強一致性、高可用服務存儲目錄

5.2 應用場景

• 服務發現

  

  服務發現要解決的也是分布式系統中最常見的問題之一，即在同一個分布式集群中的進程或服務，要如何才能找到對方並建立連接。本質上來說，服務發現就是想要了解集群中是否有進程在監聽 udp 或 tcp 端口，並且通過名字就可以查找和連接。

• 配置中心(我們實現的日志收集客戶端需要用到)

  將一些配置信息放到etcd上進行集中管理。 這類場景方式通常是這樣的：應用在啟動到時候主動從etcd獲取一次配置信息，同時在etcd節點上注冊一個Watcher並等待，以后每次配置有更新的時候，etcd都會實時通知訂閱者，以此達到獲取最新配置信息的目的。

• 分布式鎖

  因為 etcd 使用 Raft 算法保持了數據的強一致性，某次操作存儲到集群中的值必然是全局一致的，所以很容易實現分布式鎖。鎖服務有兩種使用方式，一是保持獨占，二是控制時序。

  • 保持獨占即所有獲取鎖的用戶最終只有一個可以得到。etcd 為此提供了一套實現分布式鎖原子操作 CAS（CompareAndSwap）的 API。通過設置prevExist值，可以保證在多個節點同時去創建某個目錄時，只有一個成功。而創建成功的用戶就可以認為是獲得了鎖。

  • 控制時序，即所有想要獲得鎖的用戶都會被安排執行，但是獲得鎖的順序也是全局唯一的，同時決定了執行順序。etcd 為此也提供了一套 API（自動創建有序鍵），對一個目錄建值時指定為POST動作，這樣 etcd 會自動在目錄下生成一個當前最大的值為鍵，存儲這個新的值（客戶端編號）。同時還可以使用 API 按順序列出所有當前目錄下的鍵值。此時這些鍵的值就是客戶端的時序，而這些鍵中存儲的值可以是代表客戶端的編號。

  

5.3 etcd架構

 

從etcd的架構圖中我們可以看到，etcd主要分為四個部分。

• HTTP Server： 用於處理用戶發送的API請求以及其它etcd節點的同步與心跳信息請求。

• Store：用於處理etcd支持的各類功能的事務，包括數據索引、節點狀態變更、監控與反饋、事件處理與執行等等，是etcd對用戶提供的大多數API功能的具體實現。

• Raft：Raft強一致性算法的具體實現，是etcd的核心。

• WAL：Write Ahead Log（預寫式日志），是etcd的數據存儲方式。除了在內存中存有所有數據的狀態以及節點的索引以外，etcd就通過WAL進行持久化存儲。WAL中，所有的數據提交前都會事先記錄日志。Snapshot是為了防止數據過多而進行的狀態快照；Entry表示存儲的具體日志內容。

通常，一個用戶的請求發送過來，會經由HTTP Server轉發給Store進行具體的事務處理，如果涉及到節點的修改，則交給Raft模塊進行狀態的變更、日志的記錄，然后再同步給別的etcd節點以確認數據提交，最后進行數據的提交，再次同步。

重要概念：

• Raft：etcd所采用的保證分布式系統強一致性的算法。

• Node：一個Raft狀態機實例。

• Member： 一個etcd實例。它管理着一個Node，並且可以為客戶端請求提供服務。

• Cluster：由多個Member構成可以協同工作的etcd集群。

• Peer：對同一個etcd集群中另外一個Member的稱呼。

• Client： 向etcd集群發送HTTP請求的客戶端。

• WAL：預寫式日志，etcd用於持久化存儲的日志格式。

• snapshot：etcd防止WAL文件過多而設置的快照，存儲etcd數據狀態。

• Proxy：etcd的一種模式，為etcd集群提供反向代理服務。

• Leader：Raft算法中通過競選而產生的處理所有數據提交的節點。

• Follower：競選失敗的節點作為Raft中的從屬節點，為算法提供強一致性保證。

• Candidate：當Follower超過一定時間接收不到Leader的心跳時轉變為Candidate開始競選。

• Term：某個節點成為Leader到下一次競選時間，稱為一個Term。

• Index：數據項編號。Raft中通過Term和Index來定位數據。

5.4 為什么用etcd而不是zookeeper?

etcd 實現的這些功能，ZooKeeper都能實現。那么為什么要用 etcd 而非直接使用ZooKeeper呢？

• 為什么不選擇ZooKeeper？

1. 部署維護復雜，其使用的Paxos強一致性算法復雜難懂。官方只提供了Java和C兩種語言的接口。

2. 使用Java編寫引入大量的依賴。運維人員維護起來比較麻煩。

3. 最近幾年發展緩慢，不如etcd和consul等后起之秀。

• 為什么選擇etcd？

1. 簡單。使用 Go 語言編寫部署簡單；支持HTTP/JSON API,使用簡單；使用 Raft 算法保證強一致性讓用戶易於理解。

2. etcd 默認數據一更新就進行持久化。

3. etcd 支持 SSL 客戶端安全認證。

最后，etcd 作為一個年輕的項目，正在高速迭代和開發中，這既是一個優點，也是一個缺點。優點是它的未來具有無限的可能性，缺點是無法得到大項目長時間使用的檢驗。然而，目前 CoreOS、Kubernetes和CloudFoundry等知名項目均在生產環境中使用了etcd，所以總的來說，etcd值得你去嘗試。

5.5 raft協議

• 核心要點
  • Leader選舉

• • 日志同步 （Log Replication）

• • leader收到client的更新請求后，會將更新的內容同步給所有follower。

• • 集群狀態的正確性 （Safety）

• • • 保證日志的一致性

    • 保證選舉的正確性

• • zookeeper的zad協議的區別

• • 參考：https://www.cnblogs.com/zhangyafei/p/13926838.html

• • etcd的watch

• • • etcd底層如何實現watch給客戶發通知(websocket)

5.5 etcd的下載和安裝

• 下載

  https://github.com/coreos/etcd/releases

  

  選擇對應版本的下載即可，下載之后解壓

  

• etcd啟動

  雙擊etcd.exe運行

  

• etcd客戶端

  etcdctl.exe --endpoints=127.0.0.1:2379 put zhangyafei "hahaha"

   

5.6 go操作etcd

• 下載

  go get "go.etcd.io/etcd/clientv3"

  注：若運行過程中報undefined: balancer.PickOptions類似錯誤，原始為grpc版本不兼容，需要修改go.mod

  replace google.golang.org/grpc => google.golang.org/grpc v1.26.0

• put和get

  package main
  ​
  import (
      "context"
      "fmt"
      "time"
  ​
      "go.etcd.io/etcd/clientv3"
  )
  ​
  func main() {
      // etcd client put/get demo
      // use etcd/clientv3
      cli, err := clientv3.New(clientv3.Config{
          Endpoints:   []string{"127.0.0.1:2379"},
          DialTimeout: 5 * time.Second,
      })
      if err != nil {
          // handle error!
          fmt.Printf("connect to etcd failed, err:%v\n", err)
          return
      }
      fmt.Println("connect to etcd success")
      defer cli.Close()
      // put
      ctx, cancel := context.WithTimeout(context.Background(), time.Second)
      _, err = cli.Put(ctx, "zhangyafei", "dsb")
      cancel()
      if err != nil {
          fmt.Printf("put to etcd failed, err:%v\n", err)
          return
      }
      // get
      ctx, cancel = context.WithTimeout(context.Background(), time.Second)
      resp, err := cli.Get(ctx, "zhangyafei")
      cancel()
      if err != nil {
          fmt.Printf("get from etcd failed, err:%v\n", err)
          return
      }
      for _, ev := range resp.Kvs {
          fmt.Printf("%s:%s\n", ev.Key, ev.Value)
      }
  }

  

• watch

  package main
  ​
  import (
      "context"
      "fmt"
      "time"
  ​
      "go.etcd.io/etcd/clientv3"
  )
  ​
  // watch demo
  ​
  func main() {
      cli, err := clientv3.New(clientv3.Config{
          Endpoints:   []string{"127.0.0.1:2379"},
          DialTimeout: 5 * time.Second,
      })
      if err != nil {
          fmt.Printf("connect to etcd failed, err:%v\n", err)
          return
      }
      fmt.Println("connect to etcd success")
      defer cli.Close()
      // watch key:q1mi change
      // 派一個哨兵 一直監視着 zhangyafei這個key的變化（新增 刪除 修改））
      rch := cli.Watch(context.Background(), "zhangyafei") // <-chan WatchResponse
      // 從通道嘗試取值（監視的信息）
      for wresp := range rch {
          for _, ev := range wresp.Events {
              fmt.Printf("Type: %s Key:%s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
          }
      }
  }

  我們在etcd目錄終端輸入以下命令，可以查看、新增、刪除key

  etcdctl.exe --endpoints=http://127.0.0.1:2379 put zhangyafei "hello"
  etcdctl.exe --endpoints=http://127.0.0.1:2379 get zhangyafei
  etcdctl.exe --endpoints=http://127.0.0.1:2379 del zhangyafei

  

• lease租約

  package main
   
  import (
      "fmt"
      "time"
  )
   
  // etcd lease
   
  import (
      "context"
      "log"
   
      "go.etcd.io/etcd/clientv3"
  )
   
  func main() {
      cli, err := clientv3.New(clientv3.Config{
          Endpoints:   []string{"127.0.0.1:2379"},
          DialTimeout: time.Second * 5,
      })
      if err != nil {
          log.Fatal(err)
      }
      fmt.Println("connect to etcd success.")
      defer cli.Close()
   
      // 創建一個5秒的租約
      resp, err := cli.Grant(context.TODO(), 5)
      if err != nil {
          log.Fatal(err)
      }
   
      // 5秒鍾之后, /nazha/ 這個key就會被移除
      _, err = cli.Put(context.TODO(), "/nazha/", "dsb", clientv3.WithLease(resp.ID))
      if err != nil {
          log.Fatal(err)
      }
  }

• keepAlive

  package main
   
  import (
      "context"
      "fmt"
      "log"
      "time"
   
      "go.etcd.io/etcd/clientv3"
  )
   
  // etcd keepAlive
   
  func main() {
      cli, err := clientv3.New(clientv3.Config{
          Endpoints:   []string{"127.0.0.1:2379"},
          DialTimeout: time.Second * 5,
      })
      if err != nil {
          log.Fatal(err)
      }
      fmt.Println("connect to etcd success.")
      defer cli.Close()
   
      resp, err := cli.Grant(context.TODO(), 5)
      if err != nil {
          log.Fatal(err)
      }
   
      _, err = cli.Put(context.TODO(), "/nazha/", "dsb", clientv3.WithLease(resp.ID))
      if err != nil {
          log.Fatal(err)
      }
   
      // the key 'foo' will be kept forever
      ch, kaerr := cli.KeepAlive(context.TODO(), resp.ID)
      if kaerr != nil {
          log.Fatal(kaerr)
      }
      for {
          ka := <-ch
          fmt.Println("ttl:", ka.TTL)
      }
  }

• 基於etcd實現分布式鎖

  下載

  import "go.etcd.io/etcd/clientv3/concurrency"

  go.etcd.io/etcd/clientv3/concurrency在etcd之上實現並發操作，如分布式鎖、屏障和選舉。

  示例

  cli, err := clientv3.New(clientv3.Config{Endpoints: endpoints})
  if err != nil {
      log.Fatal(err)
  }
  defer cli.Close()
   
  // 創建兩個單獨的會話用來演示鎖競爭
  s1, err := concurrency.NewSession(cli)
  if err != nil {
      log.Fatal(err)
  }
  defer s1.Close()
  m1 := concurrency.NewMutex(s1, "/my-lock/")
   
  s2, err := concurrency.NewSession(cli)
  if err != nil {
      log.Fatal(err)
  }
  defer s2.Close()
  m2 := concurrency.NewMutex(s2, "/my-lock/")
   
  // 會話s1獲取鎖
  if err := m1.Lock(context.TODO()); err != nil {
      log.Fatal(err)
  }
  fmt.Println("acquired lock for s1")
   
  m2Locked := make(chan struct{})
  go func() {
      defer close(m2Locked)
      // 等待直到會話s1釋放了/my-lock/的鎖
      if err := m2.Lock(context.TODO()); err != nil {
          log.Fatal(err)
      }
  }()
   
  if err := m1.Unlock(context.TODO()); err != nil {
      log.Fatal(err)
  }
  fmt.Println("released lock for s1")
   
  <-m2Locked
  fmt.Println("acquired lock for s2")

  輸出

  acquired lock for s1
  released lock for s1
  acquired lock for s2

• 官方文檔：https://godoc.org/go.etcd.io/etcd/clientv3

7 etcd集群

etcd 作為一個高可用鍵值存儲系統，天生就是為集群化而設計的。由於 Raft 算法在做決策時需要多數節點的投票，所以 etcd 一般部署集群推薦奇數個節點，推薦的數量為 3、5 或者 7 個節點構成一個集群。

搭建一個3節點集群示例：

在每個etcd節點指定集群成員，為了區分不同的集群最好同時配置一個獨一無二的token。

下面是提前定義好的集群信息，其中n1、n2和n3表示3個不同的etcd節點。

TOKEN=token-01
CLUSTER_STATE=new
CLUSTER=n1=http://10.240.0.17:2380,n2=http://10.240.0.18:2380,n3=http://10.240.0.19:2380

在n1這台機器上執行以下命令來啟動etcd：

etcd --data-dir=data.etcd --name n1 \
    --initial-advertise-peer-urls http://10.240.0.17:2380 --listen-peer-urls http://10.240.0.17:2380 \
    --advertise-client-urls http://10.240.0.17:2379 --listen-client-urls http://10.240.0.17:2379 \
    --initial-cluster ${CLUSTER} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

在n2這台機器上執行以下命令啟動etcd：

etcd --data-dir=data.etcd --name n2 \
    --initial-advertise-peer-urls http://10.240.0.18:2380 --listen-peer-urls http://10.240.0.18:2380 \
    --advertise-client-urls http://10.240.0.18:2379 --listen-client-urls http://10.240.0.18:2379 \
    --initial-cluster ${CLUSTER} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

在n3這台機器上執行以下命令啟動etcd：

etcd --data-dir=data.etcd --name n3 \
    --initial-advertise-peer-urls http://10.240.0.19:2380 --listen-peer-urls http://10.240.0.19:2380 \
    --advertise-client-urls http://10.240.0.19:2379 --listen-client-urls http://10.240.0.19:2379 \
    --initial-cluster ${CLUSTER} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

etcd 官網提供了一個可以公網訪問的 etcd 存儲地址。你可以通過如下命令得到 etcd 服務的目錄，並把它作為-discovery參數使用。

curl https://discovery.etcd.io/new?size=3
https://discovery.etcd.io/a81b5818e67a6ea83e9d4daea5ecbc92
 
# grab this token
TOKEN=token-01
CLUSTER_STATE=new
DISCOVERY=https://discovery.etcd.io/a81b5818e67a6ea83e9d4daea5ecbc92
 
 
etcd --data-dir=data.etcd --name n1 \
    --initial-advertise-peer-urls http://10.240.0.17:2380 --listen-peer-urls http://10.240.0.17:2380 \
    --advertise-client-urls http://10.240.0.17:2379 --listen-client-urls http://10.240.0.17:2379 \
    --discovery ${DISCOVERY} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}
 
 
etcd --data-dir=data.etcd --name n2 \
    --initial-advertise-peer-urls http://10.240.0.18:2380 --listen-peer-urls http://10.240.0.18:2380 \
    --advertise-client-urls http://10.240.0.18:2379 --listen-client-urls http://10.240.0.18:2379 \
    --discovery ${DISCOVERY} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}
 
 
etcd --data-dir=data.etcd --name n3 \
    --initial-advertise-peer-urls http://10.240.0.19:2380 --listen-peer-urls http://10.240.0.19:2380 \
    --advertise-client-urls http://10.240.0.19:2379 --listen-client-urls http:/10.240.0.19:2379 \
    --discovery ${DISCOVERY} \
    --initial-cluster-state ${CLUSTER_STATE} --initial-cluster-token ${TOKEN}

到此etcd集群就搭建起來了，可以使用etcdctl來連接etcd。

六、logagent從etcd加載收集項配置

1. 配置

value := `[{"path":"c:/nginx/nginx.log","topic":"web_log"},{"path":"d:/redis/redis.log","topic":"redis_log"},{"path":"e:/mysql/mysql.log","topic":"mysql_log"}]`

2. 將配置信息發送打破etcd

• etcd讀取配置

  

• logagent從etcd加載收集項配置

  conf/config.ini

  [kafka]
  address=127.0.0.1:9092
  ​
  [etcd]
  address=127.0.0.1:2379
  timeout=5
  collect_log_key=/logagent/collect_config

  conf/config.go

  package conf
  ​
  type Config struct {
      Kafka Kafka `ini:"kafka"`
      Etcd  Etcd  `ini:"etcd"`
  }
  ​
  type Kafka struct {
      Address string `ini:"address"`
      //Topic string `ini:"topic"`
  }
  ​
  type Etcd struct {
      Address string `ini:"address"`
      Key     string `ini:"collect_log_key"`
      Timeout int    `ini:"timeout"`
  }

  etcd/etcd.go

  package etcd
  ​
  import (
      "context"
      "encoding/json"
      "fmt"
      "go.etcd.io/etcd/clientv3"
      "strings"
      "time"
  )
  ​
  var (
      cli *clientv3.Client
  )
  ​
  type LogEntry struct {
      Path string `json:"path"`   // 日志存放的路徑
      Topic string `json:"topic"`  // 日志發往kafka中的哪個Topic
  }
  ​
  // 初始化etcd的函數
  ​
  func Init(addr string, timeout time.Duration) (err error) {
      cli, err = clientv3.New(clientv3.Config{
          Endpoints: strings.Split(addr, ";"),
          DialTimeout: timeout,
      })
      return
  }
  ​
  // 從etcd中獲取日志收集項的配置信息
  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.Printf("get from etcd failed, err:%v\n", err)
          return
      }
      for _, ev := range resp.Kvs {
          //fmt.Printf("%s:%s\n", ev.Key, ev.Value)
          err = json.Unmarshal(ev.Value, &logEntryConf)
          if err != nil {
              fmt.Printf("unmarshal etcd value failed,err:%v\n", err)
              return
          }
      }
      return
  }

  main.go

  package main
  ​
  import (
      "fmt"
      "gopkg.in/ini.v1"
      "logagent/conf"
      "logagent/etcd"
      "logagent/kafka"
      "strings"
      "time"
  )
  ​
  var config  = new(conf.Config)
  ​
  // logAgent 入口程序
  ​
  func main() {
      // 0. 加載配置文件
      err := ini.MapTo(config, "./conf/config.ini")
      if err != nil {
          fmt.Printf("Fail to read file: %v", err)
          return
      }
      // 1. 初始化kafka連接
      err = kafka.Init(strings.Split(config.Kafka.Address, ";"))
      if err != nil {
          fmt.Println("init kafka failed, err:%v\n", err)
          return
      }
      fmt.Println("init kafka success.")
  ​
      // 2. 初始化etcd
      err = etcd.Init(config.Etcd.Address, time.Duration(config.Etcd.Timeout) * time.Second)
      if err != nil {
          fmt.Printf("init etcd failed,err:%v\n", err)
          return
      }
      fmt.Println("init etcd success.")
      // 2.1 從etcd中獲取日志收集項的配置信息
      logEntryConf, err := etcd.GetConf(config.Etcd.Key)
      // 2,.2 派一個哨兵
      if err != nil {
          fmt.Printf("etcd.GetConf failed, err:%v\n", err)
          return
      }
      fmt.Printf("get conf from etcd success, %v\n", logEntryConf)
      for index, value := range logEntryConf{
          fmt.Printf("index:%v value:%v\n", index, value)
      }
  }

 項目地址：https://gitee.com/zhangyafeii/go-log-collect/tree/master/code/logagent