diff --git a/README.fr.md b/README.fr.md index 6d0505b..acf4f24 100644 --- a/README.fr.md +++ b/README.fr.md @@ -342,7 +342,9 @@ conteneur, un **format** (*Automatique* par défaut), un **hôte** facultatif (p les affiche seules. - À l'activation d'un fichier, les lignes de la dernière heure déjà présentes sont lues d'abord (`FILES_BACKFILL`, `0` = seulement les nouvelles lignes ; les lignes sans horodatage sont alors - ignorées) ; un fichier créé ensuite est lu depuis son début. L'inode et une + ignorées). Le début de cette heure est trouvé par dichotomie sur les dates des lignes : un gros + fichier qui contient des mois de logs n'est pas relu depuis son début ; un fichier créé ensuite + est lu depuis son début, jusqu'à 64 Mo par seconde, l'état indiquant ce qu'il reste à lire. L'inode et une position sont enregistrés par fichier dans `/data/files-state.json` : après un redémarrage la lecture reprend là ; après une rotation (`access.log` renommé `access.log.1`), la fin de l'ancien fichier est lue avant le nouveau ; un fichier vidé sur place (`copytruncate`) est diff --git a/README.md b/README.md index 7b928c9..4402c18 100644 --- a/README.md +++ b/README.md @@ -312,8 +312,10 @@ optional **host** (default: `/host/etc/hostname`, else `localhost`) and a switch - **Display**: source `file`, the path in `log_file`. In the list these lines have a green left edge, a light tint and a **FILE** badge before the app; the **Source** filter shows them alone. - When a file is turned on, the lines of the last hour already in it are read first - (`FILES_BACKFILL`, `0` = only new lines; lines without a timestamp are then left out); a file - created later is read from its start. The inode and an offset are saved per file in + (`FILES_BACKFILL`, `0` = only new lines; lines without a timestamp are then left out). The + start of that hour is found by bisection on the dates of the lines, so a large file that holds + months of logs is not read from its start; a file created later is read from its start, up to + 64 MB per second, the status showing what is left to read. The inode and an offset are saved per file in `/data/files-state.json`: after a restart reading resumes there; after a rotation (`access.log` renamed `access.log.1`), the end of the old file is read before the new one; a file truncated in place (`copytruncate`) is read again from its start. diff --git a/filesource.go b/filesource.go index fdecaa2..ee3820c 100644 --- a/filesource.go +++ b/filesource.go @@ -53,6 +53,7 @@ type fileSourceStatus struct { Open bool Read uint64 Detected string // format found when the source is in auto + Behind int64 // bytes left to read (history being read) Code string Error string } @@ -79,6 +80,7 @@ type fileTail struct { known bool // readOff is set pending *pendingEntry auto *fileParser // format detected from the file, in auto + behind int64 // bytes left to read after the last poll } // Files follows log files of other programs (Traefik, nginx, applications…) @@ -265,10 +267,11 @@ func (f *Files) Status() map[string]any { 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": "", + "host": s.Host, "enabled": s.Enabled, "open": false, "read": 0, "detected": "", "behind": 0, "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 + m["behind"] = st.Behind } list = append(list, m) } @@ -346,7 +349,8 @@ func (f *Files) poll(now time.Time) { } const ( - maxFileRead = 4 << 20 // per file and per poll + maxFileRead = 4 << 20 // per read + maxFileReads = 16 // reads per file and per poll (64 MB) fileContinueDelay = 2 * time.Second // an entry waits this long for continuation lines ) @@ -370,6 +374,10 @@ func (f *Files) setStatus(id string, err error) { } changed := code != st.Code st.Open, st.Detected, st.Code, st.Error = err == nil, detected, code, detail + st.Behind = 0 + if tl := f.tails[id]; tl != nil && err == nil { + st.Behind = tl.behind + } name := id for _, s := range f.cfg { if s.ID == id { @@ -384,9 +392,12 @@ func (f *Files) setStatus(id string, err error) { func (f *Files) count(id string) { f.mu.Lock() - if st := f.status[id]; st != nil { - st.Read++ + st := f.status[id] + if st == nil { // first poll of the file: its status comes after + st = &fileSourceStatus{} + f.status[id] = st } + st.Read++ f.mu.Unlock() } @@ -430,11 +441,12 @@ func (f *Files) pollFile(s *fileSource, now time.Time) error { 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 { + // (0: only new lines), found by date rather than read from the + // start of a file that can hold months of logs. + pos.Off = fi.Size() + if p := f.parserOf(s, tl, path); f.backfill > 0 && p != nil { + pos.Off = backfillStart(path, p, fi.Size(), f.st.Started[s.ID].Add(-f.backfill), now) pos.Until = fi.Size() - } else { - pos.Off = fi.Size() } } f.st.Pos[s.ID], f.dirty = pos, true @@ -469,18 +481,78 @@ func (f *Files) pollFile(s *fileSource, now time.Time) error { 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 + for i := 0; i < maxFileReads; i++ { + next, more, err := f.readFrom(s, tl, p, path, tl.readOff, pos.Until, now) + if err != nil { + return err + } + tl.readOff = next + if !more { + break + } } - tl.readOff = next } + tl.behind = max(fi.Size()-tl.readOff, 0) if pe := tl.pending; pe != nil && now.Sub(pe.seen) >= fileContinueDelay { f.flushPending(s.ID, now) } return nil } +// backfillStart finds where the lines dated notBefore or later start, by +// bisection on the dates of the lines (logs are written in order). Undated +// lines are left out of the history anyway: then only new lines are read. +func backfillStart(path string, p *fileParser, size int64, notBefore, now time.Time) int64 { + fh, err := os.Open(path) + if err != nil { + return size + } + defer fh.Close() + buf := make([]byte, 64<<10) + // lineAt reads the first line starting at or after off (off itself when + // it is 0): its start and its date. + lineAt := func(off int64) (int64, time.Time, bool) { + n, _ := fh.ReadAt(buf, off) + b := buf[:n] + start := off + if off > 0 { + nl := bytes.IndexByte(b, '\n') + if nl < 0 { + return size, time.Time{}, false + } + b, start = b[nl+1:], off+int64(nl)+1 + } + end := bytes.IndexByte(b, '\n') + if end < 0 { + return size, time.Time{}, false + } + r := p.parse(string(b[:end]), now) + return start, r.t, r.dated + } + if _, t, ok := lineAt(0); !ok || !t.Before(notBefore) { + if !ok { + // No date on the first line: try the last ones. + if _, _, ok := lineAt(max(size-int64(len(buf)), 0)); !ok { + return size + } + } else { + return 0 // all of it is recent + } + } + lo, hi := int64(0), size // line at lo: before notBefore; at hi: not + for hi-lo > int64(len(buf)) { + mid := lo + (hi-lo)/2 + _, t, ok := lineAt(mid) + if ok && t.Before(notBefore) { + lo = mid + } else { + hi = mid + } + } + start, _, _ := lineAt(lo) + return min(start, size) +} + // 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) { diff --git a/filesource_test.go b/filesource_test.go index 7bf370a..814a967 100644 --- a/filesource_test.go +++ b/filesource_test.go @@ -1,6 +1,7 @@ package main import ( + "fmt" "os" "path/filepath" "strings" @@ -382,3 +383,45 @@ func TestFilesAutoAndErrors(t *testing.T) { t.Fatalf("re-enabled: got %q", msgs(got)) } } + +// A large access log: the last hour is found by date, not read from the +// start, and every entry is counted from the first poll. +func TestFilesBackfillLargeFile(t *testing.T) { + dir, data := t.TempDir(), t.TempDir() + path := filepath.Join(dir, "access.log") + now := time.Now() + line := func(at time.Time, n int) string { + return fmt.Sprintf(`86.246.223.62 - - [%s] "GET /wp-content/x.png HTTP/2.0" 200 640365 "-" "-" %d "vblog-io-router@file" "http://wordpress" 3ms`+"\n", + at.UTC().Format("02/Jan/2006:15:04:05 -0700"), n) + } + var b strings.Builder + const n = 200000 // one line every 3 s for a week: ~32 MB + for i := 0; i < n; i++ { + b.WriteString(line(now.Add(-time.Duration(n-i)*3*time.Second), i)) + } + appendFile(t, path, b.String()) + sink := &fileSink{} + f := NewFiles(dir, data, time.Hour, sink.add) + f.Configure(filesConfig{Sources: []fileSourceDef{{ID: "a1", Name: "traefik-log", Path: path, Enabled: true}}}) + f.poll(now) + if pos := f.st.Pos["a1"]; pos.Off < int64(b.Len())*99/100 { + t.Errorf("history read from offset %d of %d", pos.Off, b.Len()) + } + appendFile(t, path, line(now, n)) + f.poll(now.Add(time.Second)) + got := sink.take() + // The last hour (1200 lines, the oldest one on the edge) and the new line. + if len(got) < 1200 || len(got) > 1201 || got[0].App != "traefik-log" || got[len(got)-1].Extra["http_status"] != "200" { + t.Fatalf("got %d entries", len(got)) + } + if st := f.Status()["sources"].([]map[string]any)[0]; st["read"] != uint64(len(got)) || st["behind"] != int64(0) || st["detected"] != fmtCLF { + t.Errorf("status %+v", st) + } + // Undated plain text: only new lines. + txt := filepath.Join(dir, "app.log") + appendFile(t, txt, strings.Repeat("no date here\n", 1000)) + p := mustParser(t, fmtText, "") + if off := backfillStart(txt, p, 13000, now.Add(-time.Hour), now); off != 13000 { + t.Errorf("undated: %d", off) + } +} diff --git a/web/app.js b/web/app.js index e4d7935..f25a815 100644 --- a/web/app.js +++ b/web/app.js @@ -101,6 +101,7 @@ const I18N = { fileDraft: 'Fill in the name and the path to save this file.', fileOk: (r) => `Followed: ${r} entries since startup.`, fileDetected: (f) => ` Format detected: ${f}.`, + fileBehind: (mb) => ` Reading the history: ${mb} MB left.`, probeFound: (f) => `File found, format detected: ${f}.`, probeSample: 'Last line: ', file_empty: 'The file is empty for now: its format will be detected from its first lines.', @@ -332,6 +333,7 @@ const I18N = { fileDraft: 'Renseignez le nom et le chemin pour enregistrer ce fichier.', fileOk: (r) => `Suivi : ${r} entrées depuis le démarrage.`, fileDetected: (f) => ` Format détecté : ${f}.`, + fileBehind: (mb) => ` Lecture de l\'historique : ${mb} Mo restants.`, probeFound: (f) => `Fichier trouvé, format détecté : ${f}.`, probeSample: 'Dernière ligne : ', file_empty: 'Le fichier est vide pour l\'instant : son format sera détecté à partir de ses premières lignes.', @@ -2279,7 +2281,10 @@ function fileStatus(s) { if (s.draft) return [t('fileDraft'), 'muted']; if (!s.enabled) return [t('fileOff'), 'muted']; if (s.code) return [t('file_' + s.code) + (s.code === 'not_found' ? '' : s.error || ''), 'bad']; - if (s.open) return [t('fileOk', fmtNum(s.read)) + (s.format === 'auto' && s.detected ? t('fileDetected', t('fmt_' + s.detected)) : ''), 'good']; + if (s.open) { + return [t('fileOk', fmtNum(s.read)) + (s.format === 'auto' && s.detected ? t('fileDetected', t('fmt_' + s.detected)) : '') + + (s.behind > 1 << 20 ? t('fileBehind', fmtNum(Math.round(s.behind / (1 << 20)))) : ''), 'good']; + } return [t('hostWaiting'), 'muted']; }