Go語言學習之11 日志收集系統kafka庫實戰

本節主要內容：

1. 日志收集系統設計
2. 日志客戶端開發

1. 項目背景
    a. 每個系統都有日志，當系統出現問題時，需要通過日志解決問題
    b. 當系統機器比較少時，登陸到服務器上查看即可滿足
    c. 當系統機器規模巨大，登陸到機器上查看幾乎不現實

2. 解決方案
    a. 把機器上的日志實時收集，統一的存儲到中心系統
    b. 然后再對這些日志建立索引，通過搜索即可以找到對應日志
    c. 通過提供界面友好的web界面，通過web即可以完成日志搜索

3. 面臨的問題
    a. 實時日志量非常大，每天幾十億條
    b. 日志准實時收集，延遲控制在分鍾級別
    c. 能夠水平可擴展

4. 業界方案ELK

    日志收集系統架構

    該方案問題：

    a. 運維成本高，每增加一個日志收集，都需要手動修改配置
    b. 監控缺失，無法准確獲取logstash的狀態
    c. 無法做定制化開發以及維護

6. 日志收集系統設計

 

    各組件介紹：
    a. Log Agent，日志收集客戶端，用來收集服務器上的日志
    b. Kafka，高吞吐量的分布式隊列，linkin開發，apache頂級開源項目
    c. ES，elasticsearch，開源的搜索引擎，提供基於http restful的web接口
    d. Hadoop，分布式計算框架，能夠對大量數據進行分布式處理的平台

7. kafka應用場景
    1. 異步處理, 把非關鍵流程異步化，提高系統的響應時間和健壯性

 

     2. 應用解耦,通過消息隊列

    3. 流量削峰3. 流量削峰

 

 8. zookeeper應用場景

     1. 服務注冊&服務發現

 

 

     2. 配置中心

    3. 分布式鎖

• Zookeeper是強一致的
• 多個客戶端同時在Zookeeper上創建相同znode，只有一個創建成功

 9. 安裝kafka

     見博客：https://www.cnblogs.com/xuejiale/p/10505391.html

10. log agent設計

11. log agent流程

 

11. kafka示例

      先導入第三方包：

github.com/Shopify/sarama

     我的kafka和ZooKeeper都安裝在Linux（Centos6.5，ip: 192.168.30.136）上：

package main

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

func main() {

    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Successes = true

    client, err := sarama.NewSyncProducer([]string{"192.168.30.136:9092"}, config)
    if err != nil {
        fmt.Println("producer close, err:", err)
        return
    }

    defer client.Close()
    for {
        msg := &sarama.ProducerMessage{}
        msg.Topic = "nginx_log"
        msg.Value = sarama.StringEncoder("this is a good test, my message is good")

        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(time.Second)
    }
}

   注意：Shopify/sarama的同步/異步producer，https://www.jianshu.com/p/666d2604e8f8

    Windows啟動程序往Linux上的kafka發送數據：

    Linux上的kafka接收數據：

  再來看一個kafka生產和消費示例：

package main

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

func main() {
    // 新建一個arama配置實例
    config := sarama.NewConfig()
    // WaitForAll waits for all in-sync replicas to commit before responding.
    config.Producer.RequiredAcks = sarama.WaitForAll
    // NewRandomPartitioner returns a Partitioner which chooses a random partition each time.
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Successes = true

    // new producer
    client, err := sarama.NewSyncProducer([]string{"192.168.30.136:9092"}, config)
    if err != nil {
        fmt.Println("producer close, err:", err)
        return
    }
    defer client.Close()

    // new message
    msg := &sarama.ProducerMessage{}
    msg.Topic = "food"
    msg.Key = sarama.StringEncoder("fruit")
    msg.Value = sarama.StringEncoder("apple")

    // send message
    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)

    // new message
    msg2 := &sarama.ProducerMessage{}
    msg2.Topic = "food"
    msg2.Key = sarama.StringEncoder("fruit")
    msg2.Value = sarama.StringEncoder("orange")

    // send message
    pid2, offset2, err := client.SendMessage(msg2)
    if err != nil {
        fmt.Println("send message failed,", err)
        return
    }
    fmt.Printf("pid2: %v, offset2:%v\n", pid2, offset2)

    fmt.Println("Produce success.")
}

package main

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

var wg sync.WaitGroup

func main() {
    consumer, err := sarama.NewConsumer([]string{"192.168.30.136:9092"}, nil)
    if err != nil {
        fmt.Println("consumer connect error:", err)
        return
    }
    fmt.Println("connnect success...")
    defer consumer.Close()

    partitions, err := consumer.Partitions("food")
    if err != nil {
        fmt.Println("geet partitions failed, err:", err)
        return
    }

    for _, p := range partitions {
        partitionConsumer, err := consumer.ConsumePartition("food", p, sarama.OffsetOldest)
        if err != nil {
            fmt.Println("partitionConsumer err:", err)
            continue
        }
        wg.Add(1)
        go func(){
            for m := range partitionConsumer.Messages() {
                fmt.Printf("key: %s, text: %s, offset: %d\n", string(m.Key), string(m.Value), m.Offset)
            }
            wg.Done()
        }()
    }
    wg.Wait()

    fmt.Println("Consumer success.")
}

