package main import ( "bytes" "context" "crypto/rand" "encoding/hex" "encoding/json" "errors" "fmt" "io" "io/fs" "os" "path/filepath" "regexp" "strings" "sync" "time" ) // fileSourceDef is one log file declared in Settings > Sources. type fileSourceDef struct { ID string `json:"id"` Name string `json:"name"` // app of the entries Path string `json:"path"` // the file, as mounted in the container Format string `json:"format"` // auto (detected), or text, json, logfmt, regex, clf, traefik Regex string `json:"regex,omitempty"` // format regex: named groups time, level, msg, host, app… Host string `json:"host,omitempty"` // host of the entries (empty: the host name of the machine) Enabled bool `json:"enabled"` } // filesConfig is saved in /data/files.json. type filesConfig struct { Sources []fileSourceDef `json:"sources"` } // filePos is where reading resumes in a file: its inode and an offset before // which every entry was sent. Until: end of the history found when the // source was turned on; entries before it are kept only when dated within // FILES_BACKFILL. type filePos struct { Ino uint64 `json:"ino"` Off int64 `json:"off"` Until int64 `json:"until,omitempty"` } type filesState struct { Pos map[string]filePos `json:"pos"` // by source id Started map[string]time.Time `json:"started"` // when each source was turned on } type fileSourceStatus struct { Open bool Read uint64 Detected string // format found when the source is in auto Code string Error string } // fileSource is a definition with its parser (nil in auto: detected from // the file). type fileSource struct { fileSourceDef parser *fileParser } // pendingEntry is an entry waiting for its continuation lines. type pendingEntry struct { rec *fileRec off int64 // offset of its first line seen time.Time // when its last line was read history bool // read from the history found when the source was turned on lines int } // fileTail is the reading state of one file, owned by the Run goroutine. type fileTail struct { readOff int64 // where the next read starts (after the lines of the pending entry) known bool // readOff is set pending *pendingEntry auto *fileParser // format detected from the file, in auto } // Files follows log files of other programs (Traefik, nginx, applications…) // mounted in the container, one entry per line in the format detected or // chosen for each file. Source type "file". type Files struct { sink func(*Entry) hostName string backfill time.Duration cfgPath string statePath string wake chan struct{} // Owned by the Run goroutine. st filesState tails map[string]*fileTail dirty bool mu sync.Mutex cfg []*fileSource resets map[string]bool // sources to read again from FILES_BACKFILL ago redetect map[string]bool // sources whose format changed status map[string]*fileSourceStatus } func NewFiles(hostRoot, dataDir string, backfill time.Duration, sink func(*Entry)) *Files { f := &Files{ sink: sink, hostName: readTrim(filepath.Join(hostRoot, "etc/hostname")), backfill: backfill, cfgPath: filepath.Join(dataDir, "files.json"), statePath: filepath.Join(dataDir, "files-state.json"), wake: make(chan struct{}, 1), tails: map[string]*fileTail{}, resets: map[string]bool{}, redetect: map[string]bool{}, status: map[string]*fileSourceStatus{}, } if f.hostName == "" { f.hostName = "localhost" } var cfg filesConfig if b, err := os.ReadFile(f.cfgPath); err == nil { _ = json.Unmarshal(b, &cfg) } for _, d := range cfg.Sources { src, err := compileFileSource(d) if err != nil { logEvent("files", 4, "file %q ignored: %v", d.Name, err) continue } f.cfg = append(f.cfg, src) } if b, err := os.ReadFile(f.statePath); err == nil { _ = json.Unmarshal(b, &f.st) } if f.st.Pos == nil { f.st.Pos = map[string]filePos{} } if f.st.Started == nil { f.st.Started = map[string]time.Time{} } return f } const maxFileSources = 50 var fileIDRe = regexp.MustCompile(`^[0-9a-f]{1,32}$`) // checkFilePath: one file, absolute, without wildcards. func checkFilePath(p string) (string, error) { p = strings.TrimSpace(p) switch { case !filepath.IsAbs(p): return "", errors.New("the path must be absolute (as mounted in the container)") case strings.ContainsAny(p, "*?["): return "", errors.New("give the path of one file (no wildcards)") } return filepath.Clean(p), nil } // compileFileSource checks a definition and builds its parser. func compileFileSource(d fileSourceDef) (*fileSource, error) { d.Name = strings.TrimSpace(d.Name) d.Host = strings.TrimSpace(d.Host) if d.Format == "" { d.Format = fmtAuto } if d.Format != fmtRegex { d.Regex = "" } switch { case !fileIDRe.MatchString(d.ID): return nil, errors.New("invalid id") case d.Name == "" || len(d.Name) > 64: return nil, errors.New("the name is required (64 characters at most)") case len(d.Host) > 253: return nil, fmt.Errorf("%s: host name too long", d.Name) } p, err := checkFilePath(d.Path) if err != nil { return nil, fmt.Errorf("%s: %v", d.Name, err) } d.Path = p src := &fileSource{fileSourceDef: d} if d.Format != fmtAuto { if src.parser, err = newFileParser(d.Format, d.Regex); err != nil { return nil, fmt.Errorf("%s: %v", d.Name, err) } } return src, nil } func newFileID() string { b := make([]byte, 6) _, _ = rand.Read(b) return hex.EncodeToString(b) } // Configure checks, saves and applies the list of files. Files turned off, // removed or given another path are read again from FILES_BACKFILL ago the // next time they are on. func (f *Files) Configure(cfg filesConfig) error { if len(cfg.Sources) > maxFileSources { return fmt.Errorf("%d files at most", maxFileSources) } var srcs []*fileSource ids := map[string]bool{} for _, d := range cfg.Sources { if d.ID == "" { d.ID = newFileID() } if ids[d.ID] { return fmt.Errorf("duplicate id %s", d.ID) } ids[d.ID] = true src, err := compileFileSource(d) if err != nil { return err } srcs = append(srcs, src) } saved := filesConfig{Sources: []fileSourceDef{}} for _, s := range srcs { saved.Sources = append(saved.Sources, s.fileSourceDef) } if err := writeJSONFile(f.cfgPath, saved); err != nil { return err } f.mu.Lock() old := map[string]*fileSource{} for _, s := range f.cfg { old[s.ID] = s } for _, s := range srcs { o := old[s.ID] if o != nil && ((o.Enabled && !s.Enabled) || o.Path != s.Path) { f.resets[s.ID] = true delete(f.status, s.ID) } if o != nil && (o.Format != s.Format || o.Regex != s.Regex) { f.redetect[s.ID] = true } delete(old, s.ID) } for id := range old { // removed f.resets[id] = true delete(f.status, id) } f.cfg = srcs f.mu.Unlock() select { case f.wake <- struct{}{}: default: } return nil } // Status is the state shown in Settings > Sources. func (f *Files) Status() map[string]any { f.mu.Lock() defer f.mu.Unlock() list := make([]map[string]any, 0, len(f.cfg)) for _, s := range f.cfg { m := map[string]any{ "id": s.ID, "name": s.Name, "path": s.Path, "format": s.Format, "regex": s.Regex, "host": s.Host, "enabled": s.Enabled, "open": false, "read": 0, "detected": "", "code": "", "error": "", } if st := f.status[s.ID]; st != nil && s.Enabled { m["open"], m["read"], m["detected"], m["code"], m["error"] = st.Open, st.Read, st.Detected, st.Code, st.Error } list = append(list, m) } return map[string]any{"sources": list, "defaultHost": f.hostName} } // Run polls the files every second. func (f *Files) Run(ctx context.Context) { tick := time.NewTicker(time.Second) defer tick.Stop() n := 0 for { select { case <-ctx.Done(): now := time.Now() for id := range f.tails { f.flushPending(id, now) } f.checkpointAll() f.saveState() return case <-tick.C: case <-f.wake: } f.poll(time.Now()) if n++; n%5 == 0 { f.saveState() } } } func (f *Files) saveState() { if !f.dirty { return } f.dirty = false if err := writeJSONFile(f.statePath, f.st); err != nil { logEvent("files", 4, "saving positions: %v", err) } } // poll reads what was added to every file that is on. func (f *Files) poll(now time.Time) { f.mu.Lock() srcs := f.cfg resets, redetect := f.resets, f.redetect f.resets, f.redetect = map[string]bool{}, map[string]bool{} f.mu.Unlock() for id := range resets { delete(f.st.Pos, id) delete(f.st.Started, id) delete(f.tails, id) f.dirty = true } for id := range redetect { if tl := f.tails[id]; tl != nil { tl.auto = nil } } active := map[string]bool{} for _, s := range srcs { if !s.Enabled { continue } active[s.ID] = true f.setStatus(s.ID, f.pollFile(s, now)) } // Files turned off: their position is kept until the next poll forgets it. for id := range f.tails { if !active[id] { delete(f.tails, id) } } f.checkpointAll() } const ( maxFileRead = 4 << 20 // per file and per poll fileContinueDelay = 2 * time.Second // an entry waits this long for continuation lines ) func (f *Files) setStatus(id string, err error) { code, detail := "", "" if err != nil { code, detail = errCode(err), err.Error() if errors.Is(err, fs.ErrNotExist) { code = "not_found" } } detected := "" if tl := f.tails[id]; tl != nil && tl.auto != nil { detected = tl.auto.format } f.mu.Lock() st := f.status[id] if st == nil { st = &fileSourceStatus{} f.status[id] = st } changed := code != st.Code st.Open, st.Detected, st.Code, st.Error = err == nil, detected, code, detail name := id for _, s := range f.cfg { if s.ID == id { name = s.Name } } f.mu.Unlock() if changed && code != "" && code != "not_found" { logEvent("files", 4, "cannot read the file %s: %v", name, err) } } func (f *Files) count(id string) { f.mu.Lock() if st := f.status[id]; st != nil { st.Read++ } f.mu.Unlock() } // parserOf is the parser of a file: the chosen format, or the one detected // from its last lines (nil while the file has none). func (f *Files) parserOf(s *fileSource, tl *fileTail, path string) *fileParser { if s.parser != nil { return s.parser } if tl.auto == nil { lines, err := sampleLines(path) if err != nil || len(lines) == 0 { return nil } tl.auto, _ = newFileParser(detectFormat(lines), "") } return tl.auto } func (f *Files) pollFile(s *fileSource, now time.Time) error { path := s.Path first := false if _, ok := f.st.Started[s.ID]; !ok { f.st.Started[s.ID], f.dirty, first = now, true, true } fi, err := os.Stat(path) if err != nil { return err } if !fi.Mode().IsRegular() { return fmt.Errorf("%s is not a regular file", path) } tl := f.tails[s.ID] if tl == nil { tl = &fileTail{} f.tails[s.ID] = tl } ino := inodeOf(fi) pos, known := f.st.Pos[s.ID] if !known { pos = filePos{Ino: ino} if first { // Turned on: the last FILES_BACKFILL of what is already there // (0: only new lines). if f.backfill > 0 { pos.Until = fi.Size() } else { pos.Off = fi.Size() } } f.st.Pos[s.ID], f.dirty = pos, true } if !tl.known { tl.readOff, tl.known = pos.Off, true } if pos.Ino != ino { // Rotated (access.log -> access.log.1): finish the old file, then // read the new one from its start. if old, err := os.Stat(path + ".1"); err == nil && inodeOf(old) == pos.Ino { if p := f.parserOf(s, tl, path+".1"); p != nil { for { next, more, err := f.readFrom(s, tl, p, path+".1", tl.readOff, pos.Until, now) if err != nil || !more { break } tl.readOff = next } } } f.flushPending(s.ID, now) pos, tl.readOff = filePos{Ino: ino}, 0 f.st.Pos[s.ID], f.dirty = pos, true } else if fi.Size() < tl.readOff { f.flushPending(s.ID, now) // truncated (copytruncate) pos.Until, tl.readOff = 0, 0 f.st.Pos[s.ID], f.dirty = pos, true } if fi.Size() > tl.readOff { p := f.parserOf(s, tl, path) if p == nil { return nil // nothing complete to detect the format from yet } next, _, err := f.readFrom(s, tl, p, path, tl.readOff, pos.Until, now) if err != nil { return err } tl.readOff = next } if pe := tl.pending; pe != nil && now.Sub(pe.seen) >= fileContinueDelay { f.flushPending(s.ID, now) } return nil } // readFrom parses the complete lines written after off and returns the new // offset; more tells whether the read window was full. func (f *Files) readFrom(s *fileSource, tl *fileTail, p *fileParser, path string, off, until int64, now time.Time) (int64, bool, error) { fh, err := os.Open(path) if err != nil { return off, false, err } defer fh.Close() buf, err := io.ReadAll(io.NewSectionReader(fh, off, maxFileRead)) if err != nil { return off, false, err } full := len(buf) == maxFileRead end := bytes.LastIndexByte(buf, '\n') if end < 0 { if !full { return off, false, nil // partial line: wait for its end } end = len(buf) - 1 // line longer than the read window: cut it } start := off for _, b := range bytes.Split(buf[:end], []byte{'\n'}) { f.line(s, tl, p, string(b), start, start < until, now) start += int64(len(b)) + 1 } return off + int64(end) + 1, full, nil } // line handles one line: a continuation of the pending entry, or a new one. func (f *Files) line(s *fileSource, tl *fileTail, p *fileParser, line string, off int64, history bool, now time.Time) { if strings.TrimSpace(line) == "" { return } if pe := tl.pending; p.multiline() && pe != nil && p.continuation(line) { if pe.lines < maxContinuation && len(pe.rec.msg) < maxFileMsg { pe.rec.msg += "\n" + strings.TrimRight(cleanLine(line), " \t") pe.lines++ } pe.seen = now return } f.flushPending(s.ID, now) pe := &pendingEntry{rec: p.parse(line, now), off: off, seen: now, history: history} if p.multiline() { tl.pending = pe return } f.emit(s, pe, now) } func (f *Files) flushPending(id string, now time.Time) { tl := f.tails[id] if tl == nil || tl.pending == nil { return } pe := tl.pending tl.pending = nil f.mu.Lock() var src *fileSource for _, s := range f.cfg { if s.ID == id { src = s } } f.mu.Unlock() if src != nil { f.emit(src, pe, now) } } // emit sends an entry; history entries are kept only when dated within // FILES_BACKFILL of when the file was turned on. func (f *Files) emit(s *fileSource, pe *pendingEntry, now time.Time) { r := pe.rec if pe.history { notBefore := f.st.Started[s.ID].Add(-f.backfill) if !r.dated || r.t.Before(notBefore) { return } } if len(r.msg) > maxFileMsg { r.msg = strings.ToValidUTF8(r.msg[:maxFileMsg], "") + "…" } extra := map[string]string{"log_file": s.Path} for k, v := range r.fields { extra[k] = v } f.sink(&Entry{ Time: r.t, Received: now, Host: firstNonEmpty(r.host, s.Host, f.hostName), App: firstNonEmpty(r.app, s.Name), SevNum: r.sev, Severity: severityNames[r.sev], Message: r.msg, Proto: "file", SourceType: "file", Extra: extra, Wait: true, // the file keeps the lines: wait for room in the queue }) f.count(s.ID) } // checkpointAll saves, for each file, an offset before which every entry is // sent: the start of its pending entry, or the read offset. func (f *Files) checkpointAll() { for id, tl := range f.tails { pos, ok := f.st.Pos[id] if !ok || !tl.known { continue } off := tl.readOff if pe := tl.pending; pe != nil && pe.off < off { off = pe.off } if off != pos.Off { pos.Off, f.dirty = off, true f.st.Pos[id] = pos } } } // DetectFile tells, for the settings, whether a file can be read and its // format, with a line as example. func DetectFile(path string) map[string]any { res := map[string]any{"format": "", "sample": "", "code": "", "error": ""} p, err := checkFilePath(path) if err != nil { res["code"], res["error"] = "invalid", err.Error() return res } var lines []string if fi, err2 := os.Stat(p); err2 != nil { err = err2 } else if !fi.Mode().IsRegular() { err = fmt.Errorf("%s is not a regular file", p) } else { lines, err = sampleLines(p) } switch { case errors.Is(err, fs.ErrNotExist): res["code"], res["error"] = "not_found", err.Error() case err != nil: res["code"], res["error"] = errCode(err), err.Error() case len(lines) == 0: res["code"] = "empty" default: res["format"] = detectFormat(lines) sample := lines[len(lines)-1] if len(sample) > 300 { sample = strings.ToValidUTF8(sample[:300], "") + "…" } res["sample"] = cleanLine(sample) } return res }