此文已由作者楊望暑授權網易雲社群釋出。

歡迎訪問網易雲社群，瞭解更多網易技術產品運營經驗。

背景

在服務端檢視log會經常使用到tail -f命令實時跟蹤檔案變化. 那麼問題來了, 如果自己寫一個同樣功能的, 該何處寫起呢? 如果你用過ELK裡的beats/filebeat的話, 應該知道filebeat做的事情就是監控日誌變化, 並把最新資料,按照自定義配置處理後, 傳送給ElasticSearch/kafka/... 對, 本文就是想介紹如何自己實現一個簡易版filebeat, 只要日誌內容發生變化(append new line), 能觸發一個訊息, 實現對這一行資料的預處理, 列印, 接入kafka等動作, 還有一個功能是, 當這個工具重啟後, 依然能從上次讀取的位置開始讀.

工具

Golang IDEA

大致流程

具體實現

從流程圖中可以看出, 我們需要解決下面幾個問題

1. 記錄上一次程式關閉前,檔案讀取位置,下次程式啟動時候載入這個位置資訊.

2. 檔案定位並按行讀取, 併發布讀取的行

3. 監測檔案內容變化,併發出通知

記錄上次讀取位置

這個問題關鍵應該是什麼時候記錄上次讀取的offset.

1. 讀取併發布後記錄 如果釋出後,做記錄前,程式掛了,那麼重啟程式後,這行資料會重新被讀一次.

2. 讀取後馬上記錄,記錄成功後,才對外發布. 這樣會產生另一個問題, 釋出前程式掛了, 重啟後, 那條未必傳送的訊息,外部是拿不到了.

如果沒理解錯, elastic的filebeat選的就是第一種,且沒做相應的異常處理, 他是設定一個channel池, 接收並非同步寫入位置資訊, 如果寫入失敗, 則列印一條error日誌就繼續走了

logp.Err("Writing of registry returned error: %v. Continuing...", err)複製程式碼

檔案定位並按行讀取, 併發布讀取的行

要讀取一個檔案, 首先要有一個reader

func (tail *Tailf) openReader() {
    tail.file, _ = os.Open(tail.FileName)
    tail.reader = bufio.NewReader(tail.file)
}複製程式碼

對於從檔案位置(offset)=0處開始讀一行, 這沒什麼問題, 直接用下面這個方法就可以了.

func (tail *Tailf) read
 
Line() (string, error) {
    line, err := tail.reader.ReadString('\n')    if err != nil {        return line, err
    }
    line = strings.TrimRight(line, "\n")    return line, err
}複製程式碼

但是, 對於檔案內容增加了, 但是還沒到一行,也就是沒出現\n 卻出現了EOF(end of file), 那這個情況下, 我們是要等待的,offset必須保持在這一行的行頭.

func (tail *Tailf) getOffset() (offset int64, err error) {
    offset, err = tail.file.Seek(0, os.SEEK_CUR)
    offset -= int64(tail.reader.Buffered())    return}func (tail *Tailf) beginWatch() {
    tail.openReader()    var offset int64
    for {       //取上一次讀取位置(行頭)
        offset, _ = tail.getOffset()
        line, err := tail.readLine()        if err == nil {
            tail.publishLine(line)
        } else if err == io.EOF {            //讀到了EOF, offset設定回到行頭
            tail.seekTo(Seek{offset: offset, whence: 0})            //block and wait for changes
            tail.waitChangeEvent()
        } else {
            fmt.Println(err)            return
        }
    }
}func (tail *Tailf) seekTo(pos Seek) error {
    tail.file.Seek(pos.offset, pos.whence)    //一旦改變了offset, 這個reader必須reset一下才能生效
    tail.reader.Reset(tail.file)    return nil}// 這裡是釋出一個訊息, 因為是demo,所以只是簡單的往channel裡一扔func (tail *Tailf) publishLine(line string) {
    tail.Lines <- line
}複製程式碼

下面說說waitChangeEvent

如何監視檔案內容變化,並通知

監測檔案內容增加的方式大體有2種

1. 監測檔案最後修改時間以及檔案大小的變化,俗稱poll--輪詢