12. tailf組件使用

    先導入第三方包：

github.com/hpcloud/tail

package main

import (
    "fmt"
    "github.com/hpcloud/tail"
    "time"
)
func main() {
    filename := "F:\\Go\\project\\src\\go_dev\\logCollect\\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 {
        msg, ok = <-tails.Lines
        if !ok {
            fmt.Printf("tail file close reopen, filename:%s\n", tails.Filename)
            time.Sleep(100 * time.Millisecond)
            continue
        }
        fmt.Println("msg:", msg)
    }
}

    my.log文件內容（unix格式）：

    在Windows上，當我的上面代碼里日志文件（my.log）為Windows格式，代碼執行結果如下：

    當時用notepade++將文件格式轉換為Unix格式，執行代碼結果如下：

    注意：最后一行必須有換行符，否則該行無法讀取。

13. 配置文件庫使用

    先導入第三方包：

github.com/astaxie/beego/config

    1) 初始化配置庫

iniconf, err := NewConfig("ini", "testini.conf")
if err != nil {
    log.Fatal(err)
}

    2) 讀取配置項

String(key string) string
Int(key string) (int, error)
Int64(key string) (int64, error)
Bool(key string) (bool, error) 
Float(key string) (float64, error)

   例如：

iniconf.String("server::listen_ip")
iniconf.Int("server::listen_port")

[server]
listen_ip = "0.0.0.0"
listen_port = 8080

[logs]
log_level=debug
log_path=./logs/logagent.log

[collect]
log_path=/home/work/logs/nginx/access.log
topic=nginx_log

package main

import (
    "fmt"
    "github.com/astaxie/beego/config"
)

func main() {
    conf, err := config.NewConfig("ini", "./logcollect.conf")
    if err != nil {
        fmt.Println("new config failed, err:", err)
        return
    }

    port, err := conf.Int("server::listen_port")
    if err != nil {
        fmt.Println("read server:port failed, err:", err)
        return
    }

    fmt.Println("Port:", port)
    log_level := conf.String("log::log_level")
    if err != nil {
        fmt.Println("read log_level failed, ", err)
        return
    }
    fmt.Println("log_level:", log_level)

    log_path := conf.String("log::log_path")
    fmt.Println("log_path:", log_path)
}

    配置文件內容：

[server]
listen_ip = "0.0.0.0"
listen_port = 8080

[log]
log_level=debug
log_path=./logs/logagent.log

[collect]
log_path=/home/work/logs/nginx/access.log
topic=nginx_log

    執行結果：

14. 日志庫的使用

        先導入第三方包：

github.com/astaxie/beego/logs

    1） 配置log組件

config := make(map[string]interface{})
config["filename"] = "./logs/logcollect.log"
config["level"] = logs.LevelDebug

configStr, err := json.Marshal(config)
if err != nil {
    fmt.Println("marshal failed, err:", err)
    return
}

    2） 初始化日志組件

logs.SetLogger(“file”, string(configStr))

package main

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

func main() {
    config := make(map[string]interface{})
    config["filename"] = "./logcollect.log"
    config["level"] = logs.LevelDebug

    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 test, my name is %s", "stu01")
    logs.Trace("this is a trace, my name is %s", "stu02")
    logs.Warn("this is a warn, my name is %s", "stu03")
}

 15. 日志收集項目整體實現

    開發環境為Windows系統，go version go1.12.1 windows/amd64， kafka_2.11-2.0.0，zookeeper-3.4.12。

   先實現了一個demo，V1版本：

（1）代碼結構圖

 

（2）代碼地址見本人github：https://github.com/XJL635438451/logCollectProject/tree/master

（3）如何運行

    1）先安裝 go， kafka，zookeeper；

    2）先啟動 zookeeper，然后啟動kafka，下面是啟動的命令；

啟動ZK
.\zkServer.cmd

啟動kafka
F:\Go\project\src\module\kafka_2.11-2.0.0>.\bin\windows\kafka-server-start.bat .\config\server.properties

創建topic
F:\Go\project\src\module\kafka_2.11-2.0.0>.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafkaTest

啟動生產者：
F:\Go\project\src\module\kafka_2.11-2.0.0>.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic kafkaTest

啟動消費者：
F:\Go\project\src\module\kafka_2.11-2.0.0>.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic nginx_log  --from-beginning

    3）如果自己不行寫日志文件，可以運行代碼中的 writeLogTest/log.go，然后運行 main.exe （如果自己修改了代碼還需要重新編譯）；

    4）可以起一個kafka的consumer來查看日志是否寫入到了kafka，方法就是上面的啟動生產者命令，如果正常就可以看到日志一直在kafka中刷新。