package main

import (
    "fmt"
    "strings"
)

type LogProcess struct {
    path string // 日志文件路徑
    dsn string // influxdb dsn
}

func (lp *LogProcess) Read() {
    path := lp.path
    fmt.Println(path)
}

func (lp *LogProcess) Process() {
    log := "hello world"
    fmt.Println(strings.ToUpper(log))
}

func (lp *LogProcess) Write()  {
    dsn := lp.dsn
    fmt.Println(dsn)
}

func main() {
    lp := &LogProcess{
        path: "test path",
        dsn: "test dsn",
    }

    // read log file
    lp.Read()

    // process log
    lp.Process()

    // write data
    lp.Write()
}

0x02: 加上 go 和 chan, 并發(fā)就是如此簡單

加上 go 關(guān)鍵字, 輕松實現(xiàn)協(xié)程:

func main() {
    lp := &LogProcess{
        path: "test path",
        dsn: "test dsn",
    }

    // read log file
    go lp.Read()

    // process log
    go lp.Process()

    // write data
    go lp.Write()

    time.Sleep(time.Second) // 新手必知: 保證程序退出前, 協(xié)程可以執(zhí)行完
}

加上 chan, 輕松實現(xiàn)協(xié)程間通信:

type LogProcess struct {
    path string // 日志文件路徑
    dsn string // influxdb dsn
    rc chan string // read chan
    wc chan string // write chan
}

func (lp *LogProcess) Read() {
    path := lp.path
    fmt.Println(path)

    lp.rc <- "test data"
}

func (lp *LogProcess) Process() {
    log := <- lp.rc
    lp.wc <- strings.ToUpper(log)
}

func (lp *LogProcess) Write()  {
    dsn := lp.dsn
    fmt.Println(dsn)

    data := <- lp.wc
    fmt.Println(data)
}

0x03: 引入 interface, 方便以后擴展

現(xiàn)在是從 文件 讀取, 如果以后要從 其他數(shù)據(jù)源 讀取呢? 這個時候就可以用上接口:

type Reader interface {
    Read(rc chan string)
}

type ReadFromFile struct {
    path string
}

func (r *ReadFromFile) Read(rc chan string) {
    // read from file
}

同理, 數(shù)據(jù)寫入到 influxdb 也可以加入接口, 方便以后擴展.

0x04: 讀取文件的細節(jié)

實時讀取日志文件要怎么實現(xiàn)呢? 直接上代碼, 細節(jié)有很多, 注意 注釋:

• 實時 讀取怎么實現(xiàn): 從文件末尾開始讀取
• 怎么一行一行的讀取日志: buf.ReadBytes('\n')
• 輸出怎么多了換行呢: 截取掉最后的換行符 line[:len(line)-1]

func (r *ReadFromFile) Read(rc chan []byte) {
    f, err := os.Open(r.path)
    if err != nil {
        panic(err)
    }
    defer f.Close()

    f.Seek(0, 2) // 文件末尾
    buf := bufio.NewReader(f) // []byte 數(shù)據(jù)類型, rc chan 的類型也相應(yīng)進行了修改

    for {
        line, err := buf.ReadBytes('\n')
        // todo: 處理日志切割, inode 變化的情況
        if err == io.EOF {
            time.Sleep(500 * time.Millisecond)
        } else if err != nil {
            panic(err)
        } else { // 需要寫到這里
            rc <- line[:len(line)-1]
        }
    }
}

還有一個需要優(yōu)化的地方, 一般日志文件都會采取 輪轉(zhuǎn) 策略(詳見上篇blog devops| 日志服務(wù)實踐), 文件可能更新了, 所以讀取文件時, 還需要加一個判斷.

0x05: 日志解析, 又見正則

日志的解析比較簡單, 按照日志的格式正則匹配即可:

// 使用結(jié)構(gòu)體來記錄匹配到的日志數(shù)據(jù)
type Log struct {
    TimeLocal                    time.Time
    BytesSent                    int
    Path, Method, Scheme, Status string
    UpstreamTime, RequestTime    float64
}

func (l *LogProcess) Process() {
    // 正則
    re := regexp.MustCompile(`([\d\.]+)\s+([^ \[]+)\s+([^ \[]+)\s+\[([^\]]+)\]\s+([a-z]+)\s+\"([^"]+)\"\s+(\d{3})\s+(
\d+)\s+\"([^"]+)\"\s+\"(.*?)\"\s+\"([\d\.-]+)\"\s+([\d\.-]+)\s+([d\.-]+)`)

    loc, _ := time.LoadLocation("PRC")
    for v := range l.rc {
        str := string(v)
        ret := re.FindStringSubmatch(str)
        if len(ret) != 14 {
            log.Println(str)
            continue
        }

        msg := &Log{}
        t, err := time.ParseInLocation("02/Jan/2006:15:04:05 +0000", ret[4], loc)
        if err != nil {
            log.Println(ret[4])
        }
        msg.TimeLocal = t

        byteSent, _ := strconv.Atoi(ret[8])
        msg.BytesSent = byteSent

        // Get /for?query=t HTTP/1.0
        reqSli := strings.Split(ret[6], " ")
        if len(reqSli) != 3 {
            log.Println(ret[6])
            continue
        }
        msg.Method = reqSli[0]
        msg.Scheme = reqSli[2]
        // url parse
        u, err := url.Parse(reqSli[1])
        if err != nil {
            log.Println(reqSli[1])
            continue
        }
        msg.Path = u.Path
        msg.Status = ret[7]
        upTime, _ := strconv.ParseFloat(ret[12], 64)
        reqTime, _ := strconv.ParseFloat(ret[13], 64)
        msg.UpstreamTime = upTime
        msg.RequestTime = reqTime

        l.wc <- msg
    }
}

