From dcb09e619eddc18b6c528999a6327ac95aafc617 Mon Sep 17 00:00:00 2001 From: Cedric Date: Mon, 5 Oct 2026 10:33:26 +0200 Subject: [PATCH] Log files: one file per entry, format detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- README.fr.md | 19 +- README.md | 22 ++- api.go | 17 +- fileformats.go | 100 +++++++++++ filesource.go | 423 +++++++++++++++++++++------------------------ filesource_test.go | 107 ++++++++---- web/app.js | 77 +++++++-- web/style.css | 7 +- 8 files changed, 488 insertions(+), 284 deletions(-) diff --git a/README.fr.md b/README.fr.md index e114a3c..2aa181f 100644 --- a/README.fr.md +++ b/README.fr.md @@ -294,17 +294,22 @@ source est **désactivée par défaut** : activez-la dans **Réglages > Sources Logstream peut suivre les fichiers de logs d'autres programmes : Traefik, nginx, Apache, ou toute application qui écrit dans un fichier. Déclarez chaque fichier dans **Paramètres > Sources de logs > Fichiers de logs** (enregistré dans `/data/files.json`, 50 au plus) : un **nom** (l'**app** -de ses entrées), le **chemin** tel que monté dans le conteneur, un **format**, un **hôte** -facultatif (par défaut `/host/etc/hostname`, sinon `localhost`) et un interrupteur. +de ses entrées, par défaut le nom du fichier), le **chemin** du fichier tel que monté dans le +conteneur, un **format** (*Automatique* par défaut), un **hôte** facultatif (par défaut +`/host/etc/hostname`, sinon `localhost`) et un interrupteur. - **Montage** du dossier en lecture seule, par ex. dans `docker-compose.yml` : `- /var/log/traefik:/logs/traefik:ro`, puis `docker compose up -d`. Les fichiers de `/var/log` sont déjà accessibles sous `/host/var/log/…`. Les fichiers doivent être lisibles par l'UID `10001` ou le groupe de `HOST_LOGS_GID` (`adm` par défaut) ; Paramètres > Sources de logs indique quand aucun fichier ne correspond ou que l'accès est refusé. -- **Chemin** : un fichier, ou un motif avec `*`, `?` ou `[…]` : `/logs/traefik/*.log` suit tous - les fichiers `.log` du dossier, y compris ceux créés plus tard (200 fichiers au plus par - motif). Les copies de rotation (`.1`, `.2.gz`…) sont ignorées. +- **Chemin** : un fichier, par ex. `/logs/traefik/access.log` (pas de jokers). Dès la saisie, le + fichier est vérifié : trouvé ou non, lisible ou non, son format détecté et sa dernière ligne. +- **Détection du format** (*Automatique*) : à partir des dernières lignes du fichier (50 au + plus), le format structuré que suivent au moins 80 % d'entre elles — accès Traefik JSON, JSON + lines, journal d'accès CLF, logfmt — sinon texte brut. Elle est refaite quand le format est + changé ; le format trouvé s'affiche dans l'état du fichier. Choisissez un autre format pour + le forcer (une expression régulière n'est jamais détectée). - **Formats** : - *Texte brut* : une entrée par ligne ; le niveau est détecté comme pour les logs Docker, un horodatage en début de ligne (ISO 8601, `2006/01/02 15:04:05`, `[02/Jan/2006:15:04:05 +0200]`, @@ -337,7 +342,7 @@ facultatif (par défaut `/host/etc/hostname`, sinon `localhost`) et un interrupt 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). Les fichiers qui apparaissent ensuite sont lus depuis leur début. L'inode et une + ignorées) ; un fichier créé ensuite est lu depuis son début. 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 @@ -620,7 +625,7 @@ Pour mettre à jour l'une d'elles : | `hostlogs.go`, `journal.go` | logs système de l'hôte : lecteur du journal systemd (sans `journalctl`) et suivi de `/var/log` | | `dmesg.go` | messages du noyau : lecture de `/dev/kmsg`, analyse des enregistrements, position par identifiant de démarrage et numéro de séquence | | `audit.go` | audit Linux : suivi de `audit.log` (rotation, position enregistrée), analyse des enregistrements, regroupement par événement, résumé | -| `filesource.go`, `fileformats.go` | fichiers de logs : suivi (motifs, rotation, positions enregistrées, lignes de suite) et formats (texte, JSON, logfmt, CLF, Traefik, expression régulière) | +| `filesource.go`, `fileformats.go` | fichiers de logs : suivi (rotation, positions enregistrées, lignes de suite) et formats (détection, texte, JSON, logfmt, CLF, Traefik, expression régulière) | | `selflog.go` | logs internes de LogStream : `logEvent`, niveau minimal, limitation des répétitions, file bornée | | `tags.go` | stockage des tags de couleur | | `presets.go`, `presets.json` | préréglages de tags (`/api/presets`), liste intégrée | diff --git a/README.md b/README.md index 3d77a7f..1b7d7e3 100644 --- a/README.md +++ b/README.md @@ -269,18 +269,22 @@ also be changed (saved in `/data/audit.json`; `AUDIT_LOG` sets the default path) Logstream can follow the log files of other programs: Traefik, nginx, Apache, or any application that writes to a file. Declare each file in **Settings > Log sources > Log files** (saved in -`/data/files.json`, up to 50): a **name** (the **app** of its entries), the **path** as mounted in -the container, a **format**, an optional **host** (default: `/host/etc/hostname`, else -`localhost`) and a switch. +`/data/files.json`, up to 50): a **name** (the **app** of its entries, by default the file name), +the **path** of the file as mounted in the container, a **format** (*Automatic* by default), an +optional **host** (default: `/host/etc/hostname`, else `localhost`) and a switch. - **Mount** the directory read-only, e.g. in `docker-compose.yml`: `- /var/log/traefik:/logs/traefik:ro`, then `docker compose up -d`. Files under `/var/log` are already reachable as `/host/var/log/…`. The files must be readable by UID `10001` or the group of `HOST_LOGS_GID` (`adm` by default); Settings > Log sources says when no file matches or access is denied. -- **Path**: a file, or a pattern with `*`, `?` or `[…]`: `/logs/traefik/*.log` follows every - `.log` file of the directory, also those created later (200 files at most per pattern). - Rotated copies (`.1`, `.2.gz`…) are left out. +- **Path**: one file, e.g. `/logs/traefik/access.log` (no wildcards). As soon as it is typed, the + file is checked: found or not, readable or not, its detected format and its last line. +- **Format detection** (*Automatic*): from the last lines of the file (50 at most), the + structured format that at least 80 % of them follow — Traefik access JSON, JSON lines, access + log CLF, logfmt — else plain text. It is done again when the format is changed; the format + found is shown in the status of the file. Choose another format to force it (a regular + expression is never detected). - **Formats**: - *Plain text*: one entry per line; the level is detected as for Docker logs, a timestamp at the start of the line (ISO 8601, `2006/01/02 15:04:05`, `[02/Jan/2006:15:04:05 +0200]`, @@ -308,8 +312,8 @@ the container, a **format**, an optional **host** (default: `/host/etc/hostname` - **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). Files - that appear later are read from their 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); a file + created later is read from its start. 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. @@ -570,7 +574,7 @@ To update one of them: | `hostlogs.go`, `journal.go` | host system logs: systemd journal reader (no `journalctl`) and `/var/log` follower | | `dmesg.go` | kernel messages: `/dev/kmsg` reader, record parsing, position by boot ID and sequence number | | `audit.go` | Linux audit: `audit.log` follower (rotation, saved offset), record parsing, grouping by event, summary | -| `filesource.go`, `fileformats.go` | log files: follower (patterns, rotation, saved offsets, continuation lines) and formats (text, JSON, logfmt, CLF, Traefik, regular expression) | +| `filesource.go`, `fileformats.go` | log files: follower (rotation, saved offsets, continuation lines) and formats (detection, text, JSON, logfmt, CLF, Traefik, regular expression) | | `selflog.go` | LogStream internal logs: `logEvent`, minimum level, throttling, bounded queue | | `tags.go` | color tag storage | | `presets.go`, `presets.json` | color tag presets (`/api/presets`), built-in list | diff --git a/api.go b/api.go index ad6aa46..d961e35 100644 --- a/api.go +++ b/api.go @@ -59,6 +59,7 @@ func (a *API) Routes(mux *http.ServeMux) { mux.HandleFunc("PUT /api/audit", a.auditConfigure) mux.HandleFunc("GET /api/files", a.filesStatus) mux.HandleFunc("PUT /api/files", a.filesConfigure) + mux.HandleFunc("POST /api/files/detect", a.filesDetect) mux.HandleFunc("GET /api/selflog", a.selfLogStatus) mux.HandleFunc("PUT /api/selflog", a.selfLogConfigure) } @@ -88,7 +89,7 @@ func (a *API) filesStatus(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, a.files.Status()) } -// PUT /api/files {"sources": [{"id", "name", "path", "format", "regex", "host", "enabled"}]} +// PUT /api/files {"sources": [{"id", "name", "path", "format" (auto or a format), "regex", "host", "enabled"}]} // (no id: a new file). func (a *API) filesConfigure(w http.ResponseWriter, r *http.Request) { var cfg filesConfig @@ -104,6 +105,20 @@ func (a *API) filesConfigure(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, a.files.Status()) } +// POST /api/files/detect {"path": "/logs/traefik/access.log"}: whether the file +// can be read, its detected format and its last line (Settings > Sources). A +// POST, so that read-only accounts cannot read files through it. +func (a *API) filesDetect(w http.ResponseWriter, r *http.Request) { + var req struct { + Path string `json:"path"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&req); err != nil { + writeErr(w, http.StatusBadRequest, err) + return + } + writeJSON(w, http.StatusOK, DetectFile(req.Path)) +} + // GET /api/selflog: state of the LogStream internal logs (Settings > Sources). func (a *API) selfLogStatus(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, a.selfLog.Status()) diff --git a/fileformats.go b/fileformats.go index c746f7a..67ef8f3 100644 --- a/fileformats.go +++ b/fileformats.go @@ -1,8 +1,11 @@ package main import ( + "bytes" "encoding/json" "fmt" + "io" + "os" "regexp" "strconv" "strings" @@ -12,6 +15,7 @@ import ( // Formats of the "Files" source (Settings > Sources), one per declared file. const ( + fmtAuto = "auto" // detected from the last lines of the file fmtText = "text" // one line per entry, level and leading timestamp detected fmtJSON = "json" // JSON lines: msg/time/level recognized, other keys as fields fmtLogfmt = "logfmt" // key=value pairs (Traefik "common" logs, Go programs…) @@ -587,3 +591,99 @@ func traefikRec(obj map[string]any, line string, now time.Time) *fileRec { r.msg = b.String() return r } + +/* ---------- Detection ---------- */ + +const detectWindow = 64 << 10 + +// sampleLines returns the last complete lines of a file (50 at most). +func sampleLines(path string) ([]string, error) { + fh, err := os.Open(path) + if err != nil { + return nil, err + } + defer fh.Close() + fi, err := fh.Stat() + if err != nil { + return nil, err + } + off := fi.Size() - detectWindow + if off < 0 { + off = 0 + } + buf, err := io.ReadAll(io.NewSectionReader(fh, off, detectWindow)) + if err != nil { + return nil, err + } + end := bytes.LastIndexByte(buf, '\n') + if end < 0 { + return nil, nil + } + buf = buf[:end] + if off > 0 { // first line cut by the window + if nl := bytes.IndexByte(buf, '\n'); nl >= 0 { + buf = buf[nl+1:] + } else { + return nil, nil + } + } + var lines []string + for _, b := range bytes.Split(buf, []byte{'\n'}) { + if len(bytes.TrimSpace(b)) > 0 { + lines = append(lines, string(bytes.TrimRight(b, "\r"))) + } + } + if len(lines) > 50 { + lines = lines[len(lines)-50:] + } + return lines, nil +} + +// logfmtKeys: the pairs have one of the usual logfmt keys. +func logfmtKeys(kv [][2]string) bool { + for _, p := range kv { + switch strings.ToLower(p[0]) { + case "level", "lvl", "msg", "message", "time", "ts", "t": + return true + } + } + return false +} + +// detectFormat guesses the format of a file from some of its lines: the +// structured format most lines follow, else plain text (a regular +// expression is never guessed). +func detectFormat(lines []string) string { + var js, traefik, clf, logfmt int + for _, l := range lines { + l = strings.TrimSpace(l) + if obj, ok := decodeJSONLine(l); ok { + js++ + if _, ok := obj["RequestMethod"]; ok { + traefik++ + } else if _, ok := obj["DownstreamStatus"]; ok { + traefik++ + } + continue + } + if clfRe.MatchString(l) { + clf++ + continue + } + if kv := parseLogfmt(l); len(kv) >= 3 && logfmtKeys(kv) { + logfmt++ + } + } + most := func(n int) bool { return n > 0 && n*5 >= len(lines)*4 } // 80 % + switch { + case most(js) && traefik*2 >= js: + return fmtTraefik + case most(js): + return fmtJSON + case most(clf): + return fmtCLF + case most(logfmt): + return fmtLogfmt + } + return fmtText +} diff --git a/filesource.go b/filesource.go index 1e67ab9..fdecaa2 100644 --- a/filesource.go +++ b/filesource.go @@ -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 +} diff --git a/filesource_test.go b/filesource_test.go index f4e3a0d..7bf370a 100644 --- a/filesource_test.go +++ b/filesource_test.go @@ -3,7 +3,6 @@ package main import ( "os" "path/filepath" - "sort" "strings" "sync" "testing" @@ -163,12 +162,58 @@ func TestParseTraefikJSON(t *testing.T) { } } +func TestDetectFormat(t *testing.T) { + tests := map[string][]string{ + fmtJSON: {`{"level":"info","msg":"a"}`, `{"level":"warn","msg":"b"}`}, + fmtTraefik: {`{"RequestMethod":"GET","DownstreamStatus":200}`, `{"level":"info","msg":"Configuration loaded"}`}, + fmtCLF: {`10.0.0.1 - - [05/Oct/2026:10:00:00 +0000] "GET / HTTP/1.1" 200 12 "-" "curl"`}, + fmtLogfmt: {`time="2026-10-05T10:00:00Z" level=info msg="Starting provider" providerName=docker`}, + fmtText: {`2026-10-05 10:00:00 INFO started`, `user=bob logged in`}, + } + for want, lines := range tests { + if got := detectFormat(lines); got != want { + t.Errorf("%q: got %s, want %s", lines, got, want) + } + } + // A few lines of another format do not change the answer (80 %). + lines := []string{`{"a":1}`, `{"a":2}`, `{"a":3}`, `{"a":4}`, `not json`} + if got := detectFormat(lines); got != fmtJSON { + t.Errorf("got %s", got) + } + lines = append(lines, `still not json`) + if got := detectFormat(lines); got != fmtText { + t.Errorf("got %s", got) + } +} + +func TestDetectFile(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "access.log") + appendFile(t, path, `10.0.0.1 - - [05/Oct/2026:10:00:00 +0000] "GET / HTTP/1.1" 200 12 "-" "curl"`+"\n"+`partial`) + res := DetectFile(path) + if res["format"] != fmtCLF || !strings.HasPrefix(res["sample"].(string), "10.0.0.1 - -") || res["code"] != "" { + t.Errorf("got %+v", res) + } + for p, code := range map[string]string{ + filepath.Join(dir, "nope.log"): "not_found", "relative.log": "invalid", filepath.Join(dir, "*.log"): "invalid", dir: "read", + } { + if res := DetectFile(p); res["code"] != code { + t.Errorf("%s: got %+v, want %s", p, res, code) + } + } + empty := filepath.Join(dir, "empty.log") + appendFile(t, empty, "") + if res := DetectFile(empty); res["code"] != "empty" { + t.Errorf("got %+v", res) + } +} + func TestFileSourceValidation(t *testing.T) { f := NewFiles(t.TempDir(), t.TempDir(), time.Hour, func(*Entry) {}) bad := []fileSourceDef{ {Name: "", Path: "/logs/a.log"}, {Name: "a", Path: "relative.log"}, - {Name: "a", Path: "/logs/[.log"}, + {Name: "a", Path: "/logs/*.log"}, {Name: "a", Path: "/logs/a.log", Format: "xml"}, {Name: "a", Path: "/logs/a.log", Format: fmtRegex, Regex: "(?P\\S+ \\S+) \\[(?P\\w+)\\] (?P.*)', fileEnabledAria: 'Follow this file', + fmt_auto: 'Automatic', fmt_text: 'Plain text', fmt_json: 'JSON lines', fmt_logfmt: 'logfmt (key=value)', @@ -95,11 +96,16 @@ const I18N = { fmt_regex: 'Regular expression', fileOff: 'Off.', fileDraft: 'Fill in the name and the path to save this file.', - fileOk: ({ f, r }) => `${f} file${f > 1 ? 's' : ''} followed: ${r} entries since startup.`, - file_not_found: 'No file at this path: mount its directory in the container (see docker-compose.yml) and check the path. ', + fileOk: (r) => `Followed: ${r} entries since startup.`, + fileDetected: (f) => ` Format detected: ${f}.`, + 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.', + file_invalid: '', + file_not_found: 'No file at this path: mount it in the container (see docker-compose.yml) and check the path.', file_permission: 'Permission denied: the file must be readable by UID 10001 or the adm group (see the README). ', file_read: 'Cannot read the file: ', - filesHelp: 'Follows the log files of other programs, mounted read-only in the container (e.g. /var/log/traefik:/logs/traefik:ro). The path can use wildcards (/logs/traefik/*.log follows every .log file of the directory, also those created later); rotated copies (.1, .gz) are left out. Each line gives an entry (badge FILE, app = the name): plain text (level and leading timestamp detected, indented lines added to the previous entry), JSON lines, logfmt, access logs of nginx, Apache or Traefik (5xx are errors, 4xx warnings) or a regular expression whose named groups time, level, msg, host and app fill these fields, the other groups giving extra fields. When turned on, the last hour of dated lines is read first (FILES_BACKFILL); rotations are followed and reading resumes where it stopped.', + filesHelp: 'Follows the log file of another program, mounted read-only in the container (e.g. /var/log/traefik:/logs/traefik:ro, then /logs/traefik/access.log here). Once the path is filled in, the file is checked and its format detected from its last lines (JSON lines, logfmt, access logs of nginx, Apache or Traefik, else plain text); choose a format to force it, or a regular expression whose named groups time, level, msg, host and app fill these fields, the other groups giving extra fields. Each line gives an entry (badge FILE, app = the name); indented lines are added to the previous entry; for access logs 5xx are errors and 4xx warnings. When turned on, the last hour of dated lines is read first (FILES_BACKFILL); rotations are followed and reading resumes where it stopped.', auditTitle: 'Linux audit (auditd)', auditEnabled: 'Collect the audit events of this machine', auditPath: 'File', @@ -305,10 +311,11 @@ const I18N = { addFile: '+ Ajouter un fichier', filesNone: 'Aucun fichier pour l\'instant : ajoutez-en un pour suivre les logs de Traefik, nginx ou de tout autre programme.', fileName: 'Nom (app)', - filePath: '/logs/traefik/*.log', + filePath: '/logs/traefik/access.log', fileHost: 'Hôte', fileRegex: '(?P