关于golang:聊聊promtail的positions

序

本文次要钻研一下 promtail 的 positions

Positions

loki/pkg/promtail/positions/positions.go

type Positions interface {
    // GetString returns how far we've through a file as a string.
    // JournalTarget writes a journal cursor to the positions file, while
    // FileTarget writes an integer offset. Use Get to read the integer
    // offset.
    GetString(path string) string
    // Get returns how far we've read through a file. Returns an error
    // if the value stored for the file is not an integer.
    Get(path string) (int64, error)
    // PutString records (asynchronously) how far we've read through a file.
    // Unlike Put, it records a string offset and is only useful for
    // JournalTargets which doesn't have integer offsets.
    PutString(path string, pos string)
    // Put records (asynchronously) how far we've read through a file.
    Put(path string, pos int64)
    // Remove removes the position tracking for a filepath
    Remove(path string)
    // SyncPeriod returns how often the positions file gets resynced
    SyncPeriod() time.Duration
    // Stop the Position tracker.
    Stop()}

Positions 接口定义了 GetString、Get、PutString、Put、Remove、SyncPeriod、Stop 办法

positions

loki/pkg/promtail/positions/positions.go

// Positions tracks how far through each file we've read.
type positions struct {
    logger    log.Logger
    cfg       Config
    mtx       sync.Mutex
    positions map[string]string
    quit      chan struct{}
    done      chan struct{}}

func (p *positions) Stop() {close(p.quit)
    <-p.done
}

func (p *positions) PutString(path string, pos string) {p.mtx.Lock()
    defer p.mtx.Unlock()
    p.positions[path] = pos
}

func (p *positions) Put(path string, pos int64) {p.PutString(path, strconv.FormatInt(pos, 10))
}

func (p *positions) GetString(path string) string {p.mtx.Lock()
    defer p.mtx.Unlock()
    return p.positions[path]
}

func (p *positions) Get(path string) (int64, error) {p.mtx.Lock()
    defer p.mtx.Unlock()
    pos, ok := p.positions[path]
    if !ok {return 0, nil}
    return strconv.ParseInt(pos, 10, 64)
}

func (p *positions) Remove(path string) {p.mtx.Lock()
    defer p.mtx.Unlock()
    p.remove(path)
}

func (p *positions) remove(path string) {delete(p.positions, path)
}

func (p *positions) SyncPeriod() time.Duration {return p.cfg.SyncPeriod}

positions 定义了 logger、cfg、mtx、positions、quit、done 属性；它实现了 Positions 接口；其 Get 办法从 p.positions 读取数据；其 Put 办法写数据到 p.positions 中；其 SyncPeriod 办法返回的是 p.cfg.SyncPeriod；其 Remove 办法将 path 从 p.positions 中删除

New

loki/pkg/promtail/positions/positions.go

// New makes a new Positions.
func New(logger log.Logger, cfg Config) (Positions, error) {positionData, err := readPositionsFile(cfg, logger)
    if err != nil {return nil, err}

    p := &positions{
        logger:    logger,
        cfg:       cfg,
        positions: positionData,
        quit:      make(chan struct{}),
        done:      make(chan struct{}),
    }

    go p.run()
    return p, nil
}

New 办法会通过 readPositionsFile 读取 positionData 创立 positions，而后异步执行 p.run()

run

loki/pkg/promtail/positions/positions.go

func (p *positions) run() {defer func() {p.save()
        level.Debug(p.logger).Log("msg", "positions saved")
        close(p.done)
    }()

    ticker := time.NewTicker(p.cfg.SyncPeriod)
    for {
        select {
        case <-p.quit:
            return
        case <-ticker.C:
            p.save()
            p.cleanup()}
    }
}

func (p *positions) save() {
    if p.cfg.ReadOnly {return}
    p.mtx.Lock()
    positions := make(map[string]string, len(p.positions))
    for k, v := range p.positions {positions[k] = v
    }
    p.mtx.Unlock()

    if err := writePositionFile(p.cfg.PositionsFile, positions); err != nil {level.Error(p.logger).Log("msg", "error writing positions file", "error", err)
    }
}

func (p *positions) cleanup() {p.mtx.Lock()
    defer p.mtx.Unlock()
    toRemove := []string{}
    for k := range p.positions {
        // If the position file is prefixed with journal, it's a
        // JournalTarget cursor and not a file on disk.
        if strings.HasPrefix(k, "journal-") {continue}

        if _, err := os.Stat(k); err != nil {if os.IsNotExist(err) {
                // File no longer exists.
                toRemove = append(toRemove, k)
            } else {
                // Can't determine if file exists or not, some other error.
                level.Warn(p.logger).Log("msg", "could not determine if log file"+
                    "still exists while cleaning positions file", "error", err)
            }
        }
    }
    for _, tr := range toRemove {p.remove(tr)
    }
}

run 办法通过 time.NewTicker(p.cfg.SyncPeriod) 来触发执行 p.save() 及 p.cleanup()；save 办法将 positions 长久化到文件；cleanup 办法遍历 p.positions，从内存中移除文件不存在的 position

readPositionsFile

loki/pkg/promtail/positions/positions.go

func readPositionsFile(cfg Config, logger log.Logger) (map[string]string, error) {cleanfn := filepath.Clean(cfg.PositionsFile)
    buf, err := ioutil.ReadFile(cleanfn)
    if err != nil {if os.IsNotExist(err) {return map[string]string{}, nil}
        return nil, err
    }

    var p File
    err = yaml.UnmarshalStrict(buf, &p)
    if err != nil {
        // return empty if cfg option enabled
        if cfg.IgnoreInvalidYaml {level.Debug(logger).Log("msg", "ignoring invalid positions file", "file", cleanfn, "error", err)
            return map[string]string{}, nil}

        return nil, fmt.Errorf("invalid yaml positions file [%s]: %v", cleanfn, err)
    }

    // p.Positions will be nil if the file exists but is empty
    if p.Positions == nil {p.Positions = map[string]string{}}

    return p.Positions, nil
}

readPositionsFile 办法从文件读取地位到 p.Positions

writePositionFile

loki/pkg/promtail/positions/positions.go

func writePositionFile(filename string, positions map[string]string) error {
    buf, err := yaml.Marshal(File{Positions: positions,})
    if err != nil {return err}

    target := filepath.Clean(filename)
    temp := target + "-new"

    err = ioutil.WriteFile(temp, buf, os.FileMode(positionFileMode))
    if err != nil {return err}

    return os.Rename(temp, target)
}

writePositionFile 办法将 positions 写入文件

小结

promtail 的 Positions 接口定义了 GetString、Get、PutString、Put、Remove、SyncPeriod、Stop 办法；positions 实现了 Positions 接口；其 Get 办法从 p.positions 读取数据；其 Put 办法写数据到 p.positions 中；其 SyncPeriod 办法返回的是 p.cfg.SyncPeriod；其 Remove 办法将 path 从 p.positions 中删除。

doc

• promtail