1. 程式人生 > >Go實現海量日誌收集系統(二)

Go實現海量日誌收集系統(二)

fig encode 文件配置 sar 架構 cli 代碼執行 CP lob

一篇文章主要是關於整體架構以及用到的軟件的一些介紹,這一篇文章是對各個軟件的使用介紹,當然這裏主要是關於架構中我們agent的實現用到的內容

關於zookeeper+kafka

我們需要先把兩者啟動,先啟動zookeeper,再啟動kafka
啟動ZooKeeper:./bin/zkServer.sh start
啟動kafka:./bin/kafka-server-start.sh ./config/server.properties

操作kafka需要安裝一個包:go get github.com/Shopify/sarama
寫一個簡單的代碼,通過go調用往kafka裏扔數據:

package main

import (
    
"github.com/Shopify/sarama" "fmt" ) func main() { config := sarama.NewConfig() config.Producer.RequiredAcks = sarama.WaitForAll config.Producer.Partitioner = sarama.NewRandomPartitioner config.Producer.Return.Successes = true msg := &sarama.ProducerMessage{} msg.Topic
= "nginx_log" msg.Value = sarama.StringEncoder("this is a good test,my message is zhaofan") client,err := sarama.NewSyncProducer([]string{"192.168.0.118:9092"},config) if err != nil{ fmt.Println("producer close err:",err) return } defer client.Close() pid,offset,err :
= client.SendMessage(msg) if err != nil{ fmt.Println("send message failed,",err) return } fmt.Printf("pid:%v offset:%v\n",pid,offset) }

config.Producer.RequiredAcks = sarama.WaitForAll 這裏表示是在給kafka扔數據的時候是否需要確認收到kafka的ack消息

msg.Topic = "nginx_log" 因為kafka是一個分布式系統,假如我們要讀的是nginx日誌,apache日誌,我們可以根據topic做區分,同時也是我們也可以有不同的分區

我們將上述代碼執行一下,就會往kafka中扔一條消息,可以通過kakfa中自帶的消費者命令查看:
./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic nginx_log --from-beginning

技術分享圖片

我們可以將最後的代碼稍微更改一下,更改為循環發送:

