Log files: find the history by date, count from the first poll

A file turned on was read from its start to find its last hour: with a
Traefik access log that holds months of lines (never rotated by
Traefik), LogStream spent minutes on old lines that were all left out,
and new lines waited behind them. Entries sent during the first poll
were also left out of the count shown in Settings.

- start of the FILES_BACKFILL history found by bisection on the dates
  of the lines; files without dates: only new lines
- up to 16 reads of 4 MB per file and per poll
- status "behind" (bytes left to read), shown as "Reading the history:
  N MB left"
- the status of a file exists before its first entries are counted

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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

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.

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 {
-				pos.Until = fi.Size()
-			} else {
+			// (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()
 			}
 		}
 		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)
+		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.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) {

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)
+	}
+}

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'];
 }