2. 利用linux的inotify命令實現監測,他會在檔案發生狀態改變後觸發事件

這裡採用第一種方式, filebeat也用的第一種. 我們自己怎麼實現呢?

//currReadPos: 檔案末尾的offset,也就是當前檔案大小func (w *PollWatcher) ChangeEvent(currReadPos int64) (*ChangeEvent, error) {

    watchingFile, err := os.Stat(w.FileName)    if err != nil {        return nil, err
    }
    changes := NewChangeEvent()    //當前的大小
    w.FileSize = currReadPos    //之前的修改時間
    previousModTime := watchingFile.ModTime()    //輪詢
    go func() {
        previousSize := w.FileSize        for {
            time.Sleep(POLL_DURATION)            //這裡省略很多程式碼, 假設檔案是存在的,且沒被重新命名,刪除之類的情況, 檔案是像日誌檔案一樣不斷append的 
            file, _ := os.Stat(w.FileName)        // ... 省略一大段程式碼
            if previousSize > 0 && previousSize < w.FileSize {                //檔案肥了
                changes.NotifyModified()
                previousSize = w.FileSize                continue
            }

            previousSize = w.FileSize            // 處理 原本沒內容, 但是加入了內容, 所以要用修改時間
            modTime := file.ModTime()            if modTime != previousModTime {
                previousModTime = modTime
                changes.NotifyModified()
            }
        }
    }()    return changes, nil}複製程式碼

這裡的changes.NotifyModified方法只是往下面例項裡Modified Channel 放入 ce.Modified <- true

type ChangeEvent struct {
    Modified  chan bool
    Truncated chan bool
    Deleted   chan bool}複製程式碼

也正是這個動作, 在主執行緒中, 就能收到檔案被修改的通知, 從而繼續出發readLine動作

// 上面有個beginWatch方法程式碼,結合這個程式碼來看func (tail *Tailf) waitChangeEvent() error {    // ... 省略初始化動作
    select {    //只測試檔案內容增加
    case <-tail.changes.Modified:
        fmt.Println(">> find Modified")        return nil
    // ... 省略其他
    }
}複製程式碼

有了這個一連串的程式碼後, 我們就能在main裡監視檔案變化了

func main() {
    t, _ := tailf.NewTailf("/Users/yws/Desktop/test.log")    for line := range t.Lines {    //這裡會block住,有新行到來,就會輸出新行
        fmt.Println(line)
    }
}複製程式碼

擴充套件點

這個擴充套件點, 和filebeat一樣.

1. 在讀取時候, 不一定是按行讀取,可以讀多行,json解析等

2. 釋出時候, 本文例子是直接寫console, 其實可以接kafka, redis, 資料庫等

3. .... 想不出來了

總結

雖然是一個很簡單的功能, 現代主流服務端程式語言基本都能實現, 但為什麼用go來實現呢? 一大堆優點和缺點就不列了..這不是軟文. 談談go初學者的看法

1. 程式碼很簡潔, 雖然不支援很多高階語言特性, 但看起來依然那麼爽, 除了那些過渡包裝的struct以及怪異的取名.

2. 寫併發(goroutine)是那麼的簡單,那麼的優雅,但也很容易被我這樣的菜鳥濫用, 這語言debug目前有點肉痛

3. goroutine通訊也是那麼的簡單, channel設計的很棒, 用著很爽

4. 不爽的地方, 多返回值的問題, 寫慣了java的xinstance.method(yInstance.method()), 當yInstance.method()是多返回值的時候,必須拆分成2行或更多, 每次編譯器報錯時候就想砸鍵盤.

參考資料

1. github.com/elastic/bea… filebeat只是其中一個feature

2. github.com/hpcloud/tai… 寫到一半發現原來別人也幹過一樣的事了, 程式碼基本大同小異, 有興趣的可以看他的程式碼, 寫的更完善.

網易雲免費體驗館，0成本體驗20+款雲產品！

更多網易技術、產品、運營經驗分享請點選。

相關文章：
【推薦】 Spark——為資料分析處理提供更為靈活的賦能