序
本文主要研究一下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中删除。
网友评论