0x06: 上手 influxdb

influxdb 是時序數(shù)據(jù)庫的一種, 包含如下基礎(chǔ)概念:

• database: 數(shù)據(jù)庫
• measurement: 數(shù)據(jù)庫中的表
• points: 表里的一行數(shù)據(jù)

其中 points 包含以下內(nèi)容:

• tags: 有索引的屬性
• fields: 值
• time: 時間戳, 也是自動生成的主索引

使用 docker 快速開啟 InfluxDb Server:

    influxdb:
        image: influxdb:1.4.3-alpine
        ports:
            - "8086:8086"
        #     - "8083:8083" # admin
        #     - "2003:2003" # graphite
        environment:
            INFLUXDB_DB: log
            INFLUXDB_USER: log
            INFLUXDB_USER_PASSWORD: logpass
        #     INFLUXDB_GRAPHITE_ENABLED: 1
        #     INFLUXDB_ADMIN_ENABLED: 1
        # volumes:
        #     - ./data/influxdb:/var/lib/influxdb

influxdb 使用 go 語言實現(xiàn), 稍微修改一下官方文檔中示例, 就可以使用 client:

InfluxDB Client: https://github.com/influxdata/influxdb/tree/master/client

// 寫入也使用接口
type Writer interface {
    Write(wc chan *Log)
}

type WriteToInfluxdb struct {
    dsn string
}

// 只在官方示例代碼上做了一點修改
func (w *WriteToInfluxdb) Write(wc chan *Log) {
    // dsn 示例: http://localhost:8086@log@logpass@log@s
    dsnSli := strings.Split(w.dsn, "@")

    // Create a new HTTPClient
    c, err := client.NewHTTPClient(client.HTTPConfig{
        Addr:     dsnSli[0],
        Username: dsnSli[1],
        Password: dsnSli[2],
    })
    if err != nil {
        log.Fatal(err)
    }
    defer c.Close()

    // Create a new point batch
    bp, err := client.NewBatchPoints(client.BatchPointsConfig{
        Database:  dsnSli[3],
        Precision: dsnSli[4],
    })
    if err != nil {
        log.Fatal(err)
    }

    for v := range wc {
        // Create a point and add to batch
        tags := map[string]string{
            "Path": v.Path,
            "Method": v.Method,
            "Scheme": v.Scheme,
            "Status": v.Status,
        }
        fields := map[string]interface{}{
            "bytesSent":   v.BytesSent,
            "upstreamTime": v.UpstreamTime,
            "RequestTime":   v.RequestTime,
        }

        pt, err := client.NewPoint("log", tags, fields, v.TimeLocal)
        if err != nil {
            log.Fatal(err)
        }
        bp.AddPoint(pt)

        // Write the batch
        if err := c.Write(bp); err != nil {
            log.Fatal(err)
        }

        // Close client resources
        if err := c.Close(); err != nil {
            log.Fatal(err)
        }
    }
}

0x07: 使用 Grafana 接入 InfluxDB 數(shù)據(jù)源

Grafana 使用 docker 也可以輕松部署:

    grafana:
        image: grafana/grafana:5.1.0-beta1
        ports:
            - "3000:3000"
        environment:
            GF_SERVER_ROOT_URL: http://grafana.server.name
            GF_SECURITY_ADMIN_PASSWORD: secret

官網(wǎng)效果圖:

Grafana

0x08: 簡單監(jiān)控系統(tǒng)實現(xiàn)

作為一個 實時 系統(tǒng), 需要后臺常駐運行, 怎么查看系統(tǒng)的運行狀態(tài)的呢?

加入一個簡單的監(jiān)控系統(tǒng), 通過 http 請求查看系統(tǒng)實時運行狀態(tài):

// 需要監(jiān)控的系統(tǒng)狀態(tài)
type SystemInfo struct {
    LogLine int `json:"logline"` // 總?cè)罩咎幚頂?shù)
    Tps float64 `json:"tps"`
    ReadChanLen int `json:"readchanlen"` // read chan 長度
    WriteChanLen int `json:"writechanlen"` // write chan 長度
    RunTime string `json:"runtime"` // 運行總時間
    ErrNum int `json:"errnum"` // 錯誤數(shù)
}

// 監(jiān)控類
type Monitor struct {
    startTime time.Time
    data SystemInfo
}

// 啟動監(jiān)控, 其實就是一個簡單的 http server
func (m *Monitor) start(lp *LogProcess) {
    http.HandleFunc("/monitor", func(writer http.ResponseWriter, request *http.Request) {
        m.data.RunTime = time.Now().Sub(m.startTime).String()
        m.data.ReadChanLen = len(lp.rc)
        m.data.WriteChanLen = len(lp.wc)

        ret, _ := json.MarshalIndent(m.data, "", "\t")

        io.WriteString(writer, string(ret))
    })

    http.ListenAndServe(":9091", nil)
}

func main() {
    ...

    // 運行監(jiān)控
    m := &Monitor{
        startTime: time.Now(),
        data: SystemInfo{},
    }
    m.start(l)
}

監(jiān)控數(shù)據(jù)中的 TPS 稍微有點難處理:

• 啟動一個定時器, 比如 5s
• 記錄下時間間隔內(nèi)的 LogLine(日志處理行數(shù))

這樣我們就可以用 LogLine 來估算系統(tǒng)的 TPS 了

0x09: 寫在最后

并發(fā)編程實戰(zhàn), 總會給人帶來 又完成了了不起的任務(wù) 的感覺, 特別是會了解更多的細節(jié).

能夠相遇, 也是一種快樂吧.

完整代碼: https://gitee.com/daydaygo/codes/sc6tyr2odf58k39npub0v72