for{
    pid,offset,err := client.SendMessage(msg)
    if err != nil{
        fmt.Println("send message failed,",err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n",pid,offset)
    time.Sleep(2*time.Second)
}

這樣當我們再次執行的程序的時候,我們可以看到客戶端在不停的消費到數據:

技術分享圖片

這樣我們就實現一個kakfa的生產者的簡單的demo

接下來我們還需要知道一個工具的使用tailf

tailf

我們的agent需要讀日誌目錄下的日誌文件,而日誌文件是不停的增加並且切換文件的,所以我們就需要借助於tailf這個包來讀文件,當然這裏的tailf和linux裏的tail -f命令雖然不同,但是效果是差不多的,都是為了獲取日誌文件新增加的內容。

而我們的客戶端非常重要的一個地方就是要讀日誌文件並且將讀到的日誌文件推送到kafka

這裏需要我們下載一個包:go get github.com/hpcloud/tail

我們通過下面一個例子對這個包進行一個基本的使用,更詳細的api說明看:https://godoc.org/github.com/hpcloud/tail

package main

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

func main() {
    filename := "/Users/zhaofan/go_project/src/go_dev/13/tailf/my.log"
    tails,err := tail.TailFile(filename,tail.Config{
        ReOpen:true,
        Follow:true,
        Location:&tail.SeekInfo{Offset:0,Whence:2},
        MustExist:false,
        Poll:true,
    })

    if err !=nil{
        fmt.Println("tail file err:",err)
        return
    }

    var msg *tail.Line
    var ok bool
    for true{
        msg,ok = <-tails.Lines
        if !ok{
            fmt.Printf("tail file close reopen,filenam:%s\n",tails,filename)
            time.Sleep(100*time.Millisecond)
            continue
        }
        fmt.Println("msg:",msg.Text)
    }
}

最終實現的效果是當你文件裏面添加內容後,就可以不斷的讀取文件中的內容

日誌庫的使用

這裏是通過beego的日誌庫實現的,beego的日誌庫是可以單獨拿出來用的,還是非常方便的,使用例子如下:

package main

import (
    "github.com/astaxie/beego/logs"
    "encoding/json"
    "fmt"
)

func main() {
    config := make(map[string]interface{})
    config["filename"] = "/Users/zhaofan/go_project/src/go_dev/13/log/logcollect.log"
    config["level"] = logs.LevelTrace
    configStr,err := json.Marshal(config)
    if err != nil{
        fmt.Println("marshal failed,err:",err)
        return
    }
    logs.SetLogger(logs.AdapterFile,string(configStr))
    logs.Debug("this is a debug,my name is %s","stu01")
    logs.Info("this is a info,my name is %s","stu02")
    logs.Trace("this is trace my name is %s","stu03")
    logs.Warn("this is a warn my name is %s","stu04")
}

簡單版本logagent的實現

這裏主要是先實現核心的功能,後續再做優化和改進,主要實現能夠根據配置文件中配置的日誌路徑去讀取日誌並將讀取的實時推送到kafka消息隊列中

關於logagent的主要結構如下:

技術分享圖片

程序目錄結構為:

├── conf
│   └── app.conf
├── config.go
├── kafka.go
├── logs
│   └── logcollect.log
├── main.go
└── server.go

app.conf :配置文件
config.go:用於初始化讀取配置文件中的內容,這裏的配置文件加載是通過之前自己實現的配置文件熱加載包處理的,博客地址:http://www.cnblogs.com/zhaof/p/8593204.html
logcollect.log:日誌文件
kafka.go:對kafka的操作,包括初始化kafka連接,以及給kafka發送消息
server.go:主要是tail 的相關操作,用於去讀日誌文件並將內容放到channel中

所以這裏我們主要的代碼邏輯或者重要的代碼邏輯就是kafka.go 以及server.go

kafka.go代碼內容為:

// 這裏主要是kafak的相關操作,包括了kafka的初始化,以及發送消息的操作
package main

import (
    "github.com/Shopify/sarama"
    "github.com/astaxie/beego/logs"
)

var (
    client sarama.SyncProducer
    kafkaSender *KafkaSender
)

type KafkaSender struct {
    client sarama.SyncProducer
    lineChan chan string
}

// 初始化kafka
func NewKafkaSender(kafkaAddr string)(kafka *KafkaSender,err error){
    kafka = &KafkaSender{
        lineChan:make(chan string,100000),
    }
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Successes = true

    client,err := sarama.NewSyncProducer([]string{kafkaAddr},config)
    if err != nil{
        logs.Error("init kafka client failed,err:%v\n",err)
        return
    }
    kafka.client = client
    for i:=0;i<appConfig.KafkaThreadNum;i++{
        // 根據配置文件循環開啟線程去發消息到kafka
        go kafka.sendToKafka()
    }
    return
}

func initKafka()(err error){
    kafkaSender,err = NewKafkaSender(appConfig.kafkaAddr)
    return
}

func (k *KafkaSender) sendToKafka(){
    //從channel中讀取日誌內容放到kafka消息隊列中
    for v := range k.lineChan{
        msg := &sarama.ProducerMessage{}
        msg.Topic = "nginx_log"
        msg.Value = sarama.StringEncoder(v)
        _,_,err := k.client.SendMessage(msg)
        if err != nil{
            logs.Error("send message to kafka failed,err:%v",err)
        }
    }
}

func (k *KafkaSender) addMessage(line string)(err error){
    //我們通過tailf讀取的日誌文件內容先放到channel裏面
    k.lineChan <- line
    return
}

server.go的代碼為:

package main

import (
    "github.com/hpcloud/tail"
    "fmt"
    "sync"
    "github.com/astaxie/beego/logs"
    "strings"
)

type TailMgr struct {
    //因為我們的agent可能是讀取多個日誌文件,這裏通過存儲為一個map
    tailObjMap map[string]*TailObj
    lock sync.Mutex
}

type TailObj struct {
    //這裏是每個讀取日誌文件的對象
    tail *tail.Tail
    offset int64  //記錄當前位置
    filename string
}

var tailMgr *TailMgr
var waitGroup sync.WaitGroup

func NewTailMgr()(*TailMgr){
    tailMgr =  &TailMgr{
        tailObjMap:make(map[string]*TailObj,16),
    }
    return tailMgr
}

func (t *TailMgr) AddLogFile(filename string)(err error){
    t.lock.Lock()
    defer t.lock.Unlock()
    _,ok := t.tailObjMap[filename]
    if ok{
        err = fmt.Errorf("duplicate filename:%s\n",filename)
        return
    }
    tail,err := tail.TailFile(filename,tail.Config{
        ReOpen:true,
        Follow:true,
        Location:&tail.SeekInfo{Offset:0,Whence:2},
        MustExist:false,
        Poll:true,
    })

    tailobj := &TailObj{
        filename:filename,
        offset:0,
        tail:tail,
    }
    t.tailObjMap[filename] = tailobj
    return
}

func (t *TailMgr) Process(){
    //開啟線程去讀日誌文件
    for _, tailObj := range t.tailObjMap{
        waitGroup.Add(1)
        go tailObj.readLog()
    }
}

func (t *TailObj) readLog(){
    //讀取每行日誌內容
    for line := range t.tail.Lines{
        if line.Err != nil {
            logs.Error("read line failed,err:%v",line.Err)
            continue
        }
        str := strings.TrimSpace(line.Text)
        if len(str)==0 || str[0] == ‘\n‘{
            continue
        }

        kafkaSender.addMessage(line.Text)
    }
    waitGroup.Done()
}


func RunServer(){
    tailMgr = NewTailMgr()
    // 這一部分是要調用tailf讀日誌文件推送到kafka中
    for _, filename := range appConfig.LogFiles{
        err := tailMgr.AddLogFile(filename)
        if err != nil{
            logs.Error("add log file failed,err:%v",err)
            continue
        }

    }
    tailMgr.Process()
    waitGroup.Wait()
}

可以整體演示一下代碼實現的效果,當我們運行程序之後我配置文件配置的目錄為:
log_files=/app/log/a.log,/Users/zhaofan/a.log
我通過一個簡單的代碼對對a.log循環追加內容,你可以從kafka的客戶端消費力看到內容了:

技術分享圖片

完成的代碼地址:https://github.com/pythonsite/logagent

小結

這次只是實現logagent的核心功能,實現了從日誌文件中通過多個線程獲取要讀的日誌內容,這裏借助了tailf,並將獲取的內容放到channel中,kafka.go會從channel中取出日誌內容並放到kafka的消息隊列中
這裏並沒有做很多細致的處理,下一篇文章會在這個代碼的基礎上進行改進。同時現在的配置文件的方式也不是最佳的,每次改動配置文件都需要重新啟動程序,後面將通過etcd的方式。

Go實現海量日誌收集系統(二)