vLab-BZH/logstream
2
0 Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Log files: one file per entry, format detection

Browse Source 
- a path is one file (wildcards refused for now); the name defaults to
  the file name
- format "auto" (the default): detected from the last 50 lines of the
  file, the structured format followed by at least 80 % of them (Traefik
  access JSON, JSON lines, CLF, logfmt), else plain text; done again when
  the format changes, shown in the status ("Format detected: …"); any
  other format forces it
- POST /api/files/detect {path}: found / permission / empty, detected
  format and last line, shown under the row as soon as the path is typed
  (POST so that read-only accounts cannot read files through it)
- README (EN/FR) updated

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
cedricandClaude Opus 5.5 committed 2026-10-05 10:33:26 +02:00
1 parent 087d9e7b30
commit dcb09e619e
8 files changed
+488 -284
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

filesource.go
+201 -222
View File
@@ -13,18 +13,17 @@ import (
"os"
"path/filepath"
"regexp"
"sort"
"strings"
"sync"
"time"
)
// fileSourceDef is one log file (or glob) declared in Settings > Sources.
// 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"` // file, or glob (/logs/traefik/*.log)
Format string `json:"format"` // text, json, logfmt, regex, clf, traefik
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"`
@@ -46,18 +45,20 @@ type filePos struct {
}
type filesState struct {
Pos map[string]filePos `json:"pos"` // key: source id + "|" + path
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 {
Files int
Read uint64
Code string
Error string
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.
// fileSource is a definition with its parser (nil in auto: detected from
// the file).
type fileSource struct {
fileSourceDef
parser *fileParser
@@ -72,9 +73,17 @@ type pendingEntry struct {
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 chosen for each
// file. Source type "file".
// 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
@@ -84,15 +93,15 @@ type Files struct {
wake chan struct{}
// Owned by the Run goroutine.
st filesState
readOff map[string]int64
pending map[string]*pendingEntry
dirty bool
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
status map[string]*fileSourceStatus
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 {
@@ -103,9 +112,9 @@ func NewFiles(hostRoot, dataDir string, backfill time.Duration, sink func(*Entry
cfgPath: filepath.Join(dataDir, "files.json"),
statePath: filepath.Join(dataDir, "files-state.json"),
wake: make(chan struct{}, 1),
readOff: map[string]int64{},
pending: map[string]*pendingEntry{},
tails: map[string]*fileTail{},
resets: map[string]bool{},
redetect: map[string]bool{},
status: map[string]*fileSourceStatus{},
}
if f.hostName == "" {
@@ -139,13 +148,24 @@ 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.Path = strings.TrimSpace(d.Path)
d.Host = strings.TrimSpace(d.Host)
if d.Format == "" {
d.Format = fmtText
d.Format = fmtAuto
}
if d.Format != fmtRegex {
d.Regex = ""
@@ -155,20 +175,21 @@ func compileFileSource(d fileSourceDef) (*fileSource, error) {
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 !filepath.IsAbs(d.Path):
return nil, fmt.Errorf("%s: the path must be absolute (as mounted in the container)", d.Name)
case len(d.Host) > 253:
return nil, fmt.Errorf("%s: host name too long", d.Name)
}
d.Path = filepath.Clean(d.Path)
if _, err := filepath.Match(d.Path, ""); err != nil {
return nil, fmt.Errorf("%s: invalid path pattern: %v", d.Name, err)
}
p, err := newFileParser(d.Format, d.Regex)
p, err := checkFilePath(d.Path)
if err != nil {
return nil, fmt.Errorf("%s: %v", d.Name, err)
}
return &fileSource{fileSourceDef: d, parser: p}, nil
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 {
@@ -177,7 +198,7 @@ func newFileID() string {
return hex.EncodeToString(b)
}
// Configure checks, saves and applies the list of files. Sources turned off,
// 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 {
@@ -213,10 +234,14 @@ func (f *Files) Configure(cfg filesConfig) error {
old[s.ID] = s
}
for _, s := range srcs {
if o := old[s.ID]; o != nil && ((o.Enabled && !s.Enabled) || o.Path != s.Path) {
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
@@ -240,10 +265,10 @@ 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, "files": 0, "read": 0, "code": "", "error": "",
"host": s.Host, "enabled": s.Enabled, "open": false, "read": 0, "detected": "", "code": "", "error": "",
}
if st := f.status[s.ID]; st != nil && s.Enabled {
m["files"], m["read"], m["code"], m["error"] = st.Files, st.Read, st.Code, st.Error
m["open"], m["read"], m["detected"], m["code"], m["error"] = st.Open, st.Read, st.Detected, st.Code, st.Error
}
list = append(list, m)
}
@@ -259,8 +284,8 @@ func (f *Files) Run(ctx context.Context) {
select {
case <-ctx.Done():
now := time.Now()
for key := range f.pending {
f.flushPending(key, now)
for id := range f.tails {
f.flushPending(id, now)
}
f.checkpointAll()
f.saveState()
@@ -285,31 +310,23 @@ func (f *Files) saveState() {
}
}
func posKey(id, path string) string { return id + "|" + path }
// forget drops the positions of a source.
func (f *Files) forget(id string) {
prefix := id + "|"
for key := range f.st.Pos {
if strings.HasPrefix(key, prefix) {
delete(f.st.Pos, key)
delete(f.readOff, key)
delete(f.pending, key)
}
}
delete(f.st.Started, id)
f.dirty = true
}
// poll reads what was added to the files of every source that is on.
// 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 := f.resets
f.resets = map[string]bool{}
resets, redetect := f.resets, f.redetect
f.resets, f.redetect = map[string]bool{}, map[string]bool{}
f.mu.Unlock()
for id := range resets {
f.forget(id)
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 {
@@ -317,69 +334,23 @@ func (f *Files) poll(now time.Time) {
continue
}
active[s.ID] = true
f.pollSource(s, now)
f.setStatus(s.ID, f.pollFile(s, now))
}
// Sources removed or off: nothing kept but their positions (removed ones
// were forgotten above).
for key := range f.pending {
id, _, _ := strings.Cut(key, "|")
// Files turned off: their position is kept until the next poll forgets it.
for id := range f.tails {
if !active[id] {
delete(f.pending, key)
delete(f.tails, id)
}
}
f.checkpointAll()
}
const (
maxFileRead = 4 << 20 // per file and per poll
maxFilesPerGlob = 200
maxFileRead = 4 << 20 // per file and per poll
fileContinueDelay = 2 * time.Second // an entry waits this long for continuation lines
)
// rotatedName: copies left by logrotate (access.log.1, app.log.2.gz,
// app-20261005.log.gz…), not read when a glob matches them.
var rotatedName = regexp.MustCompile(`(\.\d+|\.gz|\.bz2|\.xz|\.zst|\.zip|\.lz4)$`)
// matches lists the files of a source.
func (s *fileSource) matches() ([]string, error) {
if !strings.ContainsAny(s.Path, "*?[") {
fi, err := os.Stat(s.Path)
if err != nil {
return nil, err
}
if !fi.Mode().IsRegular() {
return nil, fmt.Errorf("%s is not a regular file", s.Path)
}
return []string{s.Path}, nil
}
paths, err := filepath.Glob(s.Path)
if err != nil {
return nil, err
}
var out []string
for _, p := range paths {
if rotatedName.MatchString(p) {
continue
}
if fi, err := os.Stat(p); err == nil && fi.Mode().IsRegular() {
out = append(out, p)
}
}
sort.Strings(out)
if len(out) > maxFilesPerGlob {
out = out[:maxFilesPerGlob]
}
if len(out) == 0 {
// Tell a missing directory (not mounted) from an empty one.
if _, err := os.Stat(filepath.Dir(s.Path)); err != nil && !strings.ContainsAny(filepath.Dir(s.Path), "*?[") {
return nil, err
}
return nil, fs.ErrNotExist
}
return out, nil
}
func (f *Files) setStatus(id string, files int, err error) {
func (f *Files) setStatus(id string, err error) {
code, detail := "", ""
if err != nil {
code, detail = errCode(err), err.Error()
@@ -387,6 +358,10 @@ func (f *Files) setStatus(id string, files int, err error) {
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 {
@@ -394,7 +369,7 @@ func (f *Files) setStatus(id string, files int, err error) {
f.status[id] = st
}
changed := code != st.Code
st.Files, st.Code, st.Error = files, code, detail
st.Open, st.Detected, st.Code, st.Error = err == nil, detected, code, detail
name := id
for _, s := range f.cfg {
if s.ID == id {
@@ -415,130 +390,100 @@ func (f *Files) count(id string) {
f.mu.Unlock()
}
func (f *Files) pollSource(s *fileSource, now time.Time) {
// 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
}
paths, err := s.matches()
if err != nil {
f.setStatus(s.ID, 0, err)
return
}
prefix := s.ID + "|"
matched := map[string]bool{}
for _, p := range paths {
matched[posKey(s.ID, p)] = true
}
// Files that left the glob, by inode: a file renamed within it goes on
// from where it was.
gone := map[uint64]string{}
for key, pos := range f.st.Pos {
if strings.HasPrefix(key, prefix) && !matched[key] {
gone[pos.Ino] = key
}
}
var firstErr error
read := 0
for _, p := range paths {
if err := f.pollFile(s, p, first, gone, now); err != nil {
firstErr = keepFirst(firstErr, err)
continue
}
read++
}
for key := range f.st.Pos {
if strings.HasPrefix(key, prefix) && !matched[key] {
f.flushPending(key, now)
delete(f.st.Pos, key)
delete(f.readOff, key)
f.dirty = true
}
}
if read > 0 {
firstErr = nil
}
f.setStatus(s.ID, read, firstErr)
}
func (f *Files) pollFile(s *fileSource, path string, first bool, gone map[uint64]string, now time.Time) error {
key := posKey(s.ID, path)
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[key]
pos, known := f.st.Pos[s.ID]
if !known {
if old, ok := gone[ino]; ok && ino != 0 {
// Renamed within the glob.
pos, known = f.st.Pos[old], true
if off, ok := f.readOff[old]; ok {
f.readOff[key] = off
}
if pe := f.pending[old]; pe != nil {
f.pending[key] = pe
}
delete(f.st.Pos, old)
delete(f.readOff, old)
delete(f.pending, old)
delete(gone, ino)
} else if first {
// Turned on: the last FILES_BACKFILL of the files already there
pos = filePos{Ino: ino}
if first {
// Turned on: the last FILES_BACKFILL of what is already there
// (0: only new lines).
pos = filePos{Ino: ino}
if f.backfill > 0 {
pos.Until = fi.Size()
} else {
pos.Off = fi.Size()
}
} else {
pos = filePos{Ino: ino} // new file: read from its start
}
f.st.Pos[key], f.dirty = pos, true
f.st.Pos[s.ID], f.dirty = pos, true
}
off, ok := f.readOff[key]
if !ok {
off = pos.Off
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 {
for {
next, more, err := f.readFrom(s, key, path+".1", off, pos.Until, now)
if err != nil || !more {
break
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
}
off = next
}
}
f.flushPending(key, now)
pos, off = filePos{Ino: ino}, 0
f.st.Pos[key], f.dirty = pos, true
} else if fi.Size() < off {
f.flushPending(key, now) // truncated (copytruncate)
pos.Until, off = 0, 0
f.st.Pos[key], f.dirty = pos, true
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() > off {
next, _, err := f.readFrom(s, key, path, off, pos.Until, now)
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 {
f.readOff[key] = off
return err
}
off = next
tl.readOff = next
}
f.readOff[key] = off
if pe := f.pending[key]; pe != nil && now.Sub(pe.seen) >= fileContinueDelay {
f.flushPending(key, now)
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, key, path string, off, until int64, now time.Time) (int64, bool, error) {
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
@@ -558,19 +503,18 @@ func (f *Files) readFrom(s *fileSource, key, path string, off, until int64, now
}
start := off
for _, b := range bytes.Split(buf[:end], []byte{'\n'}) {
f.line(s, key, path, string(b), start, start < until, now)
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, key, path, line string, off int64, history bool, now time.Time) {
func (f *Files) line(s *fileSource, tl *fileTail, p *fileParser, line string, off int64, history bool, now time.Time) {
if strings.TrimSpace(line) == "" {
return
}
pe := f.pending[key]
if s.parser.multiline() && pe != nil && s.parser.continuation(line) {
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++
@@ -578,23 +522,22 @@ func (f *Files) line(s *fileSource, key, path, line string, off int64, history b
pe.seen = now
return
}
f.flushPending(key, now)
rec := s.parser.parse(line, now)
pe = &pendingEntry{rec: rec, off: off, seen: now, history: history}
if s.parser.multiline() {
f.pending[key] = pe
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, path, pe, now)
f.emit(s, pe, now)
}
func (f *Files) flushPending(key string, now time.Time) {
pe := f.pending[key]
if pe == nil {
func (f *Files) flushPending(id string, now time.Time) {
tl := f.tails[id]
if tl == nil || tl.pending == nil {
return
}
delete(f.pending, key)
id, path, _ := strings.Cut(key, "|")
pe := tl.pending
tl.pending = nil
f.mu.Lock()
var src *fileSource
for _, s := range f.cfg {
@@ -604,13 +547,13 @@ func (f *Files) flushPending(key string, now time.Time) {
}
f.mu.Unlock()
if src != nil {
f.emit(src, path, pe, now)
f.emit(src, pe, now)
}
}
// emit sends an entry; history entries are kept only when dated within
// FILES_BACKFILL of when the source was turned on.
func (f *Files) emit(s *fileSource, path string, pe *pendingEntry, now time.Time) {
// 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)
@@ -621,7 +564,7 @@ func (f *Files) emit(s *fileSource, path string, pe *pendingEntry, now time.Time
if len(r.msg) > maxFileMsg {
r.msg = strings.ToValidUTF8(r.msg[:maxFileMsg], "") + "…"
}
extra := map[string]string{"log_file": path}
extra := map[string]string{"log_file": s.Path}
for k, v := range r.fields {
extra[k] = v
}
@@ -644,17 +587,53 @@ func (f *Files) emit(s *fileSource, path string, pe *pendingEntry, now time.Time
// 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 key, off := range f.readOff {
pos, ok := f.st.Pos[key]
if !ok {
for id, tl := range f.tails {
pos, ok := f.st.Pos[id]
if !ok || !tl.known {
continue
}
if pe := f.pending[key]; pe != nil && pe.off < off {
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[key] = pos
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
}