Settings: Save and Cancel buttons instead of saving each change

Changes made in the Settings window show at once but are only kept with
the Save button, placed under the tab labels; it takes the accent color
while changes are waiting. Cancel goes back to the saved settings, and
closing the window (or the page) with unsaved changes asks first.

Tags are now sent as a whole list (PUT /api/tags, nothing saved when one
is invalid, the error gives its position) and "Restore default tags"
shows GET /api/tags/defaults until Save.

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

.env.example

@@ -1,6 +1,6 @@
 # Copy this file to .env and adjust it.
-SYSLOG_PORT=514
+SYSLOG_PORT=5514
-HTTP_PORT=8080
+HTTP_PORT=8989
 TZ=Europe/Paris
 # How long logs are kept (e.g. 7d, 30d, 12w, 1y)
 RETENTION=30d
@@ -53,3 +53,21 @@ DOCKER_BACKFILL=1h
 HOST_LOGS_GID=4
 # History read when the source is turned on (e.g. 30m, 1h, 24h; 0 = only new entries)
 HOST_LOGS_BACKFILL=1h
+# Kernel messages of the host (dmesg, enable them in Settings > Sources). The container needs
+# /dev/kmsg: uncomment "devices" in docker-compose.yml. When the host has
+# kernel.dmesg_restrict=1 (default on Debian/Ubuntu), either run on the host
+#   sysctl -w kernel.dmesg_restrict=0   (persist it in /etc/sysctl.d/)
+# or run the container as root with the SYSLOG capability ("user" and "cap_add" in docker-compose.yml).
+# History read when the source is turned on (e.g. 30m, 1h, 24h; 0 = only new messages)
+DMESG_BACKFILL=1h
+# Linux audit log of the host (auditd, enable it in Settings > Sources). Uncomment the
+# /var/log/audit volume in docker-compose.yml and let the container read the file:
+# log_group = adm in /etc/audit/auditd.conf, and chgrp adm + chmod 750 on /var/log/audit.
+# History read when the source is turned on (e.g. 30m, 1h, 24h; 0 = only new events)
+AUDIT_BACKFILL=1h
+# Host name of the audit events (empty: node= of the records, else /host/etc/hostname, else localhost)
+AUDIT_HOST=
+# Log files of other programs (Traefik, nginx, applications…), declared in Settings > Log sources >
+# Log files: mount their directory read-only in docker-compose.yml (e.g. /var/log/traefik:/logs/traefik:ro).
+# History read when a file is turned on (e.g. 30m, 1h, 24h; 0 = only new lines)
+FILES_BACKFILL=1h

README.fr.md

@@ -55,7 +55,7 @@ Ouvrez ensuite <http://localhost:8080>.
 - **Test manuel** : `logger -n 127.0.0.1 -P 514 -d "hello error"` (util-linux) ou
   `echo "<14>test ok" | nc -u -w1 127.0.0.1 514`.
 
-**Paramètres > Sources > Syslog** active ou désactive la réception syslog et choisit les
+**Paramètres > Sources de logs > Syslog** active ou désactive la réception syslog et choisit les
 protocoles (UDP, TCP) sans redémarrage ; le choix est enregistré dans `/data/syslog.json`. Cet
 onglet affiche aussi l'état d'écoute (et l'erreur si le port est déjà utilisé). Le port
 lui-même est publié par docker-compose : modifiez `SYSLOG_PORT` dans `.env`, puis lancez
@@ -98,7 +98,7 @@ affiche tous ses champs.
 Les en-têtes de colonnes restent visibles pendant le défilement. Faites glisser le bord d'un
 en-tête (réception, heure message, sévérité, hôte, app) pour redimensionner la colonne, et
 double-cliquez dessus pour revenir à la largeur automatique ; le message occupe l'espace
-restant. Les largeurs sont mémorisées par le navigateur (**Paramètres > Interface >
+restant. Les largeurs sont mémorisées par le navigateur (**Paramètres > Affichage >
 Réinitialiser les colonnes** les rétablit toutes). Sur téléphone, la liste garde sa
 présentation sur deux lignes, sans colonnes.
 
@@ -126,7 +126,7 @@ lignes).
   chaque nouvel intervalle. Rien n'est rafraîchi tant que l'onglet du navigateur est masqué ;
   la frise se met à jour dès qu'il redevient visible.
 
-**Paramètres > Interface > Frise** (mémorisé par navigateur) : échelle (linéaire, √ par défaut,
+**Paramètres > Affichage > Frise** (mémorisé par navigateur) : échelle (linéaire, √ par défaut,
 log), hauteur (S/M/L : 40/80/120 px), couleur (empilement par sévérité, intensité comparée à
 la médiane de la fenêtre : calme, rafale au-delà de 3×, anomalie au-delà de 10×, ou aucune),
 barres ou aire, division (automatique, environ 100 intervalles, ou fixe : 1 s, 10 s, 1 min,
@@ -155,7 +155,7 @@ colorés et exportés comme les messages syslog :
   (`"level":"error"`), logfmt (`level=warn`), `[ERROR]`, ou un niveau en majuscules au début de
   la ligne (`ERROR`, `WARN`…). Sinon, elle vaut `info`. Les codes de couleur du terminal sont
   supprimés.
-- **Paramètres > Sources** affiche une étiquette par conteneur (`project/service`) dans un seul
+- **Paramètres > Sources de logs** affiche une étiquette par conteneur (`project/service`) dans un seul
   champ : les conteneurs suivis en couleur, puis les non suivis en gris ; un clic bascule une
   étiquette. La couleur identifie le projet compose, et les noms d'applications Docker de la
   liste utilisent la même couleur. Les conteneurs arrêtés sont masqués par défaut (« Afficher
@@ -181,7 +181,7 @@ moteur (`GET` uniquement).
 
 Logstream peut aussi collecter les logs système de la machine qui héberge la stack, sans rien
 configurer sur l'hôte. La source est **désactivée par défaut** : activez-la dans
-**Paramètres > Sources > Logs système de l'hôte** (enregistré dans `/data/hostlogs.json`).
+**Paramètres > Sources de logs > Logs système de l'hôte** (enregistré dans `/data/hostlogs.json`).
 
 - `docker-compose.yml` monte `/var/log` et `/run/log/journal` en lecture seule sous `/host`.
 - Si l'hôte utilise systemd, Logstream lit directement les fichiers du **journal systemd**
@@ -196,13 +196,174 @@ configurer sur l'hôte. La source est **désactivée par défaut** : activez-la
 - **Droits** : le conteneur tourne avec un utilisateur sans privilège et reçoit le groupe `adm`
   (gid 4), qui peut lire le journal et `/var/log` sur Debian et Ubuntu. Sur d'autres systèmes,
   réglez `HOST_LOGS_GID` dans `.env` sur le gid de `systemd-journal`
-  (`getent group systemd-journal | cut -d: -f3`). Paramètres > Sources affiche un message clair
+  (`getent group systemd-journal | cut -d: -f3`). Paramètres > Sources de logs affiche un message clair
   si l'accès est refusé.
 - Limites : les champs du journal compressés par journald (messages de plus de 512 octets,
   compressés en zstd/lz4/xz) ne peuvent pas être décodés sans bibliothèque supplémentaire ; ils
   sont comptés dans les Paramètres et affichés comme « (compressed journal entry) ». Les fichiers
   texte tournés ou compressés (`*.1`, `*.gz`) ne sont pas lus.
 
+## Messages du noyau (dmesg)
+
+Logstream peut collecter le buffer circulaire du noyau de l'hôte (ce qu'affiche `dmesg` :
+matériel, pilotes, OOM killer, pare-feu…) en lisant `/dev/kmsg`. La source est **désactivée par
+défaut** : activez-la dans **Paramètres > Sources de logs > Messages du noyau (dmesg)** (enregistré dans
+`/data/dmesg.json`).
+
+- **Périphérique** : le conteneur ne voit pas `/dev/kmsg` par défaut. Décommentez la section
+  `devices` de `docker-compose.yml` (`/dev/kmsg:/dev/kmsg:r`, en lecture seule), puis
+  `docker compose up -d`.
+- **Droits** : si l'hôte a `kernel.dmesg_restrict=1` (le défaut sur Debian et Ubuntu), seuls
+  les processus qui ont la capacité `CAP_SYSLOG` peuvent lire le log du noyau, et le conteneur
+  tourne avec un utilisateur sans privilège. Au choix :
+  - l'autoriser sur l'hôte : `sysctl -w kernel.dmesg_restrict=0`, conservé au redémarrage avec
+    `echo kernel.dmesg_restrict=0 | sudo tee /etc/sysctl.d/50-dmesg.conf` (tous les
+    utilisateurs de l'hôte peuvent alors lancer `dmesg`) ; ce réglage ne peut pas être changé
+    depuis un conteneur ;
+  - ou lancer Logstream en root avec la capacité `SYSLOG` : décommentez `user: "0:0"` et
+    `cap_add: [SYSLOG]` dans `docker-compose.yml`.
+
+  Paramètres > Sources de logs indique si le périphérique est absent ou l'accès refusé ; le reste de
+  Logstream continue de fonctionner et le périphérique est réessayé toutes les 10 secondes.
+- **Champs** : **app** vaut `kernel` (ou le nom du programme pour les messages écrits par des
+  programmes, comme `systemd[1]: …`), la sévérité et la facility viennent de l'enregistrement,
+  **host** vaut `localhost`, `subsystem` et `device` sont conservés quand le noyau les fournit.
+  Ces logs ont la source `kernel` (`proto` `kmsg`) : le filtre **Source** permet de les afficher
+  seuls.
+- **Heure** : les enregistrements portent le temps écoulé depuis le démarrage ; Logstream le
+  convertit en date grâce à l'heure de démarrage (`/proc/uptime`). Comme avec `dmesg -T`,
+  l'heure peut être trop tôt après une mise en veille de l'hôte.
+- À l'activation, la dernière heure est lue d'abord (`DMESG_BACKFILL`, `0` = seulement les
+  nouveaux messages). Le numéro de séquence du dernier message et l'identifiant de démarrage
+  sont enregistrés dans `/data/dmesg-state.json` : après un redémarrage de Logstream, la lecture
+  reprend après le dernier message (sans doublon) ; après un redémarrage de l'hôte, le nouveau
+  buffer est lu depuis `DMESG_BACKFILL`. Les messages écrasés dans le buffer du noyau avant
+  d'être lus sont comptés dans Paramètres.
+
+## Audit Linux (auditd)
+
+Logstream peut suivre le journal du démon d'audit Linux (`/var/log/audit/audit.log`) :
+connexions, `sudo`, fichiers surveillés, appels système visés par vos règles `auditctl`… La
+source est **désactivée par défaut** : activez-la dans **Réglages > Sources > Audit Linux
+(auditd)**, où le chemin du fichier peut aussi être changé (enregistré dans `/data/audit.json` ;
+`AUDIT_LOG` fixe le chemin par défaut).
+
+- **Montage** : décommentez `/var/log/audit:/var/log/audit:ro` dans `docker-compose.yml`, puis
+  `docker compose up -d`. (`/var/log` est déjà monté sous `/host` pour les logs système de
+  l'hôte : `/host/var/log/audit/audit.log` fonctionne donc aussi comme chemin, sans nouveau
+  volume.)
+- **Droits** (côté hôte) : auditd écrit son journal en `root`, mode `0600`, dans un répertoire
+  `0700`, et Logstream tourne avec l'UID `10001`. Le plus fiable est le groupe que le conteneur
+  reçoit déjà pour les logs de l'hôte (`HOST_LOGS_GID`, `adm` par défaut) :
+  ```bash
+  sudo sed -i 's/^log_group.*/log_group = adm/' /etc/audit/auditd.conf
+  sudo chgrp adm /var/log/audit && sudo chmod 750 /var/log/audit
+  sudo service auditd restart        # systemctl refuse de redémarrer auditd sur certaines distributions
+  ```
+  auditd crée alors chaque fichier, rotations comprises, en mode `0640` avec le groupe `adm`.
+  Une ACL fonctionne aussi
+  (`sudo setfacl -m u:10001:rx /var/log/audit && sudo setfacl -m u:10001:r /var/log/audit/audit.log`)
+  mais doit être réappliquée à chaque nouveau fichier après une rotation. Réglages > Sources
+  indique si le fichier est absent ou l'accès refusé ; il est réessayé chaque seconde.
+- **Événements** : les enregistrements d'un même événement (`SYSCALL`, `EXECVE`, `CWD`, `PATH`,
+  `PROCTITLE`…) partagent le même `msg=audit(heure:numéro)` et deviennent **un seul** log, dès
+  qu'aucun enregistrement n'est arrivé depuis 2 secondes. Le message est un résumé — qui, quoi,
+  résultat — comme `cedric (as root): exec apt install nginx [exec] → success` ou
+  `bob: PAM:authentication via /usr/sbin/sshd from 203.0.113.9 on ssh → failed` ; les
+  enregistrements bruts sont gardés dans `audit_raw` et affichés sous le résumé (repliés en
+  densité compacte tant que la ligne n'est pas ouverte). Les valeurs hexadécimales
+  (`proctitle`, `cmd`, arguments avec espaces) sont décodées.
+- **Champs** : `audit_type` (type du premier enregistrement, aussi utilisé comme **app**),
+  `audit_serial`, `auid` (identifiant de l'utilisateur connecté), `user` (son nom, ou l'uid s'il
+  ne peut pas être résolu), `exe`, `res` (`success` / `failed`, d'après `res=` ou `success=`) et
+  `key` (clé de la règle). La sévérité est `warning` en cas d'échec, `info` sinon ; facility
+  `authpriv` ; source `audit`. Dans la liste, les lignes d'audit ont un liseré gauche violet,
+  un fond légèrement teinté et un badge **AUDIT** devant l'app ; le filtre **Source** permet de
+  les afficher seules.
+- **Noms** : `user` vient du format enrichi (`log_format = ENRICHED`, le défaut des versions
+  récentes d'auditd), sinon du `/etc/passwd` de l'hôte s'il est monté en `/host/etc/passwd`.
+  **host** est le `node=` des enregistrements (`name_format = hostname` dans `auditd.conf`),
+  sinon `AUDIT_HOST`, sinon `/host/etc/hostname`, sinon `localhost`.
+- À l'activation, la dernière heure est lue d'abord (`AUDIT_BACKFILL`, `0` = seulement les
+  nouveaux événements). L'inode et une position avant laquelle tout événement a été envoyé sont
+  enregistrés dans `/data/audit-state.json` : après un redémarrage, la lecture reprend là ;
+  après une rotation par auditd, la fin de `audit.log.1` est lue avant le nouveau fichier.
+
+## Fichiers de logs (Traefik, nginx, applications…)
+
+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, 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, 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]`,
+    `Jan  2 15:04:05`) devient l'heure du message et est retiré du message. Les lignes indentées
+    (stack traces) sont rattachées à l'entrée précédente.
+  - *JSON lines* : `msg` / `message`, `time` / `ts` / `@timestamp` (texte ou heure Unix) et
+    `level` (mot ou nombre, pino/bunyan compris) sont reconnus ; les autres clés deviennent des
+    champs, les objets imbriqués avec des noms pointés (`req.id`).
+  - *logfmt* : paires `clé=valeur`, comme les logs par défaut de Traefik ou des programmes Go ;
+    mêmes clés que le JSON.
+  - *Journal d'accès (CLF)* : Common / Combined Log Format de nginx et Apache, et le journal
+    d'accès de Traefik dans ce format (routeur, URL du serveur et durée compris). Champs
+    `client_ip`, `http_method`, `http_path`, `http_status`, `http_size`, `user_agent`,
+    `http_referer`, `router`, `backend`, `duration_ms`.
+  - *Accès Traefik (JSON)* : le journal d'accès JSON de Traefik (`accessLog.format: json`),
+    résumé en `GET app.example.com/ 200 · 12.5 ms · app@docker · 203.0.113.9`, toutes les clés
+    dans les champs. Les logs de Traefik lui-même dans le même fichier sont lus comme du JSON.
+  - *Expression régulière* (syntaxe RE2 / Go) : les groupes nommés `time`, `level`, `msg`,
+    `host` et `app` remplissent ces champs, `status` donne le niveau d'un journal d'accès, les
+    autres groupes deviennent des champs, par ex.
+    `^(?P<time>\S+ \S+) \[(?P<level>\w+)\] (?P<module>\S+): (?P<msg>.*)$`. Une ligne qui ne
+    correspond pas est rattachée à l'entrée précédente.
+
+  Pour les journaux d'accès, les statuts 5xx sont en `err`, les 4xx en `warning`, les autres en
+  `info`. Une ligne qui ne suit pas le format choisi est gardée comme du texte brut. Les champs
+  qui portent le nom d'un champ de LogStream (`host`, `app`…) sont renommés `field_host`,
+  `field_app`…
+- **Affichage** : source `file`, le chemin dans `log_file`. Dans la liste, ces lignes ont un
+  liseré gauche vert, une légère teinte et un badge **FILE** avant l'app ; le filtre **Source**
+  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
+  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
+  relu depuis son début.
+
+## Logs internes de LogStream
+
+Logstream enregistre ses propres événements comme des logs, avec l'**hôte** `logstream`,
+l'**app** du composant (`store`, `syslog`, `docker`, `host`, `dmesg`, `audit`, `files`, `main`) et la
+source `logstream` : VictoriaLogs injoignable puis de retour, lots perdus, lots refusés depuis
+le tampon disque, erreurs du listener syslog, API Docker ou flux d'un conteneur décrochés,
+sources qui ne peuvent pas lire, démarrage et arrêt. Ils sont actifs par défaut à partir du
+niveau **warning** ; **Réglages > Sources > Logs internes de LogStream** permet de les couper
+ou de changer le niveau minimal (enregistré dans `/data/selflog.json`). Un même événement
+(composant et message) est enregistré au plus une fois toutes les 30 secondes, avec le nombre
+de répétitions écartées ; les événements passent par une file bornée et n'attendent jamais le
+stockage : un VictoriaLogs injoignable ne peut ni les accumuler ni les faire boucler. Tout
+continue d'aller dans `docker compose logs logstream` comme avant. Dans la liste, ces lignes
+ont un liseré gauche turquoise et un badge **LS** devant l'app ; le filtre **Source** permet de
+les afficher seules.
+
 ## Export CSV
 
 Le bouton **Exporter** (à côté du nombre de logs) télécharge tous les logs stockés qui
@@ -225,16 +386,21 @@ En ligne de commande : `curl -o logs.csv 'localhost:8080/api/export.csv?q=error&
 
 ## Paramètres
 
-L'icône en forme d'engrenage ouvre les paramètres, organisés en onglets. Tout, sauf les tags de
+L'icône en forme d'engrenage ouvre les paramètres, organisés en onglets. Les modifications
-couleur, est mémorisé par navigateur.
+s'affichent tout de suite mais ne sont conservées qu'avec le bouton **Valider**, sous les libellés
+des onglets : il prend la couleur d'accent tant que des modifications sont en attente. **Annuler**
+revient aux paramètres enregistrés, et fermer la fenêtre avec des modifications non validées
+demande confirmation avant de les abandonner. Localisation et Affichage sont mémorisés par
+navigateur ; les tags de couleur et les sources de logs sont enregistrés sur le serveur. La purge
+et la réinitialisation des colonnes agissent immédiatement.
 
-- **Localisation**
+- **Localisation et langue**
   - *Langue* : anglais ou français.
   - *Date et heure* : fuseau horaire (celui du navigateur, UTC ou environ 80 fuseaux courants)
     et format d'affichage de l'heure de réception : `DD/MM/YYYY HH:MM:SS` (par défaut,
     l'affichage français habituel), avec millisecondes, `YYYY-MM-DD`, horloge sur 12 heures,
     ISO 8601 ou epoch Unix. Le fuseau horaire s'applique à toutes les dates affichées.
-- **Filtres** : tags de couleur. Chaque tag a un mot-clé, une couleur de fond (le texte passe
+- **Motifs de recherche** : tags de couleur. Chaque tag a un mot-clé, une couleur de fond (le texte passe
   automatiquement en noir ou en blanc pour rester lisible) et des options : mot entier, respect
   de la casse, expression régulière, actif. Les tags sont stockés sur le serveur dans
   `/data/tags.json` (volume `logstream-data`) : ils sont donc partagés par tous les navigateurs.
@@ -253,10 +419,10 @@ couleur, est mémorisé par navigateur.
   les préréglages s'en servent pour colorer le code de statut ou la méthode, pas le texte autour.
   Chaque tag reçoit un code à deux chiffres (`01`, `02`…) attribué par le serveur : il reste
   attaché au tag jusqu'à sa suppression (les tags créés par les versions précédentes en reçoivent
-  un aussi). La colonne *Filtres* de la liste affiche, en badges gris, les codes des tags actifs
+  un aussi). La colonne *Motif de recherche* de la liste affiche, en badges gris, les codes des tags actifs
   trouvés dans chaque message ; elle a la place pour 3, au-delà elle en affiche 2 et `+N`, et
   l'infobulle les liste tous.
-- **Interface**
+- **Affichage**
   - *Thème* : Système (suit la préférence de l'ordinateur ou du téléphone), Clair ou Sombre. Le
     bouton soleil/lune de l'en-tête bascule entre clair et sombre.
   - *Affichage des logs* : taille du texte (très petite, petite, moyenne, grande), densité
@@ -270,7 +436,7 @@ couleur, est mémorisé par navigateur.
     européen de polices respectueux de la vie privée ; sans accès à internet, la police du
     système est utilisée. Les ligatures sont désactivées pour que `->` ou `!=` s'affichent tels
     quels. Le réglage le plus dense est Très petite + Compacte + Inconsolata Condensed.
-- **Données** : chiffres de la base lus dans VictoriaLogs (actualisés au plus toutes les 30 s) :
+- **Base de données** : chiffres de la base lus dans VictoriaLogs (actualisés au plus toutes les 30 s) :
   lignes stockées, taille sur disque (index compris), taille brute et taux de compression,
   période couverte avec la rétention, lignes des dernières 24 h et de la dernière heure, hôtes
   et applications distincts, espace disque libre. Les tailles sont en Ko/Mo/Go (KB/MB/GB en
@@ -396,6 +562,12 @@ résolutions.
 | `HOST_LOGS_GID` | `4` (adm) dans compose | groupe donné au conteneur pour lire les logs de l'hôte |
 | `HOST_LOGS_BACKFILL` | `1h` | historique lu à l'activation de la source « logs système de l'hôte » |
 | `HOST_LOGS_ROOT` | `/host` | emplacement de montage des répertoires de l'hôte |
+| `DMESG_BACKFILL` | `1h` | historique lu à l'activation de la source « messages du noyau » |
+| `DMESG_DEVICE` | `/dev/kmsg` | périphérique du log noyau lu par la source « messages du noyau » |
+| `AUDIT_LOG` | `/var/log/audit/audit.log` | chemin par défaut du journal d'audit Linux (modifiable dans Réglages > Sources) |
+| `AUDIT_BACKFILL` | `1h` | historique lu à l'activation de la source « audit Linux » |
+| `AUDIT_HOST` | vide | nom d'hôte des événements d'audit quand les enregistrements n'ont pas de `node=` |
+| `FILES_BACKFILL` | `1h` | historique lu à l'activation d'un fichier de logs (Paramètres > Sources de logs > Fichiers de logs) |
 | `TZ` | `Europe/Paris` | fuseau horaire des horodatages RFC 3164 (qui n'en portent pas) |
 | `BATCH_SIZE`, `FLUSH_MS`, `QUEUE_SIZE` | `1000`, `1000`, `100000` | réglage de l'ingestion |
 | `SPOOL_MAX_MB` | `1024` | taille du tampon disque des lots refusés par VictoriaLogs (`0` = pas de tampon : 15 s de tentatives, puis perte) |
@@ -436,7 +608,7 @@ Pour mettre à jour l'une d'elles :
    VictoriaLogs conserve son format de stockage entre versions mineures ; lisez le changelog
    avant un changement de version majeure.
 2. Modifiez la version dans le fichier indiqué ci-dessus, puis lancez `docker compose up -d --build`.
-3. Vérifiez la barre du bas (reçus / stockés / perdus) et **Paramètres > Sources**. Pour revenir
+3. Vérifiez la barre du bas (reçus / stockés / perdus) et **Paramètres > Sources de logs**. Pour revenir
    en arrière, remettez la version précédente et lancez la même commande.
 
 ## Organisation du code
@@ -454,8 +626,12 @@ Pour mettre à jour l'une d'elles :
 | `rdns.go` | résolutions DNS inverses avec cache |
 | `export.go` | export CSV en flux |
 | `docker.go` | logs des conteneurs Docker (API, lecteurs, positions, détection du niveau) |
-| `syslogserver.go` | écoutes syslog ouvertes et fermées depuis Paramètres > Sources |
+| `syslogserver.go` | écoutes syslog ouvertes et fermées depuis Paramètres > Sources de logs |
 | `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 (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 |
 | `api.go` | routes HTTP `/api/*` |

README.md

@@ -50,7 +50,7 @@ Then open <http://localhost:8080>.
 - **Manual test**: `logger -n 127.0.0.1 -P 514 -d "hello error"` (util-linux) or
   `echo "<14>test ok" | nc -u -w1 127.0.0.1 514`.
 
-**Settings > Sources > Syslog** turns syslog reception on or off and chooses the protocols
+**Settings > Log sources > Syslog** turns syslog reception on or off and chooses the protocols
 (UDP, TCP) without restarting; the choice is saved in `/data/syslog.json`. It also shows the
 listening state (and the error if the port is already in use). The port itself is published
 by docker-compose: change `SYSLOG_PORT` in `.env`, then run `docker compose up -d`.
@@ -89,7 +89,7 @@ Shortcuts: `/` focuses the search box, `Esc` clears it. Clicking a row shows all
 The column headers stay visible while scrolling. Drag the edge of a header (received, message
 time, severity, host, app) to resize the column, double-click it to go back to the automatic
 width; the message takes the remaining space. Widths are remembered by the browser
-(**Settings > Interface > Reset column widths** restores them all). On phones the list keeps its
+(**Settings > Appearance > Reset column widths** restores them all). On phones the list keeps its
 two-line layout without columns.
 
 ## Stream and Time range modes
@@ -114,7 +114,7 @@ time** on the server clock (so it matches the times shown in the rows).
 - In live mode the last interval grows as messages arrive, and the timeline reloads at each new
   interval. Nothing is refreshed while the browser tab is hidden; it catches up when shown again.
 
-**Settings > Interface > Timeline** (remembered per browser): scale (linear, √ by default, log),
+**Settings > Appearance > Timeline** (remembered per browser): scale (linear, √ by default, log),
 height (S/M/L: 40/80/120 px), color (stacked by severity, intensity compared with the median of
 the window: calm, burst above 3×, anomaly above 10×, or none), bars or area, division
 (automatic, about 100 intervals, or fixed: 1 s, 10 s, 1 min, 5 min, 1 h, 1 day) and refresh
@@ -141,7 +141,7 @@ colored and exported like syslog messages:
 - The severity comes from the line itself when the application writes it: JSON
   (`"level":"error"`), logfmt (`level=warn`), `[ERROR]`, or an upper-case level word at the
   start of the line (`ERROR`, `WARN`…). Otherwise it is `info`. Terminal color codes are removed.
-- **Settings > Sources** shows one label per container (`project/service`) in a single field:
+- **Settings > Log sources** shows one label per container (`project/service`) in a single field:
   followed containers in color, then not followed ones in grey; a click switches a label.
   The color identifies the compose project, and Docker app names in the log list use the
   same color. Stopped containers are hidden by default ("Show stopped containers" displays
@@ -165,7 +165,7 @@ which only lets through listing containers, reading logs, events and engine info
 
 Logstream can also collect the system logs of the machine hosting the stack, without
 configuring anything on the host. The source is **off by default**: turn it on in
-**Settings > Sources > Host system logs** (saved in `/data/hostlogs.json`).
+**Settings > Log sources > Host system logs** (saved in `/data/hostlogs.json`).
 
 - `docker-compose.yml` mounts `/var/log` and `/run/log/journal` read-only under `/host`.
 - When the host runs systemd, Logstream reads the **systemd journal** files directly (no
@@ -180,13 +180,158 @@ configuring anything on the host. The source is **off by default**: turn it on i
 - **Permissions**: the container runs as an unprivileged user and gets the `adm` group
   (gid 4), which can read the journal and `/var/log` on Debian and Ubuntu. On other systems,
   set `HOST_LOGS_GID` in `.env` to the gid of `systemd-journal`
-  (`getent group systemd-journal | cut -d: -f3`). Settings > Sources shows a clear message when
+  (`getent group systemd-journal | cut -d: -f3`). Settings > Log sources shows a clear message when
   access is denied.
 - Limits: journal fields compressed by journald (messages longer than 512 bytes, compressed
   with zstd/lz4/xz) cannot be decoded without extra libraries; they are counted in Settings and
   shown as "(compressed journal entry)". Rotated or compressed text files (`*.1`, `*.gz`) are
   not read.
 
+## Kernel messages (dmesg)
+
+Logstream can collect the kernel ring buffer of the host (what `dmesg` shows: hardware,
+drivers, OOM killer, firewall…) by reading `/dev/kmsg`. The source is **off by default**: turn it
+on in **Settings > Log sources > Kernel messages (dmesg)** (saved in `/data/dmesg.json`).
+
+- **Device**: the container does not see `/dev/kmsg` by default. Uncomment the `devices`
+  section of `docker-compose.yml` (`/dev/kmsg:/dev/kmsg:r`, read-only), then
+  `docker compose up -d`.
+- **Permissions**: when the host has `kernel.dmesg_restrict=1` (the default on Debian and
+  Ubuntu), only processes with the `CAP_SYSLOG` capability can read the kernel log, and the
+  container runs as an unprivileged user. Either:
+  - allow it on the host: `sysctl -w kernel.dmesg_restrict=0`, kept across reboots with
+    `echo kernel.dmesg_restrict=0 | sudo tee /etc/sysctl.d/50-dmesg.conf` (every user of the
+    host can then run `dmesg`); this setting cannot be changed from a container;
+  - or run Logstream as root with the `SYSLOG` capability: uncomment `user: "0:0"` and
+    `cap_add: [SYSLOG]` in `docker-compose.yml`.
+
+  Settings > Log sources says when the device is missing or access is denied; the rest of
+  Logstream keeps working and the device is tried again every 10 seconds.
+- **Fields**: **app** is `kernel` (or the program name for messages written by programs such
+  as `systemd[1]: …`), severity and facility come from the record, **host** is `localhost`,
+  `subsystem` and `device` are kept when the kernel gives them. These logs have the `kernel`
+  source (`proto` `kmsg`): the **Source** filter shows them alone.
+- **Time**: records carry the time since boot; Logstream turns it into a date using the boot
+  time (`/proc/uptime`). As with `dmesg -T`, the time can be early after the host was suspended.
+- When the source is turned on, the last hour is read first (`DMESG_BACKFILL`, `0` = only new
+  messages). The sequence number of the last message and the boot ID are saved in
+  `/data/dmesg-state.json`: after a restart of Logstream, reading resumes after the last
+  message (no duplicates); after a reboot of the host, the new buffer is read from
+  `DMESG_BACKFILL` ago. Messages overwritten in the kernel buffer before being read are counted
+  in Settings.
+
+## Linux audit (auditd)
+
+Logstream can follow the log of the Linux audit daemon (`/var/log/audit/audit.log`): logins,
+`sudo`, watched files, syscalls matched by your `auditctl` rules… The source is **off by
+default**: turn it on in **Settings > Log sources > Linux audit (auditd)**, where the file path can
+also be changed (saved in `/data/audit.json`; `AUDIT_LOG` sets the default path).
+
+- **Mount**: uncomment `/var/log/audit:/var/log/audit:ro` in `docker-compose.yml`, then
+  `docker compose up -d`. (`/var/log` is already mounted under `/host` for the host system
+  logs, so `/host/var/log/audit/audit.log` also works as the path, without a new volume.)
+- **Permissions** (host side): auditd writes its log as `root`, mode `0600`, in a `0700`
+  directory, and Logstream runs as UID `10001`. The reliable way is the group the container
+  already gets for the host logs (`HOST_LOGS_GID`, `adm` by default):
+  ```bash
+  sudo sed -i 's/^log_group.*/log_group = adm/' /etc/audit/auditd.conf
+  sudo chgrp adm /var/log/audit && sudo chmod 750 /var/log/audit
+  sudo service auditd restart        # systemctl refuses to restart auditd on some distributions
+  ```
+  auditd then creates every rotated file with mode `0640` and group `adm`. An ACL also works
+  (`sudo setfacl -m u:10001:rx /var/log/audit && sudo setfacl -m u:10001:r /var/log/audit/audit.log`)
+  but must be applied again to each new file after a rotation. Settings > Log sources says when the
+  file is missing or access is denied; it is tried again every second.
+- **Events**: the records of one event (`SYSCALL`, `EXECVE`, `CWD`, `PATH`, `PROCTITLE`…) share
+  the same `msg=audit(time:serial)` and become **one** log, once no record came for 2 seconds.
+  The message is a summary — who, what, result — such as
+  `cedric (as root): exec apt install nginx [exec] → success` or
+  `bob: PAM:authentication via /usr/sbin/sshd from 203.0.113.9 on ssh → failed`; the raw
+  records are kept in `audit_raw` and shown under the summary (folded in compact density until
+  the line is opened). Hexadecimal values (`proctitle`, `cmd`, arguments with spaces) are
+  decoded.
+- **Fields**: `audit_type` (type of the first record, also used as **app**), `audit_serial`,
+  `auid` (login user ID), `user` (its name, or the uid when it cannot be resolved), `exe`, `res`
+  (`success` / `failed`, from `res=` or `success=`) and `key` (rule key). Severity is `warning`
+  when the result is a failure, `info` otherwise; facility `authpriv`; source `audit`. In the
+  list, audit lines have a violet left edge, a light tint and an **AUDIT** badge before the app;
+  the **Source** filter shows them alone.
+- **Names**: `user` comes from the enriched format (`log_format = ENRICHED`, the default of
+  recent auditd versions), else from the host's `/etc/passwd` when it is mounted as
+  `/host/etc/passwd`. **host** is the `node=` of the records (`name_format = hostname` in
+  `auditd.conf`), else `AUDIT_HOST`, else `/host/etc/hostname`, else `localhost`.
+- When the source is turned on, the last hour is read first (`AUDIT_BACKFILL`, `0` = only new
+  events). The inode and an offset before which every event was sent are saved in
+  `/data/audit-state.json`: after a restart, reading resumes there; after a rotation by auditd,
+  the end of `audit.log.1` is read before the new file.
+
+## Log files (Traefik, nginx, applications…)
+
+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, 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**: 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]`,
+    `Jan  2 15:04:05`) becomes the message time and is removed from the message. Indented
+    lines (stack traces) are added to the previous entry.
+  - *JSON lines*: `msg` / `message`, `time` / `ts` / `@timestamp` (text or Unix time) and
+    `level` (word or number, pino/bunyan included) are recognized; the other keys become
+    fields, nested objects with dotted names (`req.id`).
+  - *logfmt*: `key=value` pairs, as in Traefik's default logs or Go programs; same keys as JSON.
+  - *Access log (CLF)*: Common / Combined Log Format of nginx and Apache, and Traefik's access
+    log in this format (router, server URL and duration included). Fields `client_ip`,
+    `http_method`, `http_path`, `http_status`, `http_size`, `user_agent`, `http_referer`,
+    `router`, `backend`, `duration_ms`.
+  - *Traefik access (JSON)*: the JSON access log of Traefik (`accessLog.format: json`),
+    summarized as `GET app.example.com/ 200 · 12.5 ms · app@docker · 203.0.113.9`, every key in
+    the fields. Traefik's own logs in the same file are read as JSON lines.
+  - *Regular expression* (RE2 / Go syntax): the named groups `time`, `level`, `msg`, `host`
+    and `app` fill these fields, `status` gives the level of an access log, the other groups
+    become fields, e.g. `^(?P<time>\S+ \S+) \[(?P<level>\w+)\] (?P<module>\S+): (?P<msg>.*)$`.
+    A line that does not match is added to the previous entry.
+
+  For access logs, 5xx statuses are `err`, 4xx `warning`, the others `info`. A line that does
+  not fit the chosen format is kept as plain text. Fields named like LogStream's own ones
+  (`host`, `app`…) are renamed `field_host`, `field_app`…
+- **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
+  `/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.
+
+## LogStream internal logs
+
+Logstream records its own events as logs, with **host** `logstream`, **app** the component
+(`store`, `syslog`, `docker`, `host`, `dmesg`, `audit`, `files`, `main`) and the `logstream` source:
+VictoriaLogs unreachable and back, lost batches, batches refused from the disk buffer, syslog
+listener errors, Docker API or container streams lost, sources that cannot read, start and
+stop. They are on by default from level **warning**; **Settings > Log sources > LogStream internal
+logs** turns them off or changes the minimum level (saved in `/data/selflog.json`). The same
+event (component and message) is recorded at most once every 30 seconds, with the number of
+repeats left out; events go through a bounded queue and never wait for the storage, so an
+unreachable VictoriaLogs cannot make them pile up or loop. Everything still goes to
+`docker compose logs logstream` as before. In the list, these lines have a turquoise left edge
+and an **LS** badge before the app; the **Source** filter shows them alone.
+
 ## CSV export
 
 The **Export** button (next to the log count) downloads every stored log matching the
@@ -206,8 +351,11 @@ From the command line: `curl -o logs.csv 'localhost:8080/api/export.csv?q=error&
 
 ## Settings
 
-The gear icon opens the settings, organized in tabs. Everything except color tags is
+The gear icon opens the settings, organized in tabs. Changes show at once but are only kept
-remembered per browser.
+with the **Save** button, under the tab labels: it takes the accent color while changes are
+waiting. **Cancel** goes back to the saved settings, and closing the window with unsaved changes
+asks before dropping them. Localization and Appearance are remembered per browser; color tags
+and log sources are saved on the server. Purging and resetting column widths act at once.
 
 - **Localization**
   - *Language*: English or French.
@@ -215,7 +363,7 @@ remembered per browser.
     format of the reception time: `DD/MM/YYYY HH:MM:SS` (default, the usual French display),
     with milliseconds, `YYYY-MM-DD`, 12-hour clock, ISO 8601 or Unix epoch. The time zone
     applies to every date shown.
-- **Filters**: color tags. Each tag has a keyword, a background color (the text automatically
+- **Patterns**: color tags. Each tag has a keyword, a background color (the text automatically
   switches to black or white to stay readable) and options: whole word, match case,
   regular expression, active. Tags are stored on the server in `/data/tags.json`
   (`logstream-data` volume), so they are shared by every browser. Default tags (pastel):
@@ -232,10 +380,10 @@ remembered per browser.
   or the method, not the text around it.
   Each tag gets a two-digit code (`01`, `02`…) assigned by the server: it stays with the tag
   until the tag is deleted (codes are also given to tags created by earlier versions). The
-  *Filters* column of the log list shows, as grey badges, the codes of the active tags found in
+  *Pattern* column of the log list shows, as grey badges, the codes of the active tags found in
   each message; it has room for 3, beyond that it shows 2 and `+N`, and the tooltip lists them
   all.
-- **Interface**
+- **Appearance**
   - *Theme*: System (follows the computer/phone preference), Light or Dark. The sun/moon
     button in the header switches between light and dark.
   - *Log display*: font size (tiny, small, medium, large), density (normal, or compact to
@@ -247,7 +395,7 @@ remembered per browser.
     [Bunny Fonts](https://fonts.bunny.net), a privacy-friendly European font service; without
     internet access, the system font is used. Ligatures are disabled so `->` or `!=` show as
     typed. The densest setting is Tiny + Compact + Inconsolata Condensed.
-- **Data**: database figures read from VictoriaLogs (refreshed at most every 30 s): stored
+- **Database**: database figures read from VictoriaLogs (refreshed at most every 30 s): stored
   lines, size on disk (index included), raw size and compression ratio, period covered with
   the retention, lines of the last 24 h and last hour, distinct hosts and apps, free disk
   space. Sizes use KB/MB/GB (Ko/Mo/Go in French). "Delete all logs" permanently erases every stored log (you must type
@@ -363,6 +511,12 @@ are only known by your router or a local DNS (Pi-hole, AdGuard, Unbound…), set
 | `HOST_LOGS_GID` | `4` (adm) in compose | group given to the container to read the host logs |
 | `HOST_LOGS_BACKFILL` | `1h` | history read when the host system logs source is turned on |
 | `HOST_LOGS_ROOT` | `/host` | where the host directories are mounted |
+| `DMESG_BACKFILL` | `1h` | history read when the kernel messages source is turned on |
+| `DMESG_DEVICE` | `/dev/kmsg` | kernel log device read by the kernel messages source |
+| `AUDIT_LOG` | `/var/log/audit/audit.log` | default path of the Linux audit log (can be changed in Settings > Log sources) |
+| `AUDIT_BACKFILL` | `1h` | history read when the Linux audit source is turned on |
+| `AUDIT_HOST` | empty | host name of the audit events when the records carry no `node=` |
+| `FILES_BACKFILL` | `1h` | history read when a log file is turned on (Settings > Log sources > Log files) |
 | `TZ` | `Europe/Paris` | time zone for RFC 3164 timestamps (which carry none) |
 | `BATCH_SIZE`, `FLUSH_MS`, `QUEUE_SIZE` | `1000`, `1000`, `100000` | ingestion tuning |
 | `SPOOL_MAX_MB` | `1024` | disk buffer size for batches VictoriaLogs could not take (`0` = no buffer: retried for 15 s, then dropped) |
@@ -401,7 +555,7 @@ To update one of them:
    VictoriaLogs keeps its storage format across minor versions; read the changelog before a
    major version change.
 2. Change the version in the file above, then run `docker compose up -d --build`.
-3. Check the bottom bar (received / stored / dropped) and **Settings > Sources**. To go back,
+3. Check the bottom bar (received / stored / dropped) and **Settings > Log sources**. To go back,
    restore the previous version and run the same command.
 
 ## Code layout
@@ -419,8 +573,12 @@ To update one of them:
 | `rdns.go` | cached reverse DNS lookups |
 | `export.go` | streamed CSV export |
 | `docker.go` | Docker container logs (API, followers, positions, level detection) |
-| `syslogserver.go` | syslog listeners opened and closed from Settings > Sources |
+| `syslogserver.go` | syslog listeners opened and closed from Settings > Log sources |
 | `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 (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 |
 | `api.go` | `/api/*` HTTP routes |

api.go

@@ -23,6 +23,10 @@ type API struct {
 	docker     *DockerManager // nil when DOCKER_LOGS is off
 	syslog     *SyslogServer
 	host       *HostLogs
+	dmesg      *Dmesg
+	audit      *Audit
+	files      *Files
+	selfLog    *SelfLog
 	dbstats    dbStatsCache
 }
 
@@ -36,6 +40,8 @@ func (a *API) Routes(mux *http.ServeMux) {
 	mux.HandleFunc("GET /api/dbstats", a.dbStats)
 	mux.HandleFunc("GET /api/tags", a.listTags)
 	mux.HandleFunc("POST /api/tags", a.createTag)
+	mux.HandleFunc("PUT /api/tags", a.replaceTags)
+	mux.HandleFunc("GET /api/tags/defaults", a.defaultTags)
 	mux.HandleFunc("POST /api/tags/reset", a.resetTags)
 	mux.HandleFunc("GET /api/presets", a.listPresets)
 	mux.HandleFunc("PUT /api/tags/{id}", a.updateTag)
@@ -49,6 +55,110 @@ func (a *API) Routes(mux *http.ServeMux) {
 	mux.HandleFunc("PUT /api/docker", a.dockerConfigure)
 	mux.HandleFunc("GET /api/hostlogs", a.hostLogsStatus)
 	mux.HandleFunc("PUT /api/hostlogs", a.hostLogsConfigure)
+	mux.HandleFunc("GET /api/dmesg", a.dmesgStatus)
+	mux.HandleFunc("PUT /api/dmesg", a.dmesgConfigure)
+	mux.HandleFunc("GET /api/audit", a.auditStatus)
+	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)
+}
+
+// GET /api/audit: state of the Linux audit source (Settings > Sources).
+func (a *API) auditStatus(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, a.audit.Status())
+}
+
+// PUT /api/audit {"enabled": bool, "path": "/var/log/audit/audit.log"}
+func (a *API) auditConfigure(w http.ResponseWriter, r *http.Request) {
+	var cfg auditConfig
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&cfg); err != nil {
+		writeErr(w, http.StatusBadRequest, err)
+		return
+	}
+	if err := a.audit.Configure(cfg); err != nil {
+		writeErr(w, http.StatusInternalServerError, err)
+		return
+	}
+	log.Printf("audit source changed from %s: %+v", r.RemoteAddr, cfg)
+	writeJSON(w, http.StatusOK, a.audit.Status())
+}
+
+// GET /api/files: the log files of the "Files" source (Settings > Sources).
+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" (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
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&cfg); err != nil {
+		writeErr(w, http.StatusBadRequest, err)
+		return
+	}
+	if err := a.files.Configure(cfg); err != nil {
+		writeErr(w, http.StatusBadRequest, err)
+		return
+	}
+	log.Printf("file sources changed from %s: %d files", r.RemoteAddr, len(cfg.Sources))
+	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())
+}
+
+// PUT /api/selflog {"enabled": bool, "min_level": "warning"}
+func (a *API) selfLogConfigure(w http.ResponseWriter, r *http.Request) {
+	var cfg selfLogConfig
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&cfg); err != nil {
+		writeErr(w, http.StatusBadRequest, err)
+		return
+	}
+	if err := a.selfLog.Configure(cfg); err != nil {
+		writeErr(w, http.StatusInternalServerError, err)
+		return
+	}
+	log.Printf("internal logs changed from %s: %+v", r.RemoteAddr, cfg)
+	writeJSON(w, http.StatusOK, a.selfLog.Status())
+}
+
+// GET /api/dmesg: state of the kernel messages source (Settings > Sources).
+func (a *API) dmesgStatus(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, a.dmesg.Status())
+}
+
+// PUT /api/dmesg {"enabled": bool}
+func (a *API) dmesgConfigure(w http.ResponseWriter, r *http.Request) {
+	var cfg dmesgConfig
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&cfg); err != nil {
+		writeErr(w, http.StatusBadRequest, err)
+		return
+	}
+	if err := a.dmesg.Configure(cfg); err != nil {
+		writeErr(w, http.StatusInternalServerError, err)
+		return
+	}
+	log.Printf("kernel messages source changed from %s: %+v", r.RemoteAddr, cfg)
+	writeJSON(w, http.StatusOK, a.dmesg.Status())
 }
 
 // GET /api/hostlogs: state of the host system logs source (Settings > Sources).
@@ -423,6 +533,41 @@ func (a *API) purge(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, map[string]string{"task_id": id})
 }
 
+// PUT /api/tags [tags]: replaces the whole list. On an invalid tag, "index"
+// gives its position and nothing is saved.
+func (a *API) replaceTags(w http.ResponseWriter, r *http.Request) {
+	var in []Tag
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&in); err != nil {
+		writeErr(w, http.StatusBadRequest, err)
+		return
+	}
+	tags, err := a.tags.Replace(in)
+	var le *tagListError
+	switch {
+	case errors.As(err, &le):
+		body := map[string]any{"error": err.Error(), "index": le.index}
+		var ce *codedError
+		if errors.As(err, &ce) {
+			body["code"] = ce.code
+			if ce.detail != "" {
+				body["detail"] = ce.detail
+			}
+		}
+		writeJSON(w, http.StatusBadRequest, body)
+	case errors.Is(err, errTooManyTags):
+		writeErr(w, http.StatusBadRequest, err)
+	case err != nil:
+		writeErr(w, http.StatusInternalServerError, err)
+	default:
+		writeJSON(w, http.StatusOK, tags)
+	}
+}
+
+// GET /api/tags/defaults: the tags "Restore default tags" shows before saving.
+func (a *API) defaultTags(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, defaultTags())
+}
+
 func (a *API) resetTags(w http.ResponseWriter, r *http.Request) {
 	tags, err := a.tags.Reset()
 	if err != nil {

audit.go

@@ -0,0 +1,830 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+	"unicode/utf8"
+)
+
+// auditConfig is saved in /data/audit.json and edited in Settings > Sources.
+type auditConfig struct {
+	Enabled bool   `json:"enabled"`
+	Path    string `json:"path"` // empty: AUDIT_LOG (/var/log/audit/audit.log)
+}
+
+// auditPos is where reading resumes: the inode of the file and an offset
+// before which every event was sent.
+type auditPos struct {
+	Ino uint64 `json:"ino"`
+	Off int64  `json:"off"`
+}
+
+// Audit follows the log of the Linux audit daemon (auditd) and sends one
+// entry per audit event: the records of an event (SYSCALL, EXECVE, PATH…)
+// share the same msg=audit(time:serial) and are grouped together.
+type Audit struct {
+	defPath   string
+	hostRoot  string // to read the host's /etc/passwd and /etc/hostname when mounted
+	hostName  string // AUDIT_HOST: host name of the entries when records have no node=
+	sink      func(*Entry)
+	backfill  time.Duration
+	cfgPath   string
+	statePath string
+	wake      chan struct{}
+
+	// Owned by the Run goroutine.
+	pos       auditPos // checkpoint saved in /data
+	readOff   int64    // where the next read starts (after the lines of open events)
+	known     bool
+	dirty     bool
+	notBefore time.Time
+	asm       *auditAssembler
+	users     *userNames
+
+	mu        sync.Mutex
+	cfg       auditConfig
+	reset     bool
+	open      bool
+	read      uint64
+	errCode   string
+	errDetail string
+}
+
+func NewAudit(defPath, hostRoot, hostName, dataDir string, backfill time.Duration, sink func(*Entry)) *Audit {
+	a := &Audit{
+		defPath:   defPath,
+		hostRoot:  hostRoot,
+		hostName:  hostName,
+		sink:      sink,
+		backfill:  backfill,
+		cfgPath:   filepath.Join(dataDir, "audit.json"),
+		statePath: filepath.Join(dataDir, "audit-state.json"),
+		wake:      make(chan struct{}, 1),
+		asm:       newAuditAssembler(),
+		users:     &userNames{path: filepath.Join(hostRoot, "etc/passwd")},
+	}
+	if b, err := os.ReadFile(a.cfgPath); err == nil {
+		_ = json.Unmarshal(b, &a.cfg)
+	}
+	if b, err := os.ReadFile(a.statePath); err == nil && json.Unmarshal(b, &a.pos) == nil {
+		a.known, a.readOff = true, a.pos.Off
+	}
+	if a.hostName == "" {
+		a.hostName = readTrim(filepath.Join(hostRoot, "etc/hostname"))
+	}
+	return a
+}
+
+func (a *Audit) path() string {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if a.cfg.Path != "" {
+		return a.cfg.Path
+	}
+	return a.defPath
+}
+
+// Run polls the file every second while the source is enabled.
+func (a *Audit) Run(ctx context.Context) {
+	tick := time.NewTicker(time.Second)
+	defer tick.Stop()
+	n := 0
+	for {
+		select {
+		case <-ctx.Done():
+			a.emit(a.asm.flushAll())
+			a.checkpoint()
+			a.saveState()
+			return
+		case <-tick.C:
+		case <-a.wake:
+		}
+		a.mu.Lock()
+		enabled, reset := a.cfg.Enabled, a.reset
+		a.reset = false
+		a.mu.Unlock()
+		if reset {
+			// Disabled then enabled again (or another file): start over from
+			// AUDIT_BACKFILL ago rather than reading everything in between.
+			a.asm = newAuditAssembler()
+			a.pos, a.readOff, a.known, a.dirty = auditPos{}, 0, false, true
+		}
+		if enabled {
+			a.poll(time.Now())
+		}
+		if n++; n%5 == 0 || reset {
+			a.saveState()
+		}
+	}
+}
+
+func (a *Audit) saveState() {
+	if !a.dirty {
+		return
+	}
+	a.dirty = false
+	if err := writeJSONFile(a.statePath, a.pos); err != nil {
+		logEvent("audit", 4, "saving position: %v", err)
+	}
+}
+
+func (a *Audit) fail(err error) {
+	code := "read"
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+		code = "not_found"
+	case errors.Is(err, fs.ErrPermission):
+		code = "permission"
+	}
+	a.mu.Lock()
+	changed := code != a.errCode
+	a.open, a.errCode, a.errDetail = false, code, err.Error()
+	a.mu.Unlock()
+	if changed {
+		logEvent("audit", 4, "cannot read the audit log: %v", err)
+	}
+}
+
+const maxAuditRead = 4 << 20 // per poll
+
+// poll reads the lines added since the previous one, follows rotations and
+// sends the events that are complete.
+func (a *Audit) poll(now time.Time) {
+	path := a.path()
+	fi, err := os.Stat(path)
+	if err != nil {
+		a.fail(err)
+		return
+	}
+	ino := inodeOf(fi)
+	if !a.known {
+		// First read: the last AUDIT_BACKFILL (0: only new events).
+		a.pos, a.readOff, a.known, a.dirty = auditPos{Ino: ino}, 0, true, true
+		a.notBefore = time.Time{}
+		if a.backfill <= 0 {
+			a.readOff = fi.Size()
+		} else {
+			a.notBefore = now.Add(-a.backfill)
+		}
+	}
+	if a.pos.Ino != ino {
+		// Rotated by auditd (audit.log -> audit.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) == a.pos.Ino {
+			for {
+				off, more, err := a.readFrom(path+".1", a.readOff, now)
+				if err != nil || !more {
+					break
+				}
+				a.readOff = off
+			}
+		}
+		a.emit(a.asm.flushAll())
+		a.pos, a.readOff, a.dirty = auditPos{Ino: ino}, 0, true
+	} else if fi.Size() < a.readOff {
+		a.emit(a.asm.flushAll()) // truncated
+		a.readOff = 0
+	}
+	if fi.Size() > a.readOff {
+		off, _, err := a.readFrom(path, a.readOff, now)
+		if err != nil {
+			a.fail(err)
+			return
+		}
+		a.readOff = off
+	}
+	a.mu.Lock()
+	if !a.open {
+		a.open, a.errCode, a.errDetail = true, "", ""
+	}
+	a.mu.Unlock()
+	a.emit(a.asm.flush(now, auditCloseDelay))
+	a.checkpoint()
+}
+
+// checkpoint saves an offset before which every event is sent: the start of
+// the oldest event still open, or the read offset.
+func (a *Audit) checkpoint() {
+	off := a.readOff
+	if o, ok := a.asm.minOff(); ok && o < off {
+		off = o
+	}
+	if off != a.pos.Off {
+		a.pos.Off, a.dirty = off, true
+	}
+}
+
+// readFrom feeds the complete lines written after off to the assembler and
+// returns the new offset; more tells whether the read window was full.
+func (a *Audit) readFrom(path string, off int64, now time.Time) (int64, bool, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return off, false, err
+	}
+	defer f.Close()
+	buf, err := io.ReadAll(io.NewSectionReader(f, off, maxAuditRead))
+	if err != nil {
+		return off, false, err
+	}
+	full := len(buf) == maxAuditRead
+	end := bytes.LastIndexByte(buf, '\n')
+	if end < 0 {
+		if !full {
+			return off, false, nil // partial line: wait for its end
+		}
+		end = len(buf) - 1
+	}
+	start := off
+	for _, line := range bytes.Split(buf[:end], []byte{'\n'}) {
+		a.emit(a.asm.add(string(line), start, now))
+		start += int64(len(line)) + 1
+	}
+	return off + int64(end) + 1, full, nil
+}
+
+func (a *Audit) emit(evs []*auditEvent) {
+	for _, ev := range evs {
+		if !a.notBefore.IsZero() && ev.Time.Before(a.notBefore) {
+			continue
+		}
+		a.sink(auditEntry(ev, a.users, a.hostName, time.Now()))
+		a.mu.Lock()
+		a.read++
+		a.mu.Unlock()
+	}
+}
+
+func inodeOf(fi os.FileInfo) uint64 {
+	if st, ok := fi.Sys().(*syscall.Stat_t); ok {
+		return uint64(st.Ino)
+	}
+	return 0
+}
+
+// Configure saves and applies the configuration.
+func (a *Audit) Configure(cfg auditConfig) error {
+	cfg.Path = strings.TrimSpace(cfg.Path)
+	a.mu.Lock()
+	if (a.cfg.Enabled && !cfg.Enabled) || cfg.Path != a.cfg.Path {
+		a.reset = true
+		a.open, a.errCode, a.errDetail = false, "", ""
+	}
+	a.cfg = cfg
+	a.mu.Unlock()
+	select {
+	case a.wake <- struct{}{}:
+	default:
+	}
+	return writeJSONFile(a.cfgPath, cfg)
+}
+
+// Status is the state shown in Settings > Sources.
+func (a *Audit) Status() map[string]any {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return map[string]any{
+		"enabled":     a.cfg.Enabled,
+		"path":        a.cfg.Path,
+		"defaultPath": a.defPath,
+		"open":        a.open && a.cfg.Enabled,
+		"read":        a.read,
+		"code":        a.errCode,
+		"error":       a.errDetail,
+	}
+}
+
+/* ---------- Records and events ---------- */
+
+// auditCloseDelay: an event is complete when no line of it came for this
+// long, or when the log has moved this far past its time (auditd does not
+// write the EOE record that ends multi-record events).
+const auditCloseDelay = 2 * time.Second
+
+type auditField struct {
+	k, v string
+	bare bool // written without quotes: may be hexadecimal
+}
+
+type auditRecord struct {
+	Type   string
+	Fields []auditField
+	Raw    string
+}
+
+type auditEvent struct {
+	ID      string // "1700000000.123:456"
+	Serial  string
+	Time    time.Time
+	Node    string
+	Records []auditRecord
+	off     int64     // offset of its first line in the file
+	seen    time.Time // when its last line was read
+	order   int
+}
+
+// parseAuditLine decodes "[node=N ]type=T msg=audit(SEC.MS:SERIAL): k=v …".
+func parseAuditLine(line string) (rec auditRecord, id, node string, ok bool) {
+	line = strings.TrimRight(line, "\r")
+	rec.Raw = line
+	head, rest, found := strings.Cut(line, "): ")
+	if !found {
+		head, found = strings.CutSuffix(line, ")")
+		if !found {
+			return rec, "", "", false
+		}
+	}
+	pre, ts, found := strings.Cut(head, "msg=audit(")
+	if !found || ts == "" {
+		return rec, "", "", false
+	}
+	for _, f := range parseAuditKV(pre) {
+		switch f.k {
+		case "type":
+			rec.Type = f.v
+		case "node":
+			node = f.v
+		}
+	}
+	if rec.Type == "" {
+		return rec, "", "", false
+	}
+	// Enriched format (log_format=ENRICHED): resolved values after a GS byte.
+	main, enriched, _ := strings.Cut(rest, "\x1d")
+	rec.Fields = parseAuditKV(main)
+	for _, f := range parseAuditKV(enriched) {
+		rec.Fields = append(rec.Fields, auditField{k: "@" + f.k, v: f.v})
+	}
+	return rec, ts, node, true
+}
+
+// hexFields are the fields auditd writes in hexadecimal when they contain
+// spaces, quotes or control characters (and then without quotes).
+var hexFields = map[string]bool{
+	"proctitle": true, "cmd": true, "exe": true, "comm": true, "key": true, "acct": true,
+	"name": true, "cwd": true, "data": true, "path": true, "old": true, "new": true,
+}
+
+// parseAuditKV splits "k=v k2="v 2" msg='k3=v3 …'"; the fields of a quoted
+// msg='…' (user space records) follow the outer ones.
+func parseAuditKV(s string) []auditField {
+	var out []auditField
+	var inner []auditField
+	for i := 0; i < len(s); {
+		for i < len(s) && s[i] == ' ' {
+			i++
+		}
+		eq := strings.IndexByte(s[i:], '=')
+		if eq <= 0 {
+			break
+		}
+		k := s[i : i+eq]
+		if sp := strings.IndexByte(k, ' '); sp >= 0 { // word without value
+			i += sp + 1
+			continue
+		}
+		i += eq + 1
+		var v string
+		quoted := i < len(s) && (s[i] == '"' || s[i] == '\'')
+		if quoted {
+			q := s[i]
+			end := strings.IndexByte(s[i+1:], q)
+			if end < 0 {
+				end = len(s) - i - 1
+			}
+			v = s[i+1 : i+1+end]
+			i += end + 2
+		} else {
+			end := strings.IndexByte(s[i:], ' ')
+			if end < 0 {
+				end = len(s) - i
+			}
+			v = s[i : i+end]
+			i += end
+		}
+		if k == "msg" && quoted && strings.Contains(v, "=") {
+			inner = append(inner, parseAuditKV(v)...)
+			continue
+		}
+		if !quoted && hexFields[k] {
+			v = decodeHex(k, v)
+		}
+		if !quoted && (v == "(null)" || v == "?") {
+			v = ""
+		}
+		out = append(out, auditField{k: k, v: v, bare: !quoted})
+	}
+	return append(out, inner...)
+}
+
+// decodeHex decodes a hexadecimal value, or returns it unchanged.
+func decodeHex(k, v string) string {
+	if len(v) < 2 || len(v)%2 != 0 {
+		return v
+	}
+	b, err := hex.DecodeString(v)
+	if err != nil {
+		return v
+	}
+	if k == "proctitle" {
+		b = bytes.ReplaceAll(bytes.TrimRight(b, "\x00"), []byte{0}, []byte{' '})
+	}
+	if !utf8.Valid(b) {
+		return v
+	}
+	return string(b)
+}
+
+func isDigits(s string) bool {
+	for i := 0; i < len(s); i++ {
+		if s[i] < '0' || s[i] > '9' {
+			return false
+		}
+	}
+	return s != ""
+}
+
+// parseAuditTime decodes "1700000000.123:456" into a time and a serial.
+func parseAuditTime(id string) (time.Time, string) {
+	ts, serial, _ := strings.Cut(id, ":")
+	sec, frac, _ := strings.Cut(ts, ".")
+	s, err := strconv.ParseInt(sec, 10, 64)
+	if err != nil {
+		return time.Time{}, serial
+	}
+	ms, _ := strconv.ParseInt((frac + "000")[:3], 10, 64)
+	return time.Unix(s, ms*int64(time.Millisecond)), serial
+}
+
+// auditAssembler groups the records of each event.
+type auditAssembler struct {
+	open   map[string]*auditEvent
+	newest time.Time // most recent event time read
+	n      int
+}
+
+func newAuditAssembler() *auditAssembler {
+	return &auditAssembler{open: map[string]*auditEvent{}}
+}
+
+const maxOpenAudit = 2000
+
+// add takes one line read at offset off and returns the events it completes
+// (EOE record, or too many events open).
+func (a *auditAssembler) add(line string, off int64, now time.Time) []*auditEvent {
+	if strings.TrimSpace(line) == "" {
+		return nil
+	}
+	rec, id, node, ok := parseAuditLine(line)
+	if !ok {
+		return nil
+	}
+	ev := a.open[id]
+	if rec.Type == "EOE" {
+		if ev == nil {
+			return nil
+		}
+		delete(a.open, id)
+		return []*auditEvent{ev}
+	}
+	if ev == nil {
+		t, serial := parseAuditTime(id)
+		a.n++
+		ev = &auditEvent{ID: id, Serial: serial, Time: t, Node: node, off: off, order: a.n}
+		a.open[id] = ev
+		if t.After(a.newest) {
+			a.newest = t
+		}
+	}
+	ev.Records = append(ev.Records, rec)
+	ev.seen = now
+	if len(a.open) > maxOpenAudit {
+		return a.closeOldest(len(a.open) - maxOpenAudit/2)
+	}
+	return nil
+}
+
+// flush returns the events with no new line for delay, or that the log has
+// left delay behind, in the order they started.
+func (a *auditAssembler) flush(now time.Time, delay time.Duration) []*auditEvent {
+	var out []*auditEvent
+	for id, ev := range a.open {
+		if now.Sub(ev.seen) >= delay || a.newest.Sub(ev.Time) >= delay {
+			out = append(out, ev)
+			delete(a.open, id)
+		}
+	}
+	sortEvents(out)
+	return out
+}
+
+func (a *auditAssembler) flushAll() []*auditEvent {
+	return a.closeOldest(len(a.open))
+}
+
+func (a *auditAssembler) closeOldest(n int) []*auditEvent {
+	all := make([]*auditEvent, 0, len(a.open))
+	for _, ev := range a.open {
+		all = append(all, ev)
+	}
+	sortEvents(all)
+	all = all[:n]
+	for _, ev := range all {
+		delete(a.open, ev.ID)
+	}
+	return all
+}
+
+func sortEvents(evs []*auditEvent) {
+	sort.Slice(evs, func(i, j int) bool { return evs[i].order < evs[j].order })
+}
+
+func (a *auditAssembler) minOff() (int64, bool) {
+	var m int64
+	ok := false
+	for _, ev := range a.open {
+		if !ok || ev.off < m {
+			m, ok = ev.off, true
+		}
+	}
+	return m, ok
+}
+
+/* ---------- Entries ---------- */
+
+// get returns the first non-empty value of a field across the records.
+func (ev *auditEvent) get(k string) string {
+	for _, r := range ev.Records {
+		for _, f := range r.Fields {
+			if f.k == k && f.v != "" {
+				return f.v
+			}
+		}
+	}
+	return ""
+}
+
+func (ev *auditEvent) record(typ string) *auditRecord {
+	for i := range ev.Records {
+		if ev.Records[i].Type == typ {
+			return &ev.Records[i]
+		}
+	}
+	return nil
+}
+
+// result is "success", "failed" or "" from res= or success=.
+func (ev *auditEvent) result() string {
+	v := strings.ToLower(firstNonEmpty(ev.get("res"), ev.get("success"), ev.get("result")))
+	switch v {
+	case "success", "yes", "1":
+		return "success"
+	case "failed", "fail", "failure", "no", "0":
+		return "failed"
+	}
+	return ""
+}
+
+const auditUnset = "4294967295"
+
+// userOf resolves a uid: enriched name, then the host's /etc/passwd, then the number.
+func (ev *auditEvent) userOf(field string, users *userNames) string {
+	uid := ev.get(field)
+	if uid == "" || uid == auditUnset || uid == "-1" {
+		return ""
+	}
+	if n := strings.Trim(ev.get("@"+strings.ToUpper(field)), `"`); n != "" && n != "unset" && n != uid {
+		return n
+	}
+	if n := users.name(uid); n != "" {
+		return n
+	}
+	return uid
+}
+
+// auditEntry turns an event into an Entry: app is the type of its first record.
+func auditEntry(ev *auditEvent, users *userNames, hostName string, now time.Time) *Entry {
+	typ := "UNKNOWN"
+	if len(ev.Records) > 0 {
+		typ = ev.Records[0].Type
+	}
+	res := ev.result()
+	auid := ev.get("auid")
+	user := ev.userOf("auid", users)
+	if user == "" {
+		user = ev.userOf("uid", users)
+	}
+	sev := 6
+	if res == "failed" {
+		sev = 4
+	}
+	raw := make([]string, len(ev.Records))
+	for i, r := range ev.Records {
+		raw[i] = r.Raw
+	}
+	host := firstNonEmpty(ev.Node, hostName, "localhost")
+	e := &Entry{
+		Time:       ev.Time,
+		Received:   now,
+		Host:       host,
+		App:        typ,
+		ProcID:     ev.get("pid"),
+		Facility:   "authpriv",
+		Severity:   severityNames[sev],
+		SevNum:     sev,
+		Message:    auditSummary(ev, typ, user, users, res),
+		Proto:      "file",
+		SourceType: "audit",
+		Extra: map[string]string{
+			"audit_type":   typ,
+			"audit_serial": ev.Serial,
+			"auid":         auid,
+			"user":         user,
+			"exe":          ev.get("exe"),
+			"res":          res,
+			"key":          ev.get("key"),
+			"audit_raw":    strings.Join(raw, "\n"),
+		},
+		Wait: true,
+	}
+	if e.Time.IsZero() {
+		e.Time = now
+	}
+	return e
+}
+
+// auditSummary writes "who: what [key] → result".
+func auditSummary(ev *auditEvent, typ, user string, users *userNames, res string) string {
+	who := firstNonEmpty(user, "?")
+	// Syscalls: the login user and, when different, the user the process ran
+	// as (sudo, su). In user space records, uid is the daemon's (sshd: root).
+	if uid := ev.userOf("uid", users); ev.get("syscall") != "" && uid != "" && user != "" && uid != user {
+		who += " (as " + uid + ")"
+	}
+	var what string
+	exe := ev.get("exe")
+	switch {
+	case typ == "USER_CMD":
+		what = "sudo " + firstNonEmpty(ev.get("cmd"), "?")
+		if cwd := ev.get("cwd"); cwd != "" {
+			what += " (in " + cwd + ")"
+		}
+	case strings.HasPrefix(typ, "USER_") || strings.HasPrefix(typ, "CRED_") || typ == "LOGIN" ||
+		typ == "ADD_USER" || typ == "DEL_USER" || typ == "ADD_GROUP" || typ == "DEL_GROUP":
+		if acct := ev.get("acct"); acct != "" {
+			who = acct
+		}
+		what = firstNonEmpty(ev.get("op"), strings.ToLower(typ))
+		if exe != "" {
+			what += " via " + exe
+		}
+		if addr := firstNonEmpty(ev.get("hostname"), ev.get("addr")); addr != "" {
+			what += " from " + addr
+		}
+		if term := ev.get("terminal"); term != "" && term != "?" {
+			what += " on " + term
+		}
+	case ev.record("EXECVE") != nil:
+		what = "exec " + execLine(ev.record("EXECVE"))
+	case ev.get("syscall") != "":
+		what = "syscall " + firstNonEmpty(ev.get("@SYSCALL"), ev.get("syscall"))
+		if name := pathName(ev); name != "" {
+			what += " " + name
+		}
+		if prog := firstNonEmpty(exe, ev.get("comm")); prog != "" {
+			what += " by " + prog
+		}
+	default:
+		what = strings.ToLower(typ)
+		if op := ev.get("op"); op != "" {
+			what += " " + op
+		}
+		if exe != "" {
+			what += " by " + exe
+		}
+	}
+	if key := ev.get("key"); key != "" {
+		what += " [" + key + "]"
+	}
+	if res != "" {
+		what += " → " + res
+	}
+	return who + ": " + what
+}
+
+// execLine joins the arguments of an EXECVE record (a0, a1…).
+func execLine(r *auditRecord) string {
+	args := map[int]string{}
+	argc := 0
+	for _, f := range r.Fields {
+		if f.k == "argc" {
+			argc, _ = strconv.Atoi(f.v)
+		} else if len(f.k) > 1 && f.k[0] == 'a' && isDigits(f.k[1:]) {
+			n, _ := strconv.Atoi(f.k[1:])
+			if f.bare {
+				args[n] = decodeHex(f.k, f.v) // arguments with spaces or quotes
+			} else {
+				args[n] = f.v
+			}
+		}
+	}
+	if argc == 0 {
+		argc = len(args)
+	}
+	parts := make([]string, 0, argc)
+	for i := 0; i < argc; i++ {
+		if v, ok := args[i]; ok {
+			parts = append(parts, v)
+		}
+	}
+	s := strings.Join(parts, " ")
+	if len(s) > 300 {
+		s = s[:300] + "…"
+	}
+	return s
+}
+
+// pathName is the file a syscall worked on: the first PATH record that is not
+// its parent directory.
+func pathName(ev *auditEvent) string {
+	var first string
+	for _, r := range ev.Records {
+		if r.Type != "PATH" {
+			continue
+		}
+		var name, nametype string
+		for _, f := range r.Fields {
+			switch f.k {
+			case "name":
+				name = f.v
+			case "nametype":
+				nametype = f.v
+			}
+		}
+		if name == "" {
+			continue
+		}
+		if nametype != "PARENT" {
+			return name
+		}
+		if first == "" {
+			first = name
+		}
+	}
+	return first
+}
+
+/* ---------- User names ---------- */
+
+// userNames reads uid -> name from the host's /etc/passwd when it is mounted,
+// at most once a minute.
+type userNames struct {
+	path    string
+	names   map[string]string
+	loaded  time.Time
+	modTime time.Time
+}
+
+func (u *userNames) name(uid string) string {
+	if u == nil {
+		return ""
+	}
+	if time.Since(u.loaded) > time.Minute {
+		u.loaded = time.Now()
+		if fi, err := os.Stat(u.path); err == nil && !fi.ModTime().Equal(u.modTime) {
+			u.modTime = fi.ModTime()
+			u.names = readPasswd(u.path)
+		}
+	}
+	return u.names[uid]
+}
+
+func readPasswd(path string) map[string]string {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil
+	}
+	defer f.Close()
+	names := map[string]string{}
+	sc := bufio.NewScanner(f)
+	for sc.Scan() {
+		p := strings.Split(sc.Text(), ":")
+		if len(p) >= 3 && p[0] != "" && !strings.HasPrefix(p[0], "#") {
+			names[p[2]] = p[0]
+		}
+	}
+	return names
+}

audit_test.go

@@ -0,0 +1,356 @@
+package main
+
+import (
+	"context"
+	"encoding/hex"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func hexOf(s string) string { return strings.ToUpper(hex.EncodeToString([]byte(s))) }
+
+// Lines of real audit.log events (trimmed).
+var (
+	auditExec = []string{
+		`type=SYSCALL msg=audit(1759572000.123:4242): arch=c000003e syscall=59 success=yes exit=0 a0=4142 a1=55d2 a2=55d3 a3=0 items=2 ppid=1000 pid=1234 auid=1000 uid=0 gid=0 euid=0 tty=pts0 ses=3 comm="apt" exe="/usr/bin/apt" subj=unconfined key="exec"`,
+		`type=EXECVE msg=audit(1759572000.123:4242): argc=3 a0="apt" a1="install" a2=` + hexOf(`my "pkg"`),
+		`type=CWD msg=audit(1759572000.123:4242): cwd="/root"`,
+		`type=PATH msg=audit(1759572000.123:4242): item=0 name="/usr/bin/apt" inode=1 dev=fd:00 mode=0100755 ouid=0 ogid=0 nametype=NORMAL`,
+		`type=PROCTITLE msg=audit(1759572000.123:4242): proctitle=` + hexOf("apt\x00install\x00my \"pkg\""),
+	}
+	auditSudo = `type=USER_CMD msg=audit(1759572001.500:4250): pid=2000 uid=1000 auid=1000 ses=3 subj=unconfined msg='cwd="/home/cedric" cmd=` + hexOf("systemctl restart nginx") + ` exe="/usr/bin/sudo" terminal=pts/0 res=success'`
+	auditSSH  = `type=USER_AUTH msg=audit(1759572002.000:4260): pid=3000 uid=0 auid=4294967295 ses=4294967295 subj=unconfined msg='op=PAM:authentication grantors=? acct="bob" exe="/usr/sbin/sshd" hostname=203.0.113.9 addr=203.0.113.9 terminal=ssh res=failed'`
+	auditOpen = []string{
+		"node=web1 type=SYSCALL msg=audit(1759572003.250:4270): arch=c000003e syscall=257 success=no exit=-13 items=1 pid=4000 auid=1001 uid=1001 comm=\"cat\" exe=\"/usr/bin/cat\" key=\"shadow\"\x1dARCH=x86_64 SYSCALL=openat AUID=\"alice\" UID=\"alice\"",
+		`node=web1 type=PATH msg=audit(1759572003.250:4270): item=0 name="/etc/shadow" inode=7 nametype=NORMAL`,
+	}
+)
+
+func TestParseAuditLine(t *testing.T) {
+	rec, id, node, ok := parseAuditLine(auditOpen[0])
+	if !ok || rec.Type != "SYSCALL" || id != "1759572003.250:4270" || node != "web1" {
+		t.Fatalf("got %+v %q %q %v", rec, id, node, ok)
+	}
+	ev := &auditEvent{Records: []auditRecord{rec}}
+	for k, want := range map[string]string{"syscall": "257", "exe": "/usr/bin/cat", "key": "shadow", "@SYSCALL": "openat", "@AUID": "alice"} {
+		if got := ev.get(k); got != want {
+			t.Errorf("%s = %q, want %q", k, got, want)
+		}
+	}
+	for _, bad := range []string{"", "garbage", "type=X no header", "msg=audit(1.2:3): no type"} {
+		if _, _, _, ok := parseAuditLine(bad); ok {
+			t.Errorf("%q parsed", bad)
+		}
+	}
+}
+
+func TestParseAuditKV(t *testing.T) {
+	tests := []struct {
+		in   string
+		want map[string]string
+	}{
+		{`a=1 b="two words" c=`, map[string]string{"a": "1", "b": "two words", "c": ""}},
+		{`pid=1 msg='op=login acct="bob" res=success'`, map[string]string{"pid": "1", "op": "login", "acct": "bob", "res": "success"}},
+		{`comm=` + hexOf("my prog") + ` exe="/bin/x"`, map[string]string{"comm": "my prog", "exe": "/bin/x"}},
+		{`key=(null) hostname=? terminal=?`, map[string]string{"key": "", "hostname": "", "terminal": ""}},
+		{`a0=4142 name=ABC`, map[string]string{"a0": "4142", "name": "ABC"}}, // not hexadecimal fields, or odd length
+		{`avc:  denied  { read } for  pid=12 comm="x"`, map[string]string{"pid": "12", "comm": "x"}},
+	}
+	for _, tt := range tests {
+		got := map[string]string{}
+		for _, f := range parseAuditKV(tt.in) {
+			got[f.k] = f.v
+		}
+		for k, v := range tt.want {
+			if got[k] != v {
+				t.Errorf("%q: %s = %q, want %q (all: %v)", tt.in, k, got[k], v, got)
+			}
+		}
+	}
+}
+
+func TestAuditAssembler(t *testing.T) {
+	now := time.Unix(1759572010, 0)
+	a := newAuditAssembler()
+	// Two interleaved events, then an EOE that closes the first one.
+	lines := []string{auditExec[0], auditSudo, auditExec[1], auditExec[2]}
+	off := int64(0)
+	for _, l := range lines {
+		if out := a.add(l, off, now); out != nil {
+			t.Fatalf("closed early: %v", out)
+		}
+		off += int64(len(l)) + 1
+	}
+	if len(a.open) != 2 {
+		t.Fatalf("%d open events", len(a.open))
+	}
+	if o, ok := a.minOff(); !ok || o != 0 {
+		t.Fatalf("minOff %d %v", o, ok)
+	}
+	out := a.add(`type=EOE msg=audit(1759572000.123:4242): `, off, now)
+	if len(out) != 1 || out[0].Serial != "4242" || len(out[0].Records) != 3 {
+		t.Fatalf("EOE: %+v", out)
+	}
+	if o, _ := a.minOff(); o != int64(len(lines[0])+1) {
+		t.Fatalf("minOff after EOE: %d", o)
+	}
+	// An EOE of an unknown event is ignored.
+	if out := a.add(`type=EOE msg=audit(1.0:1): `, 0, now); out != nil {
+		t.Fatal(out)
+	}
+	// The sudo event is closed by time.
+	if out := a.flush(now.Add(time.Second), 2*time.Second); len(out) != 0 {
+		t.Fatalf("closed too early: %v", out)
+	}
+	out = a.flush(now.Add(3*time.Second), 2*time.Second)
+	if len(out) != 1 || out[0].Records[0].Type != "USER_CMD" {
+		t.Fatalf("flush: %+v", out)
+	}
+	// Events the log has left behind are closed without waiting.
+	a.add(auditSSH, 0, now)
+	a.add(auditOpen[0], 0, now)
+	out = a.flush(now, 1*time.Second)
+	if len(out) != 1 || out[0].Serial != "4260" || len(a.open) != 1 {
+		t.Fatalf("newest: %+v, open %d", out, len(a.open))
+	}
+	if out = a.flushAll(); len(out) != 1 || out[0].Serial != "4270" || len(a.open) != 0 {
+		t.Fatalf("flushAll: %+v", out)
+	}
+}
+
+func auditEv(t *testing.T, lines ...string) *auditEvent {
+	t.Helper()
+	a := newAuditAssembler()
+	for _, l := range lines {
+		a.add(l, 0, time.Time{})
+	}
+	out := a.flushAll()
+	if len(out) != 1 {
+		t.Fatalf("%d events", len(out))
+	}
+	return out[0]
+}
+
+func TestAuditEntry(t *testing.T) {
+	dir := t.TempDir()
+	passwd := filepath.Join(dir, "passwd")
+	if err := os.WriteFile(passwd, []byte("root:x:0:0::/root:/bin/sh\ncedric:x:1000:1000::/home/cedric:/bin/bash\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Unix(1759572100, 0)
+	tests := []struct {
+		name   string
+		lines  []string
+		host   string
+		app    string
+		sev    string
+		msg    string
+		fields map[string]string
+	}{
+		{"exec", auditExec, "srv", "SYSCALL", "info",
+			`cedric (as root): exec apt install my "pkg" [exec] → success`,
+			map[string]string{"audit_type": "SYSCALL", "audit_serial": "4242", "auid": "1000", "user": "cedric", "exe": "/usr/bin/apt", "res": "success", "key": "exec"}},
+		{"sudo", []string{auditSudo}, "srv", "USER_CMD", "info",
+			"cedric: sudo systemctl restart nginx (in /home/cedric) → success",
+			map[string]string{"user": "cedric", "exe": "/usr/bin/sudo", "res": "success", "key": ""}},
+		{"ssh failed", []string{auditSSH}, "srv", "USER_AUTH", "warning",
+			"bob: PAM:authentication via /usr/sbin/sshd from 203.0.113.9 on ssh → failed",
+			map[string]string{"auid": "4294967295", "user": "root", "res": "failed"}},
+		{"enriched, node", auditOpen, "web1", "SYSCALL", "warning",
+			"alice: syscall openat /etc/shadow by /usr/bin/cat [shadow] → failed",
+			map[string]string{"user": "alice", "auid": "1001", "res": "failed", "key": "shadow"}},
+		{"unknown uid", []string{`type=USER_LOGIN msg=audit(1759572004.000:4280): pid=1 uid=0 auid=4242 ses=9 msg='op=login id=4242 exe="/usr/sbin/sshd" hostname=? addr=10.0.0.5 terminal=/dev/pts/1 res=success'`},
+			"srv", "USER_LOGIN", "info", "4242: login via /usr/sbin/sshd from 10.0.0.5 on /dev/pts/1 → success",
+			map[string]string{"user": "4242"}},
+		{"no result", []string{`type=CONFIG_CHANGE msg=audit(1759572005.000:4290): auid=1000 ses=3 op=add_rule key="watch" list=4`},
+			"srv", "CONFIG_CHANGE", "info", "cedric: config_change add_rule [watch]",
+			map[string]string{"res": ""}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ev := auditEv(t, tt.lines...)
+			e := auditEntry(ev, &userNames{path: passwd}, "srv", now)
+			if e.Host != tt.host || e.App != tt.app || e.Severity != tt.sev || e.SourceType != "audit" || !e.Wait {
+				t.Errorf("entry %+v", e)
+			}
+			if e.Message != tt.msg {
+				t.Errorf("message\n got %q\nwant %q", e.Message, tt.msg)
+			}
+			for k, v := range tt.fields {
+				if e.Extra[k] != v {
+					t.Errorf("%s = %q, want %q", k, e.Extra[k], v)
+				}
+			}
+			if e.Extra["audit_raw"] != strings.Join(tt.lines, "\n") {
+				t.Errorf("raw %q", e.Extra["audit_raw"])
+			}
+			if !e.Time.Equal(ev.Time) || e.Time.IsZero() {
+				t.Errorf("time %v", e.Time)
+			}
+		})
+	}
+}
+
+type entrySink struct {
+	mu sync.Mutex
+	es []*Entry
+}
+
+func (s *entrySink) add(e *Entry) {
+	s.mu.Lock()
+	s.es = append(s.es, e)
+	s.mu.Unlock()
+}
+
+func (s *entrySink) serials() []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []string
+	for _, e := range s.es {
+		out = append(out, e.Extra["audit_serial"])
+	}
+	return out
+}
+
+func appendLines(t *testing.T, path string, lines ...string) {
+	t.Helper()
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+	for _, l := range lines {
+		if _, err := f.WriteString(l + "\n"); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestAuditFollow(t *testing.T) {
+	dir := t.TempDir()
+	logPath := filepath.Join(dir, "audit.log")
+	data := filepath.Join(dir, "data")
+	if err := os.Mkdir(data, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	sink := &entrySink{}
+	newAuditT := func() *Audit {
+		a := NewAudit(logPath, dir, "srv", data, 100*365*24*time.Hour, sink.add)
+		a.cfg.Enabled = true
+		return a
+	}
+	a := newAuditT()
+	now := time.Unix(1759572100, 0)
+
+	// Missing file: reported, then read once it exists.
+	a.poll(now)
+	if st := a.Status(); st["code"] != "not_found" || st["open"] != false {
+		t.Fatalf("status %v", st)
+	}
+	appendLines(t, logPath, auditExec[:3]...)
+	appendLines(t, logPath, auditSudo)
+	a.poll(now)
+	if got := sink.serials(); len(got) != 0 { // events may still get records
+		t.Fatalf("after first poll: %v", got)
+	}
+	// A partial line is not read.
+	f, _ := os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0)
+	f.WriteString(auditSSH[:20])
+	f.Close()
+	a.poll(now.Add(5 * time.Second))
+	if got := sink.serials(); len(got) != 2 || got[0] != "4242" || got[1] != "4250" {
+		t.Fatalf("after second poll: %v", got)
+	}
+	if a.Status()["open"] != true || a.Status()["read"] != uint64(2) {
+		t.Fatalf("status %v", a.Status())
+	}
+	a.saveState()
+
+	// Restart: nothing is sent twice, the partial line is read once complete.
+	a = newAuditT()
+	f, _ = os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0)
+	f.WriteString(auditSSH[20:] + "\n")
+	f.Close()
+	a.poll(now.Add(10 * time.Second))
+	a.poll(now.Add(20 * time.Second))
+	if got := sink.serials(); len(got) != 3 || got[2] != "4260" {
+		t.Fatalf("after restart: %v", got)
+	}
+
+	// Rotation: lines added to the old file before the switch are not lost.
+	appendLines(t, logPath, auditOpen[0])
+	a.poll(now.Add(21 * time.Second)) // read, still open
+	appendLines(t, logPath, auditOpen[1])
+	if err := os.Rename(logPath, logPath+".1"); err != nil {
+		t.Fatal(err)
+	}
+	appendLines(t, logPath, strings.Replace(auditSudo, "4250", "4300", 1))
+	a.poll(now.Add(22 * time.Second))
+	a.poll(now.Add(30 * time.Second))
+	got := sink.serials()
+	if len(got) != 5 || got[3] != "4270" || got[4] != "4300" {
+		t.Fatalf("after rotation: %v", got)
+	}
+	sink.mu.Lock()
+	if n := len(sink.es[3].Extra["audit_raw"]); n == 0 || !strings.Contains(sink.es[3].Extra["audit_raw"], "/etc/shadow") {
+		t.Errorf("rotated event missing its PATH record: %q", sink.es[3].Extra["audit_raw"])
+	}
+	sink.mu.Unlock()
+}
+
+func TestAuditBackfillAndConfigure(t *testing.T) {
+	dir := t.TempDir()
+	logPath := filepath.Join(dir, "audit.log")
+	appendLines(t, logPath, auditExec...)
+	appendLines(t, logPath, auditSudo)
+	sink := &entrySink{}
+	// Only events newer than the backfill window are sent on the first read.
+	a := NewAudit(logPath, dir, "", dir, time.Hour, sink.add)
+	a.cfg.Enabled = true
+	a.poll(time.Unix(1759572001+7200, 0))
+	a.emit(a.asm.flushAll())
+	if got := sink.serials(); len(got) != 0 {
+		t.Fatalf("too old: %v", got)
+	}
+	a = NewAudit(logPath, dir, "", filepath.Join(dir, "other"), time.Hour, sink.add)
+	if err := os.Mkdir(filepath.Join(dir, "other"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	a.cfg.Enabled = true
+	a.poll(time.Unix(1759572001+60, 0))
+	a.emit(a.asm.flushAll())
+	if got := sink.serials(); len(got) != 2 {
+		t.Fatalf("backfill: %v", got)
+	}
+	if e := sink.es[0]; e.Host != "localhost" {
+		t.Errorf("host %q", e.Host)
+	}
+
+	// Run picks the configuration up; turning it off resets the position.
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { a.Run(ctx); close(done) }()
+	if err := a.Configure(auditConfig{Enabled: false}); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if b, err := os.ReadFile(filepath.Join(dir, "other", "audit-state.json")); err == nil && strings.Contains(string(b), `"off": 0`) {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+	<-done
+	if st := a.Status(); st["enabled"] != false || st["defaultPath"] != logPath {
+		t.Fatalf("status %v", st)
+	}
+	b, _ := os.ReadFile(filepath.Join(dir, "other", "audit-state.json"))
+	if !strings.Contains(string(b), `"off": 0`) {
+		t.Fatalf("state %s", b)
+	}
+}

dmesg.go

@@ -0,0 +1,481 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+	"unicode/utf8"
+)
+
+// dmesgConfig is saved in /data/dmesg.json and edited in Settings > Sources.
+type dmesgConfig struct {
+	Enabled bool `json:"enabled"`
+}
+
+// dmesgPos is where reading resumes: the last record sent, valid for one boot
+// only (the kernel numbers its records from 0 at each boot).
+type dmesgPos struct {
+	BootID string `json:"boot_id"`
+	Seq    uint64 `json:"seq"`
+}
+
+// Dmesg collects the kernel ring buffer of the machine hosting the stack by
+// reading /dev/kmsg, the device dmesg itself reads.
+type Dmesg struct {
+	device     string // /dev/kmsg
+	bootIDPath string // identifies the current boot
+	uptimePath string // gives the boot time, to date the records
+	sink       func(*Entry)
+	backfill   time.Duration
+	cfgPath    string
+	statePath  string
+	retry      time.Duration // delay before opening the device again after a failure
+	wake       chan struct{}
+
+	mu        sync.Mutex
+	cfg       dmesgConfig
+	reset     bool
+	pos       dmesgPos
+	dirty     bool
+	open      bool
+	read      uint64
+	lost      uint64
+	errCode   string
+	errDetail string
+}
+
+func NewDmesg(device, dataDir string, backfill time.Duration, sink func(*Entry)) *Dmesg {
+	d := &Dmesg{
+		device:     device,
+		bootIDPath: "/proc/sys/kernel/random/boot_id",
+		uptimePath: "/proc/uptime",
+		sink:       sink,
+		backfill:   backfill,
+		cfgPath:    filepath.Join(dataDir, "dmesg.json"),
+		statePath:  filepath.Join(dataDir, "dmesg-state.json"),
+		retry:      10 * time.Second,
+		wake:       make(chan struct{}, 1),
+	}
+	if b, err := os.ReadFile(d.cfgPath); err == nil {
+		_ = json.Unmarshal(b, &d.cfg)
+	}
+	if b, err := os.ReadFile(d.statePath); err == nil {
+		_ = json.Unmarshal(b, &d.pos)
+	}
+	return d
+}
+
+// kmsgSession is one opening of the device, read by its own goroutine.
+type kmsgSession struct {
+	f       *os.File
+	done    chan struct{}
+	err     error
+	stopped bool // closed on purpose: err is not a failure
+}
+
+func (s *kmsgSession) stop() {
+	s.stopped = true
+	s.f.Close() // ends a pending read (/dev/kmsg goes through the poller)
+	select {
+	case <-s.done:
+	case <-time.After(2 * time.Second):
+	}
+}
+
+// Run keeps the device open while the source is enabled.
+func (d *Dmesg) Run(ctx context.Context) {
+	save := time.NewTicker(5 * time.Second)
+	defer save.Stop()
+	var cur *kmsgSession
+	var retryAt time.Time
+	for {
+		d.mu.Lock()
+		enabled, reset := d.cfg.Enabled, d.reset
+		d.reset = false
+		d.mu.Unlock()
+		if reset {
+			// Disabled then enabled again: start over from DMESG_BACKFILL ago
+			// rather than reading everything logged in between.
+			if cur != nil {
+				cur.stop()
+				cur = nil
+			}
+			d.mu.Lock()
+			d.pos, d.dirty = dmesgPos{}, true
+			d.mu.Unlock()
+			d.saveState()
+			retryAt = time.Time{}
+		}
+		if !enabled && cur != nil {
+			cur.stop()
+			cur = nil
+			d.mu.Lock()
+			d.open = false
+			d.mu.Unlock()
+		}
+		if enabled && cur == nil && !time.Now().Before(retryAt) {
+			var err error
+			if cur, err = d.start(); err != nil {
+				d.fail(err)
+				retryAt = time.Now().Add(d.retry)
+			}
+		}
+		var done chan struct{}
+		if cur != nil {
+			done = cur.done
+		}
+		wait := d.retry
+		if cur == nil && enabled {
+			wait = max(time.Until(retryAt), 10*time.Millisecond)
+		}
+		timer := time.NewTimer(wait)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			if cur != nil {
+				cur.stop()
+			}
+			d.saveState()
+			return
+		case <-d.wake:
+		case <-save.C:
+			d.saveState()
+		case <-timer.C:
+		case <-done:
+			if !cur.stopped {
+				d.fail(cur.err)
+			}
+			cur = nil
+			retryAt = time.Now().Add(d.retry)
+		}
+		timer.Stop()
+	}
+}
+
+func (d *Dmesg) saveState() {
+	d.mu.Lock()
+	if !d.dirty {
+		d.mu.Unlock()
+		return
+	}
+	d.dirty = false
+	pos := d.pos
+	d.mu.Unlock()
+	if err := writeJSONFile(d.statePath, pos); err != nil {
+		logEvent("dmesg", 4, "saving position: %v", err)
+	}
+}
+
+// fail records why the device is not being read (nil error: end of a
+// regular file, only seen in tests).
+func (d *Dmesg) fail(err error) {
+	code, detail := "", ""
+	if err != nil && !errors.Is(err, io.EOF) {
+		detail = err.Error()
+		switch {
+		case errors.Is(err, fs.ErrNotExist):
+			code = "not_found"
+		case errors.Is(err, fs.ErrPermission):
+			code = "permission"
+		default:
+			code = "read"
+		}
+	}
+	d.mu.Lock()
+	changed := code != d.errCode
+	d.open, d.errCode, d.errDetail = false, code, detail
+	d.mu.Unlock()
+	if changed && code != "" {
+		logEvent("dmesg", 4, "cannot read the kernel messages: %v", err)
+	}
+}
+
+// start opens the device and starts reading it in the background.
+func (d *Dmesg) start() (*kmsgSession, error) {
+	f, err := os.Open(d.device)
+	if err != nil {
+		return nil, err
+	}
+	bootID := readTrim(d.bootIDPath)
+	boot := d.bootTime(time.Now())
+	d.mu.Lock()
+	pos := d.pos
+	d.open, d.errCode, d.errDetail = true, "", ""
+	d.mu.Unlock()
+
+	// Same boot as the saved position: skip what was already sent. Otherwise
+	// read from DMESG_BACKFILL ago (0: only new records).
+	resume := pos.BootID != "" && pos.BootID == bootID
+	var notBefore time.Time
+	if !resume {
+		if d.backfill <= 0 {
+			if _, err := f.Seek(0, io.SeekEnd); err != nil {
+				f.Close()
+				return nil, err
+			}
+		} else if !boot.IsZero() {
+			notBefore = time.Now().Add(-d.backfill)
+		}
+		d.mu.Lock()
+		d.pos, d.dirty = dmesgPos{BootID: bootID}, true
+		d.mu.Unlock()
+	}
+	s := &kmsgSession{f: f, done: make(chan struct{})}
+	go func() {
+		defer close(s.done)
+		s.err = d.consume(f, bootID, boot, resume, pos.Seq, notBefore)
+	}()
+	return s, nil
+}
+
+func readTrim(path string) string {
+	b, err := os.ReadFile(path)
+	if err != nil {
+		return ""
+	}
+	return strings.TrimSpace(string(b))
+}
+
+// bootTime is now minus the uptime; zero when unknown (records are then dated
+// when they are read).
+func (d *Dmesg) bootTime(now time.Time) time.Time {
+	f := strings.Fields(readTrim(d.uptimePath))
+	if len(f) == 0 {
+		return time.Time{}
+	}
+	up, err := strconv.ParseFloat(f[0], 64)
+	if err != nil || up <= 0 {
+		return time.Time{}
+	}
+	return now.Add(-time.Duration(up * float64(time.Second)))
+}
+
+// consume reads records until an error. Each read of /dev/kmsg returns one
+// record; several records in one read (a regular file, in tests) are split.
+func (d *Dmesg) consume(r io.Reader, bootID string, boot time.Time, resume bool, after uint64, notBefore time.Time) error {
+	buf := make([]byte, 64<<10) // a record is at most 8 KiB
+	last, haveLast := after, resume
+	for {
+		n, err := r.Read(buf)
+		if n > 0 {
+			for _, raw := range splitKmsg(buf[:n]) {
+				rec, ok := parseKmsg(raw)
+				if !ok {
+					continue
+				}
+				if haveLast && rec.Seq <= last {
+					continue // already sent before a restart
+				}
+				var lost uint64
+				if haveLast && rec.Seq > last+1 {
+					lost = rec.Seq - last - 1 // overwritten in the ring buffer before being read
+				}
+				last, haveLast = rec.Seq, true
+				now := time.Now()
+				t := now
+				if !boot.IsZero() {
+					t = boot.Add(time.Duration(rec.Usec) * time.Microsecond)
+				}
+				if !notBefore.IsZero() && t.Before(notBefore) {
+					d.advance(bootID, rec.Seq, 0, 0)
+					continue
+				}
+				d.sink(kmsgEntry(rec, t, now))
+				d.advance(bootID, rec.Seq, 1, lost)
+			}
+		}
+		if err != nil {
+			if errors.Is(err, os.ErrClosed) {
+				return nil
+			}
+			if errors.Is(err, syscall.EPIPE) {
+				continue // records overwritten while reading: the next read goes on
+			}
+			return err
+		}
+	}
+}
+
+func (d *Dmesg) advance(bootID string, seq, read, lost uint64) {
+	d.mu.Lock()
+	d.pos, d.dirty = dmesgPos{BootID: bootID, Seq: seq}, true
+	d.read += read
+	d.lost += lost
+	d.mu.Unlock()
+}
+
+// kmsgRecord is one record of /dev/kmsg:
+//
+//	prio,seq,usec,flags[,...];message
+//	 KEY=value          (optional dictionary lines)
+type kmsgRecord struct {
+	Prio int
+	Seq  uint64
+	Usec uint64
+	Msg  string
+	Dict map[string]string
+}
+
+// splitKmsg cuts a buffer into records: a new record starts after a newline
+// that is not followed by a space (dictionary lines start with one).
+func splitKmsg(b []byte) [][]byte {
+	var out [][]byte
+	start := 0
+	for i := 0; i < len(b); i++ {
+		if b[i] == '\n' && (i+1 == len(b) || b[i+1] != ' ') {
+			out = append(out, b[start:i])
+			start = i + 1
+		}
+	}
+	if start < len(b) {
+		out = append(out, b[start:])
+	}
+	return out
+}
+
+func parseKmsg(raw []byte) (kmsgRecord, bool) {
+	var rec kmsgRecord
+	lines := bytes.Split(raw, []byte{'\n'})
+	head, msg, ok := bytes.Cut(lines[0], []byte{';'})
+	if !ok {
+		return rec, false
+	}
+	f := strings.Split(string(head), ",")
+	if len(f) < 3 {
+		return rec, false
+	}
+	prio, err1 := strconv.Atoi(f[0])
+	seq, err2 := strconv.ParseUint(f[1], 10, 64)
+	usec, err3 := strconv.ParseUint(f[2], 10, 64)
+	if err1 != nil || err2 != nil || err3 != nil || prio < 0 {
+		return rec, false
+	}
+	rec.Prio, rec.Seq, rec.Usec = prio, seq, usec
+	rec.Msg = unescapeKmsg(string(msg))
+	for _, l := range lines[1:] {
+		if k, v, ok := strings.Cut(strings.TrimPrefix(string(l), " "), "="); ok && k != "" {
+			if rec.Dict == nil {
+				rec.Dict = map[string]string{}
+			}
+			rec.Dict[k] = unescapeKmsg(v)
+		}
+	}
+	return rec, true
+}
+
+// unescapeKmsg decodes the \xNN escapes the kernel uses for non-printable bytes.
+func unescapeKmsg(s string) string {
+	if !strings.Contains(s, `\x`) {
+		return s
+	}
+	var b strings.Builder
+	for i := 0; i < len(s); i++ {
+		if s[i] == '\\' && i+3 < len(s) && s[i+1] == 'x' {
+			if v, err := strconv.ParseUint(s[i+2:i+4], 16, 8); err == nil {
+				b.WriteByte(byte(v))
+				i += 3
+				continue
+			}
+		}
+		b.WriteByte(s[i])
+	}
+	return b.String()
+}
+
+// kmsgEntry turns a record into an Entry. Kernel records have the kern
+// facility; programs writing to /dev/kmsg (systemd at boot…) use another
+// one and prefix their name: "systemd[1]: …".
+func kmsgEntry(rec kmsgRecord, t, now time.Time) *Entry {
+	sev := rec.Prio & 7
+	fac := rec.Prio >> 3
+	if fac >= len(facilityNames) {
+		fac = 1 // user
+	}
+	msg := strings.TrimRight(rec.Msg, "\n")
+	if !utf8.ValidString(msg) {
+		msg = strings.ToValidUTF8(msg, "�")
+	}
+	app, pid := "kernel", ""
+	if fac != 0 {
+		if a, p, rest, ok := splitTag(msg); ok {
+			app, pid, msg = a, p, rest
+		}
+	}
+	e := &Entry{
+		Time:       t,
+		Received:   now,
+		Host:       "localhost",
+		App:        app,
+		ProcID:     pid,
+		Facility:   facilityNames[fac],
+		Severity:   severityNames[sev],
+		SevNum:     sev,
+		Message:    msg,
+		Proto:      "kmsg",
+		SourceType: "kernel",
+		Wait:       true,
+	}
+	if s, dev := rec.Dict["SUBSYSTEM"], rec.Dict["DEVICE"]; s != "" || dev != "" {
+		e.Extra = map[string]string{}
+		if s != "" {
+			e.Extra["subsystem"] = s
+		}
+		if dev != "" {
+			e.Extra["device"] = dev
+		}
+	}
+	return e
+}
+
+// splitTag cuts "name[pid]: message" or "name: message".
+func splitTag(msg string) (app, pid, rest string, ok bool) {
+	tag, rest, found := strings.Cut(msg, ": ")
+	if !found || tag == "" || len(tag) > 48 || strings.ContainsAny(tag, " \t") {
+		return "", "", "", false
+	}
+	if i := strings.IndexByte(tag, '['); i > 0 && strings.HasSuffix(tag, "]") {
+		return tag[:i], tag[i+1 : len(tag)-1], rest, true
+	}
+	return tag, "", rest, true
+}
+
+// Configure saves and applies the configuration.
+func (d *Dmesg) Configure(cfg dmesgConfig) error {
+	d.mu.Lock()
+	if d.cfg.Enabled && !cfg.Enabled {
+		d.reset = true
+		d.errCode, d.errDetail = "", ""
+	}
+	d.cfg = cfg
+	d.mu.Unlock()
+	select {
+	case d.wake <- struct{}{}:
+	default:
+	}
+	return writeJSONFile(d.cfgPath, cfg)
+}
+
+// Status is the state shown in Settings > Sources.
+func (d *Dmesg) Status() map[string]any {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	return map[string]any{
+		"enabled": d.cfg.Enabled,
+		"open":    d.open && d.cfg.Enabled,
+		"device":  d.device,
+		"read":    d.read,
+		"lost":    d.lost,
+		"code":    d.errCode,
+		"error":   d.errDetail,
+	}
+}

dmesg_test.go

@@ -0,0 +1,265 @@
+package main
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestParseKmsg(t *testing.T) {
+	rec, ok := parseKmsg([]byte("6,339,5140900,-;NET: Registered protocol family 10\n SUBSYSTEM=net\n DEVICE=n2"))
+	if !ok || rec.Prio != 6 || rec.Seq != 339 || rec.Usec != 5140900 || rec.Msg != "NET: Registered protocol family 10" {
+		t.Fatalf("record %+v", rec)
+	}
+	if rec.Dict["SUBSYSTEM"] != "net" || rec.Dict["DEVICE"] != "n2" {
+		t.Fatalf("dict %+v", rec.Dict)
+	}
+	// Extra header fields (newer kernels: caller id) and a ';' in the message.
+	rec, ok = parseKmsg([]byte("4,12,99,c,caller=T1;a;b \\x1b[31mred\\x5c"))
+	if !ok || rec.Seq != 12 || rec.Msg != "a;b \x1b[31mred\\" {
+		t.Fatalf("record %+v", rec)
+	}
+	for _, bad := range []string{"", "no header", "6,1;short", "x,1,2,-;msg", "6,-1,2,-;msg", "-1,1,2,-;msg"} {
+		if _, ok := parseKmsg([]byte(bad)); ok {
+			t.Errorf("%q parsed", bad)
+		}
+	}
+}
+
+func TestUnescapeKmsg(t *testing.T) {
+	for in, want := range map[string]string{
+		`plain`:       "plain",
+		`a\x0ab`:      "a\nb",
+		`end\x4`:      `end\x4`,
+		`bad\xzz`:     `bad\xzz`,
+		`\x41\x42\x4`: `AB\x4`,
+	} {
+		if got := unescapeKmsg(in); got != want {
+			t.Errorf("%q: got %q, want %q", in, got, want)
+		}
+	}
+}
+
+func TestSplitKmsg(t *testing.T) {
+	got := splitKmsg([]byte("6,1,0,-;one\n SUBSYSTEM=pci\n6,2,0,-;two\n6,3,0,-;three"))
+	if len(got) != 3 || string(got[0]) != "6,1,0,-;one\n SUBSYSTEM=pci" || string(got[2]) != "6,3,0,-;three" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestKmsgEntry(t *testing.T) {
+	boot := time.Date(2026, 10, 4, 6, 0, 0, 0, time.UTC)
+	now := boot.Add(time.Hour)
+	rec, _ := parseKmsg([]byte("3,7,2500000,-;usb 1-1: device descriptor read/64, error -71\n SUBSYSTEM=usb\n DEVICE=c189:1"))
+	e := kmsgEntry(rec, boot.Add(2500*time.Millisecond), now)
+	if e.App != "kernel" || e.Facility != "kern" || e.Severity != "err" || e.SevNum != 3 || e.SourceType != "kernel" || e.Proto != "kmsg" || !e.Wait {
+		t.Fatalf("entry %+v", e)
+	}
+	if e.Message != "usb 1-1: device descriptor read/64, error -71" || e.Extra["subsystem"] != "usb" || e.Extra["device"] != "c189:1" {
+		t.Fatalf("entry %+v", e)
+	}
+	if !e.Time.Equal(boot.Add(2500*time.Millisecond)) || !e.Received.Equal(now) {
+		t.Fatalf("times %v %v", e.Time, e.Received)
+	}
+	// Written by a program (facility daemon = 3<<3 | info).
+	rec, _ = parseKmsg([]byte("30,8,3000000,-;systemd[1]: Started Journal Service."))
+	e = kmsgEntry(rec, now, now)
+	if e.App != "systemd" || e.ProcID != "1" || e.Facility != "daemon" || e.Severity != "info" || e.Message != "Started Journal Service." || e.Extra != nil {
+		t.Fatalf("entry %+v", e)
+	}
+	// Facility out of range.
+	rec, _ = parseKmsg([]byte("255,9,0,-;odd"))
+	if e = kmsgEntry(rec, now, now); e.Facility != "user" || e.Severity != "debug" {
+		t.Fatalf("entry %+v", e)
+	}
+}
+
+// fakeHost writes a kmsg file, a boot id and an uptime in a temp dir.
+func fakeHost(t *testing.T, uptime float64, records ...string) (*Dmesg, *[]*Entry, *sync.Mutex) {
+	t.Helper()
+	dir := t.TempDir()
+	write := func(name, s string) string {
+		p := filepath.Join(dir, name)
+		if err := os.WriteFile(p, []byte(s), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		return p
+	}
+	var mu sync.Mutex
+	var got []*Entry
+	d := NewDmesg(write("kmsg", strings.Join(records, "\n")+"\n"), filepath.Join(dir, "data"), time.Hour, func(e *Entry) {
+		mu.Lock()
+		got = append(got, e)
+		mu.Unlock()
+	})
+	d.bootIDPath = write("boot_id", "b1\n")
+	d.uptimePath = write("uptime", strconv.FormatFloat(uptime, 'f', 2, 64)+" 1234.00\n")
+	d.retry = 20 * time.Millisecond
+	return d, &got, &mu
+}
+
+func TestDmesgBackfillAndResume(t *testing.T) {
+	// Booted 3 h ago: the record at 1 s is outside the 1 h backfill window,
+	// the ones at 2.5 h and 2.9 h are inside.
+	d, got, mu := fakeHost(t, 3*3600,
+		"6,1,1000000,-;Linux version 6.12",
+		"4,2,9000000000,-;eth0: link down",
+		"6,3,10440000000,-;eth0: link up",
+	)
+	s, err := d.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-s.done
+	mu.Lock()
+	if len(*got) != 2 || (*got)[0].Message != "eth0: link down" || (*got)[1].Message != "eth0: link up" {
+		t.Fatalf("got %+v", *got)
+	}
+	if age := time.Since((*got)[1].Time); age < 5*time.Minute || age > 7*time.Minute {
+		t.Fatalf("record dated %v ago", age)
+	}
+	mu.Unlock()
+	d.saveState()
+
+	// Restart with the saved position: same boot, nothing is sent again,
+	// and a new record is.
+	f, _ := os.OpenFile(d.device, os.O_APPEND|os.O_WRONLY, 0)
+	_, _ = f.WriteString("6,6,10500000000,-;eth0: renamed\n")
+	f.Close()
+	d2 := NewDmesg(d.device, filepath.Dir(d.statePath), time.Hour, d.sink)
+	d2.bootIDPath, d2.uptimePath = d.bootIDPath, d.uptimePath
+	if d2.pos != (dmesgPos{BootID: "b1", Seq: 3}) {
+		t.Fatalf("saved position %+v", d2.pos)
+	}
+	s, err = d2.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-s.done
+	mu.Lock()
+	defer mu.Unlock()
+	if len(*got) != 3 || (*got)[2].Message != "eth0: renamed" || (*got)[2].App != "kernel" {
+		t.Fatalf("after restart %+v", (*got)[2:])
+	}
+	if st := d2.Status(); st["lost"] != uint64(2) || st["read"] != uint64(1) {
+		t.Fatalf("status %+v", st) // records 4 and 5 were overwritten
+	}
+}
+
+func TestDmesgNewBoot(t *testing.T) {
+	d, got, mu := fakeHost(t, 60, "6,1,1000000,-;booted")
+	d.pos = dmesgPos{BootID: "old", Seq: 500}
+	s, err := d.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-s.done
+	mu.Lock()
+	defer mu.Unlock()
+	if len(*got) != 1 || d.pos != (dmesgPos{BootID: "b1", Seq: 1}) {
+		t.Fatalf("got %d entries, position %+v", len(*got), d.pos)
+	}
+}
+
+func TestDmesgNoBackfill(t *testing.T) {
+	d, got, mu := fakeHost(t, 60, "6,1,1000000,-;old")
+	d.backfill = 0
+	s, err := d.start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-s.done
+	mu.Lock()
+	defer mu.Unlock()
+	if len(*got) != 0 {
+		t.Fatalf("got %+v", *got)
+	}
+}
+
+func TestDmesgRun(t *testing.T) {
+	d, got, mu := fakeHost(t, 60, "6,1,1000000,-;first", "6,2,2000000,-;second")
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		d.Run(ctx)
+		close(done)
+	}()
+	if err := d.Configure(dmesgConfig{Enabled: true}); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		mu.Lock()
+		n := len(*got)
+		mu.Unlock()
+		if n >= 2 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("got %d entries", n)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	// The file is re-opened after each end of file: still no duplicate.
+	time.Sleep(100 * time.Millisecond)
+	cancel()
+	<-done
+	mu.Lock()
+	defer mu.Unlock()
+	if len(*got) != 2 {
+		t.Fatalf("got %d entries", len(*got))
+	}
+	b, err := os.ReadFile(d.statePath)
+	if err != nil || !strings.Contains(string(b), `"seq": 2`) {
+		t.Fatalf("state %s %v", b, err)
+	}
+	b, _ = os.ReadFile(d.cfgPath)
+	if !strings.Contains(string(b), `"enabled": true`) {
+		t.Fatalf("config %s", b)
+	}
+}
+
+func TestDmesgMissingDevice(t *testing.T) {
+	d := NewDmesg(filepath.Join(t.TempDir(), "kmsg"), t.TempDir(), time.Hour, func(*Entry) {})
+	d.retry = 20 * time.Millisecond
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		d.Run(ctx)
+		close(done)
+	}()
+	_ = d.Configure(dmesgConfig{Enabled: true})
+	deadline := time.Now().Add(5 * time.Second)
+	for d.Status()["code"] != "not_found" {
+		if time.Now().After(deadline) {
+			t.Fatalf("status %+v", d.Status())
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	// Turning it off clears the error.
+	_ = d.Configure(dmesgConfig{Enabled: false})
+	if st := d.Status(); st["code"] != "" || st["enabled"] != false {
+		t.Fatalf("status %+v", st)
+	}
+	cancel()
+	<-done
+}
+
+func TestDmesgPermission(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root reads everything")
+	}
+	p := filepath.Join(t.TempDir(), "kmsg")
+	_ = os.WriteFile(p, []byte("6,1,0,-;x\n"), 0o000)
+	d := NewDmesg(p, t.TempDir(), time.Hour, func(*Entry) {})
+	_, err := d.start()
+	d.fail(err)
+	if st := d.Status(); st["code"] != "permission" {
+		t.Fatalf("status %+v", st)
+	}
+}

docker-compose.yml

@@ -7,9 +7,9 @@ services:
       - victorialogs
       - docker-proxy
     ports:
-      - "${SYSLOG_PORT:-514}:5514/udp"
+      - "${SYSLOG_PORT:-5514}:5514/udp"
-      - "${SYSLOG_PORT:-514}:5514/tcp"
+      - "${SYSLOG_PORT:-5514}:5514/tcp"
-      - "${HTTP_PORT:-8080}:8080"
+      - "${HTTP_PORT:-8989}:8080"
     environment:
       VLOGS_URL: http://victorialogs:9428
       SYSLOG_PUBLIC_PORT: ${SYSLOG_PORT:-514}     # le port d'ecoute syslog par defaut (attention aux ports <1024)
@@ -37,13 +37,35 @@ services:
       DOCKER_HOST: tcp://docker-proxy:2375        # lecture seul de l'API Docker
       DOCKER_BACKFILL: ${DOCKER_BACKFILL:-1h}
       HOST_LOGS_BACKFILL: ${HOST_LOGS_BACKFILL:-1h}   # historique lu a l'activation des logs systeme de l'hote
+      DMESG_BACKFILL: ${DMESG_BACKFILL:-1h}   # historique lu a l'activation des messages du noyau (dmesg)
+      AUDIT_BACKFILL: ${AUDIT_BACKFILL:-1h}   # historique lu a l'activation de l'audit Linux (auditd)
+      FILES_BACKFILL: ${FILES_BACKFILL:-1h}   # historique lu a l'activation d'un fichier de logs (Reglages > Sources)
+      AUDIT_HOST: ${AUDIT_HOST:-}             # nom d'hote des evenements d'audit (vide : node= des enregistrements, sinon /host/etc/hostname)
     group_add:
       - "${HOST_LOGS_GID:-4}"        # groupe autorise a lire les logs de l'hote (4 = adm sur Debian/Ubuntu)
+    # messages du noyau de l'hote (dmesg), source a activer dans Reglages > Sources :
+    # decommenter devices pour donner /dev/kmsg (lecture seule) au conteneur
+    # devices:
+    #   - /dev/kmsg:/dev/kmsg:r
+    # si l'hote a kernel.dmesg_restrict=1 (defaut Debian/Ubuntu), soit sur l'hote :
+    #   sysctl -w kernel.dmesg_restrict=0   (et kernel.dmesg_restrict = 0 dans /etc/sysctl.d/ pour le garder)
+    # soit lancer le conteneur en root avec la capacite SYSLOG (decommenter les deux) :
+    # user: "0:0"
+    # cap_add:
+    #   - SYSLOG
     volumes:
       - logstream-data:/data         # tags.json, docker.json (les choix des conteneurs)
       # logs systeme de l'hote, en lecture seule (source a activer dans Reglages > Sources)
       - /var/log:/host/var/log:ro                     # journal systemd persistant et fichiers texte
       - /run/log/journal:/host/run/log/journal:ro     # journal systemd volatile
+      # audit Linux (auditd), source a activer dans Reglages > Sources ; le fichier doit etre
+      # lisible par l'UID 10001 : log_group = adm dans /etc/audit/auditd.conf (voir le README)
+      # - /var/log/audit:/var/log/audit:ro
+      # noms des utilisateurs et de l'hote pour les evenements d'audit (optionnel)
+      # - /etc/passwd:/host/etc/passwd:ro
+      # - /etc/hostname:/host/etc/hostname:ro
+      # fichiers de logs d'autres programmes (Traefik, nginx...), a declarer dans Reglages > Sources > Fichiers de logs
+      # - /var/log/traefik:/logs/traefik:ro
       # logo de la page de connexion, avec LOGIN_LOGO=/config/logo.png dans .env
       # - ./logo.png:/config/logo.png:ro
       # prereglages de tags personnalises, avec PRESETS_FILE=/config/presets.json dans .env

docker.go

@@ -242,7 +242,7 @@ func (m *DockerManager) setError(err error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.lastErr != err.Error() {
-		log.Printf("docker: %v", err)
+		logEvent("docker", 3, "Docker API unreachable: %v", err)
 	}
 	m.connected = false
 	m.lastErr = err.Error()
@@ -290,7 +290,7 @@ func (m *DockerManager) sync(ctx context.Context) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if !m.connected {
-		log.Printf("docker: connected, %d containers", len(list))
+		logEvent("docker", 6, "connected to the Docker API, %d containers", len(list))
 	}
 	m.connected, m.lastErr = true, ""
 	seen := map[string]bool{}
@@ -371,7 +371,7 @@ func (m *DockerManager) follow(ctx context.Context, c DockerContainer) {
 			if ctx.Err() != nil {
 				return
 			}
-			log.Printf("docker %s: %v", c.Name, err)
+			logEvent("docker", 4, "container %s: cannot inspect it: %v", c.Name, err)
 			sleepCtx(ctx, 5*time.Second)
 			continue
 		}
@@ -383,7 +383,7 @@ func (m *DockerManager) follow(ctx context.Context, c DockerContainer) {
 			return
 		}
 		if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
-			log.Printf("docker %s: %v", c.Name, err)
+			logEvent("docker", 4, "container %s: log stream lost, reconnecting: %v", c.Name, err)
 		}
 		sleepCtx(ctx, time.Second)
 	}

docs/presets.fr.md

@@ -2,7 +2,7 @@
 
 # Préréglages de tags de couleur
 
-Dans **Paramètres › Filtres**, le menu **+ Préréglage…** ajoute d'un clic un groupe de tags de
+Dans **Paramètres › Motifs de recherche**, le menu **+ Préréglage…** ajoute d'un clic un groupe de tags de
 couleur tout faits. Les tags ajoutés sont des tags ordinaires : vous pouvez changer leur couleur,
 leur motif ou leurs options, ou les supprimer. Un tag dont le motif est déjà dans la liste n'est
 pas ajouté en double.

docs/presets.md

@@ -2,7 +2,7 @@
 
 # Color tag presets
 
-In **Settings › Filters**, the **+ Preset…** menu adds a group of ready-made color tags in one
+In **Settings › Patterns**, the **+ Preset…** menu adds a group of ready-made color tags in one
 click. Added tags are ordinary tags: you can change their color, pattern or options, or delete
 them. A tag whose pattern is already in the list is not added twice.
 

fileformats.go

@@ -0,0 +1,689 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+	"unicode/utf8"
+)
+
+// 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…)
+	fmtRegex   = "regex"   // regular expression with named groups
+	fmtCLF     = "clf"     // access logs in Common/Combined Log Format (nginx, Apache, Traefik)
+	fmtTraefik = "traefik" // Traefik access log in JSON
+)
+
+var fileFormats = map[string]bool{fmtText: true, fmtJSON: true, fmtLogfmt: true, fmtRegex: true, fmtCLF: true, fmtTraefik: true}
+
+// fileRec is one parsed line (or several, for continuation lines).
+type fileRec struct {
+	t      time.Time
+	dated  bool // t comes from the line
+	sev    int
+	msg    string
+	host   string
+	app    string
+	fields map[string]string
+}
+
+const (
+	maxFileFields   = 64
+	maxFieldValue   = 4096
+	maxFileMsg      = 64 << 10 // message with its continuation lines
+	maxContinuation = 500      // continuation lines added to one entry
+)
+
+// fileParser turns a line of a file into an entry, for one format.
+type fileParser struct {
+	format string
+	re     *regexp.Regexp // fmtRegex
+}
+
+func newFileParser(format, expr string) (*fileParser, error) {
+	if !fileFormats[format] {
+		return nil, fmt.Errorf("unknown format %q", format)
+	}
+	p := &fileParser{format: format}
+	if format == fmtRegex {
+		if strings.TrimSpace(expr) == "" {
+			return nil, fmt.Errorf("the regex format needs a regular expression")
+		}
+		re, err := regexp.Compile(expr)
+		if err != nil {
+			return nil, fmt.Errorf("invalid regular expression: %v", err)
+		}
+		p.re = re
+	}
+	return p, nil
+}
+
+// multiline tells whether lines that do not start an entry are added to the
+// previous one (stack traces…). Structured formats are one entry per line.
+func (p *fileParser) multiline() bool {
+	return p.format == fmtText || p.format == fmtRegex
+}
+
+// continuation reports a line that belongs to the previous entry: indented,
+// or not matching the regular expression.
+func (p *fileParser) continuation(line string) bool {
+	if line != "" && (line[0] == ' ' || line[0] == '\t') {
+		return true
+	}
+	return p.format == fmtRegex && !p.re.MatchString(line)
+}
+
+// parse decodes one line; it never fails (a line in another format is kept as text).
+func (p *fileParser) parse(line string, now time.Time) *fileRec {
+	line = cleanLine(line)
+	switch p.format {
+	case fmtJSON, fmtTraefik:
+		if obj, ok := decodeJSONLine(line); ok {
+			if p.format == fmtTraefik {
+				if r := traefikRec(obj, line, now); r != nil {
+					return r
+				}
+			}
+			return jsonRec(obj, line, now)
+		}
+	case fmtLogfmt:
+		if kv := parseLogfmt(line); len(kv) > 0 {
+			return logfmtRec(kv, line, now)
+		}
+	case fmtRegex:
+		if m := p.re.FindStringSubmatch(line); m != nil {
+			return p.regexRec(m, line, now)
+		}
+	case fmtCLF:
+		if r := clfRec(line, now); r != nil {
+			return r
+		}
+	}
+	return textRec(line, now)
+}
+
+func cleanLine(s string) string {
+	s = strings.TrimRight(s, "\r")
+	s = ansiRe.ReplaceAllString(s, "")
+	if !utf8.ValidString(s) {
+		s = strings.ToValidUTF8(s, "�")
+	}
+	return s
+}
+
+// textRec: the line itself, without its leading timestamp when there is one.
+func textRec(line string, now time.Time) *fileRec {
+	r := &fileRec{t: now, sev: 6, msg: line}
+	if m := leadTimeRe.FindStringSubmatchIndex(line); m != nil {
+		if t, ok := parseLogTime(line[m[2]:m[3]], now); ok {
+			r.t, r.dated = t, true
+			if rest := line[m[1]:]; strings.TrimSpace(rest) != "" {
+				r.msg = rest
+			}
+		}
+	}
+	if n, ok := detectLevel(r.msg); ok {
+		r.sev = n
+	}
+	return r
+}
+
+/* ---------- Timestamps ---------- */
+
+// leadTimeRe: a timestamp at the start of a line, possibly in brackets
+// (ISO 8601 / RFC 3339, 2006/01/02 15:04:05, CLF, syslog "Jan _2 15:04:05").
+var leadTimeRe = regexp.MustCompile(`^\[?(\d{4}[-/]\d{2}[-/]\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?: ?(?:Z|[+-]\d{2}:?\d{2}))?|\d{2}/[A-Z][a-z]{2}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4}|[A-Z][a-z]{2} +\d{1,2} \d{2}:\d{2}:\d{2})\]?[ \t]*(?:[-|:][ \t]+)?`)
+
+var timeLayouts = []string{
+	time.RFC3339Nano,
+	"2006-01-02T15:04:05Z0700",
+	"2006-01-02T15:04:05",
+	"02/Jan/2006:15:04:05 -0700",
+	"Jan _2 15:04:05",
+	time.RFC1123Z,
+	time.RFC1123,
+	time.UnixDate,
+	time.ANSIC,
+}
+
+// parseLogTime reads the usual log timestamps; times without a zone are local.
+func parseLogTime(s string, now time.Time) (time.Time, bool) {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return time.Time{}, false
+	}
+	if s[0] >= '0' && s[0] <= '9' && !strings.ContainsAny(s, ":/-") {
+		if whole, _, _ := strings.Cut(s, "."); len(whole) < 9 {
+			return time.Time{}, false // not a Unix time of this century
+		}
+		return epochTime(s)
+	}
+	// 2006/01/02 15:04:05,123 +02:00 -> 2006-01-02T15:04:05.123+02:00
+	if len(s) >= 19 && s[4] == '/' && s[7] == '/' {
+		s = s[:4] + "-" + s[5:7] + "-" + s[8:]
+	}
+	if len(s) >= 19 && s[4] == '-' && s[10] == ' ' {
+		s = s[:10] + "T" + s[11:]
+	}
+	if len(s) > 19 && s[4] == '-' {
+		s = s[:19] + strings.Replace(strings.Replace(s[19:], ",", ".", 1), " ", "", 1)
+	}
+	s = strings.Join(strings.Fields(s), " ") // "Jan  5" -> "Jan 5"
+	for _, layout := range timeLayouts {
+		if layout == "Jan _2 15:04:05" {
+			layout = "Jan 2 15:04:05"
+		}
+		t, err := time.ParseInLocation(layout, s, time.Local)
+		if err != nil {
+			continue
+		}
+		if t.Year() == 0 { // syslog timestamps have no year
+			t = t.AddDate(now.Year(), 0, 0)
+			if t.After(now.Add(24 * time.Hour)) {
+				t = t.AddDate(-1, 0, 0)
+			}
+		}
+		return t, true
+	}
+	return time.Time{}, false
+}
+
+// epochTime reads a Unix time in seconds, milliseconds, microseconds or
+// nanoseconds (guessed from its size), with an optional fraction.
+func epochTime(s string) (time.Time, bool) {
+	f, err := strconv.ParseFloat(s, 64)
+	if err != nil || f <= 0 {
+		return time.Time{}, false
+	}
+	switch {
+	case f < 1e11:
+		return time.Unix(0, int64(f*1e9)), true
+	case f < 1e14:
+		return time.UnixMilli(int64(f)), true
+	case f < 1e17:
+		return time.UnixMicro(int64(f)), true
+	}
+	return time.Unix(0, int64(f)), true
+}
+
+/* ---------- Fields ---------- */
+
+// reservedFields are stored by LogStream itself: fields of the logs with the
+// same name are renamed field_<name>.
+var reservedFields = map[string]bool{
+	"_time": true, "_msg": true, "_stream": true, "_stream_id": true, "received": true, "msg_time": true,
+	"host": true, "host_name": true, "host_ip": true, "app": true, "facility": true, "severity": true,
+	"sevnum": true, "source": true, "proto": true, "procid": true, "sd": true, "source_type": true, "log_file": true,
+}
+
+func fieldName(k string) string {
+	k = strings.Map(func(r rune) rune {
+		if r == '_' || r == '.' || r == '-' || r >= '0' && r <= '9' || r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' {
+			return r
+		}
+		return '_'
+	}, k)
+	if k == "" || reservedFields[k] || k[0] == '_' {
+		k = "field_" + strings.TrimLeft(k, "_")
+	}
+	return k
+}
+
+func (r *fileRec) set(k, v string) {
+	if v == "" {
+		return
+	}
+	if r.fields == nil {
+		r.fields = map[string]string{}
+	}
+	k = fieldName(k)
+	if _, ok := r.fields[k]; !ok && len(r.fields) >= maxFileFields {
+		return
+	}
+	if len(v) > maxFieldValue {
+		v = strings.ToValidUTF8(v[:maxFieldValue], "") + "…"
+	}
+	r.fields[k] = v
+}
+
+/* ---------- JSON lines ---------- */
+
+func decodeJSONLine(line string) (map[string]any, bool) {
+	s := strings.TrimSpace(line)
+	if len(s) < 2 || s[0] != '{' {
+		return nil, false
+	}
+	dec := json.NewDecoder(strings.NewReader(s))
+	dec.UseNumber()
+	var obj map[string]any
+	if dec.Decode(&obj) != nil {
+		return nil, false
+	}
+	return obj, true
+}
+
+var (
+	jsonMsgKeys   = []string{"msg", "message", "@message", "log", "text"}
+	jsonTimeKeys  = []string{"time", "ts", "timestamp", "@timestamp", "@t", "t", "datetime", "date"}
+	jsonLevelKeys = []string{"level", "lvl", "severity", "log.level", "@level", "@l", "loglevel", "levelname"}
+)
+
+// takeString removes the first of keys found in obj and returns it as text.
+func takeString(obj map[string]any, keys []string) (string, bool) {
+	for _, k := range keys {
+		if v, ok := obj[k]; ok {
+			if s := scalarString(v); s != "" {
+				delete(obj, k)
+				return s, true
+			}
+		}
+	}
+	return "", false
+}
+
+func scalarString(v any) string {
+	switch v := v.(type) {
+	case string:
+		return v
+	case json.Number:
+		return v.String()
+	case bool:
+		return strconv.FormatBool(v)
+	}
+	return ""
+}
+
+func jsonRec(obj map[string]any, line string, now time.Time) *fileRec {
+	r := &fileRec{t: now, sev: 6, msg: line}
+	for _, k := range jsonTimeKeys {
+		if t, ok := parseLogTime(scalarString(obj[k]), now); ok {
+			r.t, r.dated = t, true
+			delete(obj, k)
+			break
+		}
+	}
+	lvl, hasLevel := takeString(obj, jsonLevelKeys)
+	if hasLevel {
+		if n, ok := levelNum(lvl); ok {
+			r.sev = n
+		} else if n, err := strconv.Atoi(lvl); err == nil {
+			r.sev = numericLevel(n)
+		} else {
+			r.set("level", lvl)
+		}
+	}
+	if s, ok := takeString(obj, jsonMsgKeys); ok {
+		r.msg = strings.TrimRight(s, "\n")
+	}
+	flattenJSON("", obj, r)
+	if !hasLevel {
+		if n, ok := detectLevel(r.msg); ok {
+			r.sev = n
+		}
+	}
+	return r
+}
+
+// numericLevel maps numeric levels: syslog (0-7) or pino/bunyan (10-60).
+func numericLevel(n int) int {
+	switch {
+	case n >= 0 && n <= 7:
+		return n
+	case n >= 60:
+		return 2
+	case n >= 50:
+		return 3
+	case n >= 40:
+		return 4
+	case n >= 30:
+		return 6
+	}
+	return 7
+}
+
+// flattenJSON stores the remaining keys as fields; nested objects give
+// dotted names, arrays stay in JSON.
+func flattenJSON(prefix string, obj map[string]any, r *fileRec) {
+	for k, v := range obj {
+		name := prefix + k
+		switch v := v.(type) {
+		case map[string]any:
+			flattenJSON(name+".", v, r)
+		case []any:
+			b, _ := json.Marshal(v)
+			r.set(name, string(b))
+		case nil:
+		default:
+			r.set(name, scalarString(v))
+		}
+	}
+}
+
+/* ---------- logfmt ---------- */
+
+// parseLogfmt splits `k=v k2="v 2"` (with \" escapes); words without a value
+// are skipped. Nil when the line has no pair.
+func parseLogfmt(s string) [][2]string {
+	var out [][2]string
+	for i := 0; i < len(s); {
+		for i < len(s) && (s[i] == ' ' || s[i] == '\t') {
+			i++
+		}
+		start := i
+		for i < len(s) && s[i] != '=' && s[i] != ' ' && s[i] != '\t' {
+			i++
+		}
+		if i >= len(s) || s[i] != '=' || i == start {
+			for i < len(s) && s[i] != ' ' && s[i] != '\t' {
+				i++
+			}
+			continue
+		}
+		k := s[start:i]
+		i++
+		var v string
+		if i < len(s) && s[i] == '"' {
+			var b strings.Builder
+			i++
+			for i < len(s) && s[i] != '"' {
+				if s[i] == '\\' && i+1 < len(s) {
+					i++
+					switch s[i] {
+					case 'n':
+						b.WriteByte('\n')
+					case 't':
+						b.WriteByte('\t')
+					default:
+						b.WriteByte(s[i])
+					}
+				} else {
+					b.WriteByte(s[i])
+				}
+				i++
+			}
+			i++ // closing quote
+			v = b.String()
+		} else {
+			end := i
+			for end < len(s) && s[end] != ' ' && s[end] != '\t' {
+				end++
+			}
+			v, i = s[i:end], end
+		}
+		out = append(out, [2]string{k, v})
+	}
+	return out
+}
+
+func logfmtRec(kv [][2]string, line string, now time.Time) *fileRec {
+	obj := make(map[string]any, len(kv))
+	for _, p := range kv {
+		obj[p[0]] = p[1]
+	}
+	return jsonRec(obj, line, now)
+}
+
+/* ---------- Regular expression ---------- */
+
+func (p *fileParser) regexRec(m []string, line string, now time.Time) *fileRec {
+	r := &fileRec{t: now, sev: 6, msg: line}
+	hasLevel := false
+	for i, name := range p.re.SubexpNames() {
+		if i == 0 || name == "" || m[i] == "" {
+			continue
+		}
+		v := m[i]
+		switch strings.ToLower(name) {
+		case "time", "timestamp", "ts":
+			if t, ok := parseLogTime(v, now); ok {
+				r.t, r.dated = t, true
+			}
+		case "level", "severity", "lvl":
+			if n, ok := levelNum(v); ok {
+				r.sev, hasLevel = n, true
+			}
+		case "msg", "message":
+			r.msg = v
+		case "host":
+			r.host = v
+		case "app":
+			r.app = v
+		case "status":
+			r.set(name, v)
+			if n, ok := statusLevel(v); ok && !hasLevel {
+				r.sev, hasLevel = n, true
+			}
+		default:
+			r.set(name, v)
+		}
+	}
+	if !hasLevel {
+		if n, ok := detectLevel(r.msg); ok {
+			r.sev = n
+		}
+	}
+	return r
+}
+
+/* ---------- Access logs ---------- */
+
+// statusLevel: 5xx are errors, 4xx warnings, the rest info.
+func statusLevel(s string) (int, bool) {
+	n, err := strconv.Atoi(s)
+	if err != nil || n < 100 || n > 999 {
+		return 0, false
+	}
+	switch {
+	case n >= 500:
+		return 3, true
+	case n >= 400:
+		return 4, true
+	}
+	return 6, true
+}
+
+// clfRe: Common Log Format, the combined extension (referer, user agent) and
+// the Traefik one (request count, router, server URL, duration).
+var clfRe = regexp.MustCompile(`^(\S+) (\S+) (\S+) \[([^\]]+)\] "((?:[^"\\]|\\.)*)" (\d{3}) (\d+|-)(?: "((?:[^"\\]|\\.)*)" "((?:[^"\\]|\\.)*)")?(?: (\d+) "([^"]*)" "([^"]*)" (\d+)ms)?`)
+
+func clfRec(line string, now time.Time) *fileRec {
+	m := clfRe.FindStringSubmatch(line)
+	if m == nil {
+		return nil
+	}
+	r := &fileRec{t: now, sev: 6, msg: line}
+	if t, ok := parseLogTime(m[4], now); ok {
+		r.t, r.dated = t, true
+	}
+	dash := func(s string) string {
+		if s == "-" {
+			return ""
+		}
+		return s
+	}
+	r.set("client_ip", m[1])
+	r.set("http_user", dash(m[3]))
+	if req := strings.Fields(m[5]); len(req) == 3 {
+		r.set("http_method", req[0])
+		r.set("http_path", req[1])
+		r.set("http_proto", req[2])
+	} else {
+		r.set("http_request", m[5])
+	}
+	r.set("http_status", m[6])
+	r.set("http_size", dash(m[7]))
+	r.set("http_referer", dash(m[8]))
+	r.set("user_agent", dash(m[9]))
+	r.set("router", dash(m[11]))
+	r.set("backend", dash(m[12]))
+	r.set("duration_ms", m[13])
+	r.sev, _ = statusLevel(m[6])
+	return r
+}
+
+// traefikRec reads a Traefik access log line in JSON; nil for the other
+// lines (Traefik's own logs), read as generic JSON.
+func traefikRec(obj map[string]any, line string, now time.Time) *fileRec {
+	status := scalarString(obj["DownstreamStatus"])
+	if status == "" {
+		status = scalarString(obj["OriginStatus"])
+	}
+	method := scalarString(obj["RequestMethod"])
+	if status == "" && method == "" {
+		return nil
+	}
+	r := &fileRec{t: now, sev: 6}
+	if t, ok := parseLogTime(firstNonEmpty(scalarString(obj["StartUTC"]), scalarString(obj["time"])), now); ok {
+		r.t, r.dated = t, true
+	}
+	if n, ok := statusLevel(status); ok {
+		r.sev = n
+	}
+	rename := map[string]string{
+		"ClientHost": "client_ip", "ClientUsername": "http_user", "RequestMethod": "http_method",
+		"RequestHost": "http_host", "RequestPath": "http_path", "RequestProtocol": "http_proto",
+		"DownstreamStatus": "http_status", "DownstreamContentSize": "http_size", "OriginStatus": "origin_status",
+		"RouterName": "router", "ServiceName": "service", "ServiceURL": "backend", "entryPointName": "entrypoint",
+		"request_User-Agent": "user_agent", "request_Referer": "http_referer",
+	}
+	dur := ""
+	if d, err := strconv.ParseFloat(scalarString(obj["Duration"]), 64); err == nil {
+		dur = strconv.FormatFloat(d/1e6, 'f', -1, 64)
+		r.set("duration_ms", dur)
+	}
+	for _, k := range []string{"Duration", "StartUTC", "StartLocal", "time", "level", "msg"} {
+		delete(obj, k)
+	}
+	for k, v := range obj {
+		if to, ok := rename[k]; ok {
+			r.set(to, scalarString(v))
+			delete(obj, k)
+		}
+	}
+	flattenJSON("", obj, r)
+	// "GET example.com/path 404 · 12.5 ms · router@docker"
+	var b strings.Builder
+	b.WriteString(firstNonEmpty(method, "-"))
+	b.WriteByte(' ')
+	b.WriteString(r.fields["http_host"])
+	b.WriteString(r.fields["http_path"])
+	b.WriteByte(' ')
+	b.WriteString(firstNonEmpty(status, "-"))
+	if dur != "" {
+		b.WriteString(" · " + dur + " ms")
+	}
+	if rt := r.fields["router"]; rt != "" {
+		b.WriteString(" · " + rt)
+	}
+	if ip := r.fields["client_ip"]; ip != "" {
+		b.WriteString(" · " + ip)
+	}
+	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
+}

filesource.go

@@ -0,0 +1,639 @@
+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
+}

filesource_test.go

@@ -0,0 +1,384 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+func mustParser(t *testing.T, format, expr string) *fileParser {
+	t.Helper()
+	p, err := newFileParser(format, expr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return p
+}
+
+func TestParseLogTime(t *testing.T) {
+	now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
+	utc := func(s string) time.Time { t, _ := time.Parse(time.RFC3339Nano, s); return t }
+	tests := []struct {
+		in   string
+		want time.Time
+	}{
+		{"2026-10-05T10:00:00.123Z", utc("2026-10-05T10:00:00.123Z")},
+		{"2026-10-05T10:00:00+02:00", utc("2026-10-05T08:00:00Z")},
+		{"2026-10-05 10:00:00,250 +0200", utc("2026-10-05T08:00:00.25Z")},
+		{"2026/10/05 10:00:00Z", utc("2026-10-05T10:00:00Z")},
+		{"05/Oct/2026:10:00:00 +0200", utc("2026-10-05T08:00:00Z")},
+		{"1759572000", time.Unix(1759572000, 0)},
+		{"1759572000123", time.UnixMilli(1759572000123)},
+	}
+	for _, tt := range tests {
+		got, ok := parseLogTime(tt.in, now)
+		if !ok || !got.Equal(tt.want) {
+			t.Errorf("%q: got %v %v, want %v", tt.in, got, ok, tt.want)
+		}
+	}
+	// Syslog timestamps: no year, the last one before tomorrow.
+	if got, ok := parseLogTime("Dec 31 23:00:00", now); !ok || got.Year() != 2025 {
+		t.Errorf("Dec 31: %v %v", got, ok)
+	}
+	if got, ok := parseLogTime("Oct  5 11:00:00", now); !ok || got.Year() != 2026 || got.Day() != 5 {
+		t.Errorf("Oct 5: %v %v", got, ok)
+	}
+	for _, bad := range []string{"", "hello", "200", "12345"} {
+		if _, ok := parseLogTime(bad, now); ok {
+			t.Errorf("%q parsed", bad)
+		}
+	}
+}
+
+func TestParseText(t *testing.T) {
+	now := time.Now()
+	p := mustParser(t, fmtText, "")
+	r := p.parse("2026-10-05 10:00:00.5 ERROR db: connection refused", now)
+	if !r.dated || r.msg != "ERROR db: connection refused" || r.sev != 3 {
+		t.Errorf("got %+v", r)
+	}
+	r = p.parse("[2026-10-05T10:00:00Z] - started", now)
+	if !r.dated || r.msg != "started" || r.sev != 6 {
+		t.Errorf("got %+v", r)
+	}
+	r = p.parse("\x1b[33mWARN\x1b[0m no timestamp here", now)
+	if r.dated || r.msg != "WARN no timestamp here" || r.sev != 4 || !r.t.Equal(now) {
+		t.Errorf("got %+v", r)
+	}
+}
+
+func TestParseJSON(t *testing.T) {
+	now := time.Now()
+	p := mustParser(t, fmtJSON, "")
+	r := p.parse(`{"time":"2026-10-05T10:00:00Z","level":"warn","msg":"slow query","host":"db.example","req":{"id":7,"tags":["a"]},"ok":true}`, now)
+	if !r.dated || r.sev != 4 || r.msg != "slow query" {
+		t.Errorf("got %+v", r)
+	}
+	for k, want := range map[string]string{"field_host": "db.example", "req.id": "7", "req.tags": `["a"]`, "ok": "true"} {
+		if r.fields[k] != want {
+			t.Errorf("%s = %q, want %q", k, r.fields[k], want)
+		}
+	}
+	// pino: numeric level, epoch in ms.
+	r = p.parse(`{"level":50,"time":1759572000123,"msg":"boom"}`, now)
+	if r.sev != 3 || !r.t.Equal(time.UnixMilli(1759572000123)) {
+		t.Errorf("got %+v", r)
+	}
+	// No message key: the whole line; not JSON: text.
+	line := `{"event":"login","user":"bob"}`
+	if r = p.parse(line, now); r.msg != line || r.fields["user"] != "bob" {
+		t.Errorf("got %+v", r)
+	}
+	if r = p.parse("plain ERROR line", now); r.msg != "plain ERROR line" || r.sev != 3 {
+		t.Errorf("got %+v", r)
+	}
+}
+
+func TestParseLogfmt(t *testing.T) {
+	p := mustParser(t, fmtLogfmt, "")
+	r := p.parse(`time="2026-10-05T10:00:00Z" level=error msg="Cannot connect: \"docker\" down" providerName=docker flag`, time.Now())
+	if !r.dated || r.sev != 3 || r.msg != `Cannot connect: "docker" down` || r.fields["providerName"] != "docker" {
+		t.Errorf("got %+v", r)
+	}
+}
+
+func TestParseRegex(t *testing.T) {
+	if _, err := newFileParser(fmtRegex, "("); err == nil {
+		t.Error("invalid regex accepted")
+	}
+	if _, err := newFileParser(fmtRegex, ""); err == nil {
+		t.Error("empty regex accepted")
+	}
+	p := mustParser(t, fmtRegex, `^(?P<time>\S+ \S+) \[(?P<level>\w+)\] (?P<module>[\w.]+): (?P<msg>.*)$`)
+	r := p.parse("2026-10-05 10:00:00 [warning] app.db: pool exhausted", time.Now())
+	if !r.dated || r.sev != 4 || r.msg != "pool exhausted" || r.fields["module"] != "app.db" {
+		t.Errorf("got %+v", r)
+	}
+	if !p.continuation("Traceback (most recent call last):") || p.continuation("2026-10-05 10:00:01 [info] a: b") {
+		t.Error("continuation")
+	}
+}
+
+func TestParseCLF(t *testing.T) {
+	p := mustParser(t, fmtCLF, "")
+	r := p.parse(`203.0.113.9 - - [05/Oct/2026:10:00:00 +0200] "GET /api/x?y=1 HTTP/1.1" 502 157 "-" "curl/8.0" 4211 "api@docker" "http://172.18.0.5:8080" 31ms`, time.Now())
+	if !r.dated || r.sev != 3 || !strings.HasPrefix(r.msg, "203.0.113.9 - - [") {
+		t.Errorf("got %+v", r)
+	}
+	for k, want := range map[string]string{"client_ip": "203.0.113.9", "http_method": "GET", "http_path": "/api/x?y=1", "http_status": "502",
+		"user_agent": "curl/8.0", "router": "api@docker", "backend": "http://172.18.0.5:8080", "duration_ms": "31"} {
+		if r.fields[k] != want {
+			t.Errorf("%s = %q, want %q", k, r.fields[k], want)
+		}
+	}
+	if _, ok := r.fields["http_referer"]; ok {
+		t.Error("empty referer kept")
+	}
+	// nginx combined, 404.
+	r = p.parse(`10.0.0.1 - alice [05/Oct/2026:10:00:00 +0000] "POST /login HTTP/2.0" 404 0 "https://example.com/" "Mozilla/5.0"`, time.Now())
+	if r.sev != 4 || r.fields["http_user"] != "alice" || r.fields["http_referer"] != "https://example.com/" {
+		t.Errorf("got %+v", r)
+	}
+}
+
+func TestParseTraefikJSON(t *testing.T) {
+	p := mustParser(t, fmtTraefik, "")
+	r := p.parse(`{"ClientHost":"203.0.113.9","DownstreamStatus":200,"Duration":12500000,"RequestHost":"app.example.com","RequestMethod":"GET","RequestPath":"/","RequestProtocol":"HTTP/2.0","RouterName":"app@docker","StartUTC":"2026-10-05T10:00:00.1Z","entryPointName":"websecure","level":"info","msg":"","time":"2026-10-05T10:00:00Z","TLSVersion":"1.3"}`, time.Now())
+	if !r.dated || r.sev != 6 || r.msg != "GET app.example.com/ 200 · 12.5 ms · app@docker · 203.0.113.9" {
+		t.Errorf("got %+v", r)
+	}
+	for k, want := range map[string]string{"entrypoint": "websecure", "http_status": "200", "TLSVersion": "1.3", "duration_ms": "12.5"} {
+		if r.fields[k] != want {
+			t.Errorf("%s = %q, want %q", k, r.fields[k], want)
+		}
+	}
+	// Traefik's own logs in the same file: generic JSON.
+	r = p.parse(`{"level":"error","msg":"Provider error","time":"2026-10-05T10:00:00Z"}`, time.Now())
+	if r.sev != 3 || r.msg != "Provider error" {
+		t.Errorf("got %+v", r)
+	}
+}
+
+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/a.log", Format: "xml"},
+		{Name: "a", Path: "/logs/a.log", Format: fmtRegex, Regex: "(?P<x"},
+	}
+	for _, d := range bad {
+		if err := f.Configure(filesConfig{Sources: []fileSourceDef{d}}); err == nil {
+			t.Errorf("%+v accepted", d)
+		}
+	}
+	if err := f.Configure(filesConfig{Sources: []fileSourceDef{{Name: " traefik ", Path: "/logs/traefik/access.log"}}}); err != nil {
+		t.Fatal(err)
+	}
+	src := f.Status()["sources"].([]map[string]any)
+	if len(src) != 1 || src[0]["name"] != "traefik" || src[0]["id"] == "" || src[0]["format"] != fmtAuto {
+		t.Errorf("got %+v", src)
+	}
+}
+
+type fileSink struct {
+	mu sync.Mutex
+	e  []*Entry
+}
+
+func (s *fileSink) add(e *Entry) { s.mu.Lock(); s.e = append(s.e, e); s.mu.Unlock() }
+func (s *fileSink) take() []*Entry {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	e := s.e
+	s.e = nil
+	return e
+}
+
+func appendFile(t *testing.T, path, s string) {
+	t.Helper()
+	fh, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := fh.WriteString(s); err != nil {
+		t.Fatal(err)
+	}
+	fh.Close()
+}
+
+func msgs(es []*Entry) []string {
+	var out []string
+	for _, e := range es {
+		out = append(out, e.Message)
+	}
+	return out
+}
+
+func TestFilesFollow(t *testing.T) {
+	dir, data := t.TempDir(), t.TempDir()
+	now := time.Now()
+	old := now.Add(-3 * time.Hour).UTC().Format(time.RFC3339)
+	recent := now.Add(-10 * time.Minute).UTC().Format(time.RFC3339)
+	app := filepath.Join(dir, "app.log")
+	appendFile(t, app, old+" INFO too old\n"+recent+" ERROR recent\n  at line 1\nundated history line\n")
+
+	sink := &fileSink{}
+	f := NewFiles(dir, data, time.Hour, sink.add)
+	if err := f.Configure(filesConfig{Sources: []fileSourceDef{{Name: "myapp", Path: app, Host: "web1", Enabled: true}}}); err != nil {
+		t.Fatal(err)
+	}
+	f.poll(now)
+	f.poll(now.Add(3 * time.Second)) // the pending entry waited for continuation lines
+	got := sink.take()
+	if len(got) != 1 || got[0].Message != "ERROR recent\n  at line 1" || got[0].SevNum != 3 || got[0].App != "myapp" ||
+		got[0].Host != "web1" || got[0].SourceType != "file" || got[0].Extra["log_file"] != app {
+		t.Fatalf("history: got %q %+v", msgs(got), got)
+	}
+	if st := f.Status()["sources"].([]map[string]any)[0]; st["detected"] != fmtText || st["open"] != true {
+		t.Errorf("status %+v", st)
+	}
+
+	appendFile(t, app, "plain new line\n\tcontinued\n")
+	f.poll(now.Add(4 * time.Second))
+	appendFile(t, app, "after\n")
+	f.poll(now.Add(5 * time.Second))
+	f.poll(now.Add(8 * time.Second))
+	if got = sink.take(); strings.Join(msgs(got), "|") != "plain new line\n\tcontinued|after" {
+		t.Fatalf("new lines: got %q", msgs(got))
+	}
+
+	appendFile(t, app, "before rotation\n")
+	if err := os.Rename(app, app+".1"); err != nil {
+		t.Fatal(err)
+	}
+	appendFile(t, app, "after rotation\n")
+	f.poll(now.Add(9 * time.Second))
+	f.poll(now.Add(12 * time.Second))
+	if got = sink.take(); strings.Join(msgs(got), "|") != "before rotation|after rotation" {
+		t.Fatalf("rotation: got %q", msgs(got))
+	}
+
+	// Restart: resumes where it stopped, nothing sent twice.
+	f.checkpointAll()
+	f.dirty = true
+	f.saveState()
+	appendFile(t, app, "while stopped\n")
+	f2 := NewFiles(dir, data, time.Hour, sink.add)
+	f2.poll(now.Add(13 * time.Second))
+	f2.poll(now.Add(16 * time.Second))
+	if got = sink.take(); strings.Join(msgs(got), "|") != "while stopped" {
+		t.Fatalf("restart: got %q", msgs(got))
+	}
+
+	// Truncated in place (copytruncate).
+	os.Truncate(app, 0)
+	appendFile(t, app, "x\n")
+	f2.poll(now.Add(17 * time.Second))
+	f2.poll(now.Add(20 * time.Second))
+	if got = sink.take(); strings.Join(msgs(got), "|") != "x" {
+		t.Fatalf("truncate: got %q", msgs(got))
+	}
+}
+
+func TestFilesAutoAndErrors(t *testing.T) {
+	dir, data := t.TempDir(), t.TempDir()
+	path := filepath.Join(dir, "access.json")
+	sink := &fileSink{}
+	f := NewFiles(dir, data, 0, sink.add) // FILES_BACKFILL=0: only new lines
+	cfg := filesConfig{Sources: []fileSourceDef{
+		{ID: "a1", Name: "api", Path: path, Enabled: true}, // auto
+		{ID: "b2", Name: "missing", Path: filepath.Join(dir, "nope.log"), Format: fmtText, Enabled: true},
+	}}
+	if err := f.Configure(cfg); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now()
+	appendFile(t, path, "")
+	f.poll(now) // empty file: format not known yet
+	appendFile(t, path, `{"msg":"one","level":"error"}`+"\n"+`{"msg":"two"}`+"\n"+`{"msg":"partial`)
+	f.poll(now.Add(time.Second))
+	got := sink.take()
+	if strings.Join(msgs(got), "|") != "one|two" || got[0].SevNum != 3 {
+		t.Fatalf("got %q", msgs(got))
+	}
+	src := f.Status()["sources"].([]map[string]any)
+	if src[0]["detected"] != fmtJSON || src[0]["open"] != true || src[1]["code"] != "not_found" {
+		t.Errorf("status %+v", src)
+	}
+
+	// Forced format: the JSON lines are read as text.
+	cfg.Sources[0].Format = fmtText
+	f.Configure(cfg)
+	appendFile(t, path, `"}`+"\n")
+	f.poll(now.Add(2 * time.Second))
+	f.poll(now.Add(5 * time.Second))
+	if got = sink.take(); strings.Join(msgs(got), "|") != `{"msg":"partial"}` {
+		t.Fatalf("forced: got %q", msgs(got))
+	}
+
+	// Turned off then on: starts over (only new lines here).
+	cfg.Sources[0].Enabled = false
+	f.Configure(cfg)
+	f.poll(now.Add(6 * time.Second))
+	appendFile(t, path, "while off\n")
+	cfg.Sources[0].Enabled = true
+	f.Configure(cfg)
+	f.poll(now.Add(7 * time.Second))
+	appendFile(t, path, "three\n")
+	f.poll(now.Add(8 * time.Second))
+	f.poll(now.Add(11 * time.Second))
+	if got = sink.take(); strings.Join(msgs(got), "|") != "three" {
+		t.Fatalf("re-enabled: got %q", msgs(got))
+	}
+}

hostlogs.go

@@ -7,7 +7,6 @@ import (
 	"errors"
 	"io"
 	"io/fs"
-	"log"
 	"os"
 	"path/filepath"
 	"sort"
@@ -116,14 +115,18 @@ func (h *HostLogs) saveState() {
 	}
 	h.dirty = false
 	if err := writeJSONFile(h.statePath, h.pos); err != nil {
-		log.Printf("host logs: saving positions: %v", err)
+		logEvent("host", 4, "saving positions: %v", err)
 	}
 }
 
 func (h *HostLogs) setStatus(mode string, files int, code, detail string) {
 	h.mu.Lock()
+	changed := code != h.errCode
 	h.mode, h.files, h.errCode, h.errDetail = mode, files, code, detail
 	h.mu.Unlock()
+	if changed && code != "" && code != "empty" {
+		logEvent("host", 4, "cannot read the host system logs (%s): %s", code, detail)
+	}
 }
 
 // scan reads what was added since the previous poll.

main.go

@@ -38,6 +38,12 @@ type config struct {
 	backfill   time.Duration
 	hostRoot   string
 	hostFill   time.Duration
+	kmsgDevice string
+	kmsgFill   time.Duration
+	auditLog   string
+	auditHost  string
+	auditFill  time.Duration
+	filesFill  time.Duration
 	batchSize  int
 	queueSize  int
 	flushEvery time.Duration
@@ -115,6 +121,12 @@ func main() {
 		backfill:   getenvDuration("DOCKER_BACKFILL", time.Hour),
 		hostRoot:   getenv("HOST_LOGS_ROOT", "/host"),
 		hostFill:   getenvDuration("HOST_LOGS_BACKFILL", time.Hour),
+		kmsgDevice: getenv("DMESG_DEVICE", "/dev/kmsg"),
+		kmsgFill:   getenvDuration("DMESG_BACKFILL", time.Hour),
+		auditLog:   getenv("AUDIT_LOG", "/var/log/audit/audit.log"),
+		auditHost:  os.Getenv("AUDIT_HOST"),
+		auditFill:  getenvDuration("AUDIT_BACKFILL", time.Hour),
+		filesFill:  getenvDuration("FILES_BACKFILL", time.Hour),
 		batchSize:  getenvInt("BATCH_SIZE", 1000),
 		queueSize:  getenvInt("QUEUE_SIZE", 100000),
 		flushEvery: time.Duration(getenvInt("FLUSH_MS", 1000)) * time.Millisecond,
@@ -134,14 +146,16 @@ func main() {
 	var spool *Spool
 	if cfg.spoolMax > 0 {
 		if spool, err = OpenSpool(filepath.Join(cfg.dataDir, "spool"), cfg.spoolMax); err != nil {
-			log.Printf("disk buffer disabled: %v", err)
+			log.Printf("disk buffer disabled: %v", err) // before the self-logs exist
 			spool = nil
 		}
 	}
 	store := NewStore(cfg.vlogsURL, cfg.batchSize, cfg.queueSize, cfg.flushEvery, spool)
+	// The store stops after everything else, so the stop event is still stored.
+	storeCtx, stopStore := context.WithCancel(context.Background())
 	storeDone := make(chan struct{})
 	go func() {
-		store.Run(ctx)
+		store.Run(storeCtx)
 		close(storeDone)
 	}()
 
@@ -159,6 +173,20 @@ func main() {
 			hub.Publish(e)
 		})
 	}
+	// Internal events of LogStream (host=logstream), WARNING and above unless
+	// changed in Settings > Sources.
+	selfLog = NewSelfLog(cfg.dataDir, func(e *Entry) {
+		store.Enqueue(e)
+		hub.Publish(e)
+	})
+	selfCtx, stopSelf := context.WithCancel(context.Background())
+	selfDone := make(chan struct{})
+	go func() {
+		selfLog.Run(selfCtx)
+		close(selfDone)
+	}()
+	logEvent("main", 5, "LogStream started")
+
 	tcpMaxConns = getenvInt("SYSLOG_TCP_MAX_CONNS", tcpMaxConns)
 	if d := getenvDuration("SYSLOG_TCP_IDLE", tcpIdle); d > 0 {
 		tcpIdle = d
@@ -176,7 +204,7 @@ func main() {
 	if cfg.dockerLogs {
 		dm, err := NewDockerManager(cfg.dockerHost, cfg.dataDir, cfg.backfill, sink)
 		if err != nil {
-			log.Printf("docker logs disabled: %v", err)
+			logEvent("docker", 3, "Docker logs disabled: %v", err)
 		} else {
 			go dm.Run(ctx)
 			api.docker = dm
@@ -186,6 +214,17 @@ func main() {
 	// System logs of the host (journal or /var/log), off until enabled in Settings > Sources.
 	api.host = NewHostLogs(cfg.hostRoot, cfg.dataDir, cfg.hostFill, sink)
 	go api.host.Run(ctx)
+	// Kernel messages of the host (/dev/kmsg), off until enabled in Settings > Sources.
+	api.dmesg = NewDmesg(cfg.kmsgDevice, cfg.dataDir, cfg.kmsgFill, sink)
+	go api.dmesg.Run(ctx)
+	// Linux audit log of the host (auditd), off until enabled in Settings > Sources.
+	api.audit = NewAudit(cfg.auditLog, cfg.hostRoot, cfg.auditHost, cfg.dataDir, cfg.auditFill, sink)
+	go api.audit.Run(ctx)
+	// Log files of other programs (Traefik, nginx…) mounted in the container,
+	// declared in Settings > Sources.
+	api.files = NewFiles(cfg.hostRoot, cfg.dataDir, cfg.filesFill, sink)
+	go api.files.Run(ctx)
+	api.selfLog = selfLog
 	api.Routes(mux)
 	mux.Handle("GET /", http.FileServer(http.FS(static)))
 
@@ -222,6 +261,10 @@ func main() {
 	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		log.Fatalf("http: %v", err)
 	}
+	logEvent("main", 5, "LogStream stopping")
+	stopSelf()
+	<-selfDone
+	stopStore()
 	<-storeDone
 	log.Println("shutdown complete")
 }

query.go

@@ -119,10 +119,10 @@ func (f Filter) filterExpr() string {
 		parts = append(parts, "app:="+strconv.Quote(f.App))
 	}
 	switch f.Source {
-	case "docker", "host":
+	case "docker", "host", "kernel", "audit", "logstream", "file":
 		parts = append(parts, `source_type:=`+strconv.Quote(f.Source))
 	case "syslog": // also matches logs stored before source_type existed
-		parts = append(parts, `!(source_type:in("docker","host"))`)
+		parts = append(parts, `!(source_type:in("docker","host","kernel","audit","logstream","file"))`)
 	}
 	if f.Severity >= 0 && f.Severity < 7 {
 		names := make([]string, 0, 8)

selflog.go

@@ -0,0 +1,187 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+)
+
+// selfLogConfig is saved in /data/selflog.json and edited in Settings > Sources.
+type selfLogConfig struct {
+	Enabled  bool   `json:"enabled"`
+	MinLevel string `json:"min_level"` // severity name: "err", "warning", "notice", "info"…
+}
+
+// SelfLog turns the internal events of LogStream (storage errors, listener
+// failures, sources that stop…) into log entries, with host=logstream and
+// app=component. Events go through a bounded queue and a per-message
+// throttle: an event about a failing Store is itself stored through that
+// Store, so it must never be able to loop or pile up.
+type SelfLog struct {
+	cfgPath  string
+	sink     func(*Entry)
+	queue    chan *Entry
+	interval time.Duration // one event per component and message format per interval
+
+	mu         sync.Mutex
+	cfg        selfLogConfig
+	last       map[string]time.Time
+	suppressed map[string]int
+	emitted    uint64
+	dropped    uint64
+	throttled  uint64
+}
+
+// selfLog receives the events of logEvent; nil (tests) only writes them to
+// the process log.
+var selfLog *SelfLog
+
+func NewSelfLog(dataDir string, sink func(*Entry)) *SelfLog {
+	s := &SelfLog{
+		cfgPath:    filepath.Join(dataDir, "selflog.json"),
+		sink:       sink,
+		queue:      make(chan *Entry, 256),
+		interval:   30 * time.Second,
+		cfg:        selfLogConfig{Enabled: true, MinLevel: "warning"},
+		last:       map[string]time.Time{},
+		suppressed: map[string]int{},
+	}
+	if b, err := os.ReadFile(s.cfgPath); err == nil {
+		var cfg selfLogConfig
+		if json.Unmarshal(b, &cfg) == nil {
+			s.cfg = normSelfLogConfig(cfg)
+		}
+	}
+	return s
+}
+
+func normSelfLogConfig(cfg selfLogConfig) selfLogConfig {
+	if sevNum(cfg.MinLevel) < 0 {
+		cfg.MinLevel = "warning"
+	}
+	return cfg
+}
+
+func sevNum(name string) int {
+	for i, n := range severityNames {
+		if n == name {
+			return i
+		}
+	}
+	return -1
+}
+
+// logEvent writes an internal event to the process log and, when the
+// self-logs source is on and the level is high enough, emits it as an entry.
+// sev is a syslog severity (3 err, 4 warning, 5 notice, 6 info).
+func logEvent(component string, sev int, format string, args ...any) {
+	msg := fmt.Sprintf(format, args...)
+	log.Printf("%s: %s", component, msg)
+	if s := selfLog; s != nil {
+		s.Emit(component, sev, format, msg, time.Now())
+	}
+}
+
+// Emit queues an event without ever blocking. The same component and message
+// format are let through once per interval; the next one says how many were
+// left out in between.
+func (s *SelfLog) Emit(component string, sev int, format, msg string, now time.Time) {
+	sev = min(max(sev, 0), 7)
+	key := component + "\x00" + format
+	s.mu.Lock()
+	if !s.cfg.Enabled || sev > sevNum(s.cfg.MinLevel) {
+		s.mu.Unlock()
+		return
+	}
+	if t, ok := s.last[key]; ok && now.Sub(t) < s.interval {
+		s.suppressed[key]++
+		s.throttled++
+		s.mu.Unlock()
+		return
+	}
+	s.last[key] = now
+	if n := s.suppressed[key]; n > 0 {
+		msg += fmt.Sprintf(" (%d similar events not shown)", n)
+		delete(s.suppressed, key)
+	}
+	if len(s.last) > 1000 { // formats are constants: only a bug could get here
+		clear(s.last)
+	}
+	s.mu.Unlock()
+
+	e := &Entry{
+		Time:       now,
+		Received:   now,
+		Host:       "logstream",
+		App:        component,
+		Facility:   "syslog",
+		Severity:   severityNames[sev],
+		SevNum:     sev,
+		Message:    msg,
+		Proto:      "internal",
+		SourceType: "logstream",
+		// Wait stays false: when the Store queue is full, the event is dropped
+		// instead of blocking the component that reports it.
+	}
+	select {
+	case s.queue <- e:
+	default:
+		s.mu.Lock()
+		s.dropped++
+		s.mu.Unlock()
+	}
+}
+
+// Run hands the queued events to the sink until the context ends, then
+// delivers what is left (the stop event among them).
+func (s *SelfLog) Run(ctx context.Context) {
+	for {
+		select {
+		case e := <-s.queue:
+			s.deliver(e)
+		case <-ctx.Done():
+			for {
+				select {
+				case e := <-s.queue:
+					s.deliver(e)
+				default:
+					return
+				}
+			}
+		}
+	}
+}
+
+func (s *SelfLog) deliver(e *Entry) {
+	s.sink(e)
+	s.mu.Lock()
+	s.emitted++
+	s.mu.Unlock()
+}
+
+// Configure saves and applies the configuration.
+func (s *SelfLog) Configure(cfg selfLogConfig) error {
+	cfg = normSelfLogConfig(cfg)
+	s.mu.Lock()
+	s.cfg = cfg
+	s.mu.Unlock()
+	return writeJSONFile(s.cfgPath, cfg)
+}
+
+// Status is the state shown in Settings > Sources.
+func (s *SelfLog) Status() map[string]any {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return map[string]any{
+		"enabled":   s.cfg.Enabled,
+		"min_level": s.cfg.MinLevel,
+		"emitted":   s.emitted,
+		"dropped":   s.dropped,
+		"throttled": s.throttled,
+	}
+}

selflog_test.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSelfLogEmit(t *testing.T) {
+	now := time.Unix(1759572000, 0)
+	tests := []struct {
+		name   string
+		cfg    selfLogConfig
+		events []int // severities, one second apart, same message format
+		want   int   // entries queued
+		suffix string
+	}{
+		{"warning and above", selfLogConfig{Enabled: true, MinLevel: "warning"}, []int{3}, 1, ""},
+		{"below the level", selfLogConfig{Enabled: true, MinLevel: "warning"}, []int{5, 6}, 0, ""},
+		{"info level", selfLogConfig{Enabled: true, MinLevel: "info"}, []int{6}, 1, ""},
+		{"disabled", selfLogConfig{Enabled: false, MinLevel: "debug"}, []int{0}, 0, ""},
+		{"bad level falls back to warning", normSelfLogConfig(selfLogConfig{Enabled: true, MinLevel: "loud"}), []int{4, 5}, 1, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := NewSelfLog(t.TempDir(), func(*Entry) {})
+			s.cfg = tt.cfg
+			for i, sev := range tt.events {
+				s.Emit("store", sev, "fmt %d", "msg", now.Add(time.Duration(i)*time.Second))
+			}
+			if len(s.queue) != tt.want {
+				t.Fatalf("%d queued, want %d", len(s.queue), tt.want)
+			}
+		})
+	}
+
+	// Same component and format: once per interval, then a count of the others.
+	s := NewSelfLog(t.TempDir(), func(*Entry) {})
+	for i := 0; i < 5; i++ {
+		s.Emit("store", 3, "%d messages lost", "x", now.Add(time.Duration(i)*time.Second))
+	}
+	s.Emit("syslog", 3, "%d messages lost", "y", now) // another component
+	s.Emit("store", 3, "%d messages lost", "z", now.Add(s.interval))
+	if len(s.queue) != 3 {
+		t.Fatalf("%d queued", len(s.queue))
+	}
+	<-s.queue
+	if e := <-s.queue; e.App != "syslog" || e.Host != "logstream" || e.SourceType != "logstream" || e.Wait {
+		t.Fatalf("entry %+v", e)
+	}
+	if e := <-s.queue; e.Message != "z (4 similar events not shown)" || e.Severity != "err" {
+		t.Fatalf("message %q", e.Message)
+	}
+	if st := s.Status(); st["throttled"] != uint64(4) {
+		t.Fatalf("status %v", st)
+	}
+
+	// A full queue drops events instead of blocking.
+	s = NewSelfLog(t.TempDir(), func(*Entry) {})
+	for i := 0; i < cap(s.queue)+10; i++ {
+		s.Emit("store", 3, "event %d", "x", now.Add(time.Duration(i)*s.interval))
+	}
+	if st := s.Status(); st["dropped"] != uint64(10) {
+		t.Fatalf("status %v", st)
+	}
+}
+
+func TestSelfLogRunAndConfigure(t *testing.T) {
+	dir := t.TempDir()
+	sink := &entrySink{}
+	s := NewSelfLog(dir, sink.add)
+	if err := s.Configure(selfLogConfig{Enabled: true, MinLevel: "notice"}); err != nil {
+		t.Fatal(err)
+	}
+	if b, _ := os.ReadFile(filepath.Join(dir, "selflog.json")); !strings.Contains(string(b), `"notice"`) {
+		t.Fatalf("saved %s", b)
+	}
+	if s2 := NewSelfLog(dir, nil); s2.cfg.MinLevel != "notice" {
+		t.Fatalf("reloaded %+v", s2.cfg)
+	}
+	old := selfLog
+	selfLog = s
+	defer func() { selfLog = old }()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { s.Run(ctx); close(done) }()
+	logEvent("main", 5, "LogStream started")
+	logEvent("main", 6, "too verbose")
+	cancel()
+	<-done
+	logEvent("main", 5, "LogStream stopping") // queued after Run: not lost if Run drains it
+	sink.mu.Lock()
+	n := len(sink.es)
+	sink.mu.Unlock()
+	if n != 1 || s.Status()["emitted"] != uint64(1) {
+		t.Fatalf("%d entries, status %v", n, s.Status())
+	}
+}

store.go

@@ -8,7 +8,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log"
 	"net/http"
 	"net/url"
 	"strings"
@@ -34,6 +33,7 @@ type Store struct {
 	ingested atomic.Int64
 	dropped  atomic.Int64
 	lastErr  atomic.Value // string
+	failing  atomic.Bool  // VictoriaLogs refused or missed the last send
 }
 
 func NewStore(base string, batchSize, queueSize int, flushEvery time.Duration, spool *Spool) *Store {
@@ -133,7 +133,7 @@ func (s *Store) store(ctx context.Context, batch []*Entry) {
 	if err != nil {
 		s.dropped.Add(int64(len(batch)))
 		s.lastErr.Store(err.Error())
-		log.Printf("victorialogs: %d messages dropped: %v", len(batch), err)
+		logEvent("store", 3, "%d messages lost: %v", len(batch), err)
 		return
 	}
 	for _, e := range batch {
@@ -153,9 +153,10 @@ func (s *Store) save(ctx context.Context, body []byte, lines int) error {
 			}
 			if err = s.post(ctx, body); err == nil {
 				s.ingested.Add(int64(lines))
-				s.lastErr.Store("")
+				s.sent()
 				return nil
 			}
+			s.sendFailed(err)
 		}
 		return err
 	}
@@ -164,10 +165,10 @@ func (s *Store) save(ctx context.Context, body []byte, lines int) error {
 	if waiting, _ := s.spool.Pending(); waiting == 0 {
 		if postErr = s.post(ctx, body); postErr == nil {
 			s.ingested.Add(int64(lines))
-			s.lastErr.Store("")
+			s.sent()
 			return nil
 		}
-		s.lastErr.Store(postErr.Error())
+		s.sendFailed(postErr)
 	}
 	err := s.spool.Write(body, lines)
 	if err == nil {
@@ -192,13 +193,13 @@ func (s *Store) save(ctx context.Context, body []byte, lines int) error {
 // an increasing pause (up to 30 s) while it keeps failing.
 func (s *Store) replay(ctx context.Context) {
 	if n, size := s.spool.Pending(); n > 0 {
-		log.Printf("spool: %d messages (%d bytes) waiting from a previous run", n, size)
+		logEvent("store", 5, "disk spool: %d messages (%d bytes) waiting from a previous run", n, size)
 	}
 	backoff := time.Second
 	for {
 		f, body, ok, err := s.spool.Oldest()
 		if err != nil {
-			log.Printf("spool: %v", err)
+			logEvent("store", 4, "disk spool: %v", err)
 		}
 		if !ok {
 			select {
@@ -214,17 +215,17 @@ func (s *Store) replay(ctx context.Context) {
 		case err == nil:
 			s.spool.Remove(f)
 			s.ingested.Add(f.lines)
-			s.lastErr.Store("")
+			s.sent()
 			backoff = time.Second
 			continue
 		case isRejected(err):
 			// VictoriaLogs refuses the data itself: sending it again would not help.
 			s.spool.Remove(f)
 			s.dropped.Add(f.lines)
-			log.Printf("spool: %d messages refused by victorialogs: %v", f.lines, err)
+			logEvent("store", 3, "disk spool: %d messages refused by VictoriaLogs and lost: %v", f.lines, err)
 			continue
 		}
-		s.lastErr.Store(err.Error())
+		s.sendFailed(err)
 		if !sleepCtx(ctx, backoff) {
 			return
 		}
@@ -232,6 +233,26 @@ func (s *Store) replay(ctx context.Context) {
 	}
 }
 
+// sendFailed and sent report when VictoriaLogs stops and starts taking the
+// batches again, once per change.
+func (s *Store) sendFailed(err error) {
+	s.lastErr.Store(err.Error())
+	if !s.failing.Swap(true) {
+		if s.spool != nil {
+			logEvent("store", 4, "VictoriaLogs unreachable, batches are kept in the disk spool: %v", err)
+		} else {
+			logEvent("store", 4, "VictoriaLogs unreachable, retrying: %v", err)
+		}
+	}
+}
+
+func (s *Store) sent() {
+	s.lastErr.Store("")
+	if s.failing.Swap(false) {
+		logEvent("store", 5, "VictoriaLogs reachable again")
+	}
+}
+
 func encodeBatch(batch []*Entry) ([]byte, error) {
 	var buf bytes.Buffer
 	enc := json.NewEncoder(&buf)

syslog.go

@@ -36,7 +36,7 @@ type Entry struct {
 	Proto    string // udp or tcp
 	HostIP   string // original host value when it was an IP resolved through DNS
 
-	SourceType string            // "syslog" or "docker"
+	SourceType string            // "syslog", "docker", "host", "kernel", "audit", "file" or "logstream"
 	Extra      map[string]string // additional fields (docker: container, image, …)
 
 	// Not stored. Wait: the producer can be slowed down when the queue is full
@@ -239,7 +239,7 @@ func serveUDP(ctx context.Context, pc net.PacketConn, sink func(*Entry)) {
 			if ctx.Err() != nil {
 				return
 			}
-			log.Printf("syslog udp: %v", err)
+			logEvent("syslog", 4, "UDP read error: %v", err)
 			time.Sleep(100 * time.Millisecond)
 			continue
 		}
@@ -262,14 +262,14 @@ func serveTCP(ctx context.Context, ln net.Listener, sink func(*Entry)) {
 			if ctx.Err() != nil {
 				return
 			}
-			log.Printf("syslog tcp: %v", err)
+			logEvent("syslog", 4, "TCP accept error: %v", err)
 			time.Sleep(100 * time.Millisecond)
 			continue
 		}
 		select {
 		case slots <- struct{}{}:
 		default:
-			log.Printf("syslog tcp: %d connections already open, refusing %s", tcpMaxConns, conn.RemoteAddr())
+			logEvent("syslog", 4, "%d TCP connections already open, refusing %s", tcpMaxConns, conn.RemoteAddr())
 			conn.Close()
 			continue
 		}

syslogserver.go

@@ -94,7 +94,7 @@ func (s *SyslogServer) apply() {
 		pc, err := net.ListenPacket("udp", s.addr)
 		if err != nil {
 			s.errUDP = err.Error()
-			log.Printf("syslog udp: %v", err)
+			logEvent("syslog", 3, "UDP listener failed: %v", err)
 		} else {
 			if uc, ok := pc.(*net.UDPConn); ok {
 				_ = uc.SetReadBuffer(8 << 20)
@@ -108,7 +108,7 @@ func (s *SyslogServer) apply() {
 		ln, err := net.Listen("tcp", s.addr)
 		if err != nil {
 			s.errTCP = err.Error()
-			log.Printf("syslog tcp: %v", err)
+			logEvent("syslog", 3, "TCP listener failed: %v", err)
 		} else {
 			s.ln = ln
 			go serveTCP(ctx, ln, s.sink)

tags.go

@@ -217,6 +217,51 @@ func (s *TagStore) Delete(id string) error {
 	return errTagNotFound
 }
 
+// tagListError tells which tag of a list (PUT /api/tags) is invalid.
+type tagListError struct {
+	index int
+	err   error
+}
+
+func (e *tagListError) Error() string { return e.err.Error() }
+func (e *tagListError) Unwrap() error { return e.err }
+
+// Replace saves a whole list at once (the Save button of the settings). A tag
+// whose id is known keeps its code; the others get a new id and a free code.
+// Nothing is saved when a tag is invalid: the error gives its position.
+func (s *TagStore) Replace(in []Tag) ([]Tag, error) {
+	if len(in) > 99 {
+		return nil, errTooManyTags
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	codes := map[string]string{}
+	for _, t := range s.tags {
+		codes[t.ID] = t.Code
+	}
+	out := make([]Tag, len(in))
+	seen := map[string]bool{}
+	for i, t := range in {
+		if err := t.validate(); err != nil {
+			return nil, &tagListError{index: i, err: err}
+		}
+		if c, ok := codes[t.ID]; ok && !seen[t.ID] {
+			t.Code = c
+		} else {
+			t.ID, t.Code = newID(), ""
+		}
+		seen[t.ID] = true
+		out[i] = t
+	}
+	for i := range out {
+		if out[i].Code == "" {
+			out[i].Code = freeCode(out)
+		}
+	}
+	s.tags = out
+	return append([]Tag(nil), out...), s.save()
+}
+
 func (s *TagStore) Reset() ([]Tag, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

tags_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"errors"
 	"os"
 	"path/filepath"
 	"slices"
@@ -52,3 +53,37 @@ func TestTagCodes(t *testing.T) {
 		t.Fatalf("reloaded code = %q, want 07", got)
 	}
 }
+
+func TestTagReplace(t *testing.T) {
+	s, err := LoadTagStore(filepath.Join(t.TempDir(), "tags.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Defaults: warning 01, error 02. Keep error, drop warning, add two tags
+	// (one with an unknown id, one repeating a known id).
+	got, err := s.Replace([]Tag{
+		{ID: "new1", Pattern: "a", Color: "#000000"},
+		{ID: "error", Code: "42", Pattern: "error", Color: "#111111", Enabled: true},
+		{ID: "error", Pattern: "b", Color: "#000000"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	codes := []string{got[0].Code, got[1].Code, got[2].Code}
+	if want := []string{"01", "02", "03"}; !slices.Equal(codes, want) {
+		t.Fatalf("codes = %v, want %v", codes, want)
+	}
+	if got[1].ID != "error" || got[0].ID == "new1" || got[2].ID == "error" || got[1].Color != "#111111" {
+		t.Fatalf("ids/fields = %+v", got)
+	}
+
+	// An invalid tag saves nothing and gives its position.
+	_, err = s.Replace([]Tag{{Pattern: "ok", Color: "#000000"}, {Pattern: "(", Regex: true, Color: "#000000"}})
+	var le *tagListError
+	if !errors.As(err, &le) || le.index != 1 {
+		t.Fatalf("err = %v, want tagListError at 1", err)
+	}
+	if n := len(s.List()); n != 3 {
+		t.Fatalf("list after a failed Replace has %d tags, want 3", n)
+	}
+}

web/index.html

@@ -41,6 +41,7 @@
         <svg class="sun" viewBox="0 0 24 24"><circle cx="12" cy="12" r="4"/><path d="M12 2v2M12 20v2M4.9 4.9l1.4 1.4M17.7 17.7l1.4 1.4M2 12h2M20 12h2M4.9 19.1l1.4-1.4M17.7 6.3l1.4-1.4"/></svg>
         <svg class="moon" viewBox="0 0 24 24"><path d="M21 12.8A9 9 0 1 1 11.2 3a7 7 0 0 0 9.8 9.8z"/></svg>
       </button>
+      <span id="whoami" class="whoami" hidden><span class="who-name"></span><span class="who-badge"></span></span>
       <a id="logoutBtn" class="icon-btn" href="/auth/logout" hidden data-i18n-title="logout" data-i18n-aria="logout">
         <!-- Log out icon (Lucide "log-out", ISC license) -->
         <svg viewBox="0 0 24 24"><path d="M9 21H5a2 2 0 0 1-2-2V5a2 2 0 0 1 2-2h4"/><path d="m16 17 5-5-5-5"/><path d="M21 12H9"/></svg>
@@ -102,6 +103,10 @@
       <option value="syslog">Syslog</option>
       <option value="docker">Docker</option>
       <option value="host" data-i18n="srcHost">Host system</option>
+      <option value="kernel" data-i18n="srcKernel">Kernel (dmesg)</option>
+      <option value="audit">Audit</option>
+      <option value="file" data-i18n="srcFile">Files</option>
+      <option value="logstream">LogStream</option>
     </select>
     <span class="spacer"></span>
     <span id="count" class="muted"></span>
@@ -140,7 +145,7 @@
       <span data-col="sev"><span class="lbl" data-i18n="colSev">Severity</span><i class="grip" data-i18n-title="colGrip"></i></span>
       <span data-col="host"><span class="lbl" data-i18n="colHost">Host</span><i class="grip" data-i18n-title="colGrip"></i></span>
       <span data-col="app"><span class="lbl" data-i18n="colApp">App</span><i class="grip" data-i18n-title="colGrip"></i></span>
-      <span data-col="codes"><span class="lbl" data-i18n="colCodes" data-i18n-title="codesTitle">Filters</span></span>
+      <span data-col="codes"><span class="lbl" data-i18n="colCodes" data-i18n-title="codesTitle">Pattern</span></span>
       <span data-col="msg"><span class="lbl" data-i18n="colMsg">Message</span></span>
     </div>
     <main id="list" class="list"></main>
@@ -169,6 +174,7 @@
       </header>
 
       <div class="set-body">
+        <div class="set-side">
         <nav id="setNav" class="set-nav" role="tablist" aria-orientation="vertical">
           <button type="button" role="tab" data-tab="locale">
             <svg viewBox="0 0 24 24"><circle cx="12" cy="12" r="9"/><path d="M3 12h18M12 3a14 14 0 0 1 0 18M12 3a14 14 0 0 0 0 18"/></svg>
@@ -176,21 +182,30 @@
           </button>
           <button type="button" role="tab" data-tab="filters">
             <svg viewBox="0 0 24 24"><path d="M3 12V4a1 1 0 0 1 1-1h8l9 9-9 9z"/><circle cx="8" cy="8" r="1.5"/></svg>
-            <span data-i18n="tabFilters">Filters</span>
+            <span data-i18n="tabFilters">Patterns</span>
           </button>
           <button type="button" role="tab" data-tab="sources">
             <svg viewBox="0 0 24 24"><path d="M21 8 12 3 3 8v8l9 5 9-5z"/><path d="m3 8 9 5 9-5M12 13v8"/></svg>
-            <span data-i18n="tabSources">Sources</span>
+            <span data-i18n="tabSources">Log sources</span>
           </button>
           <button type="button" role="tab" data-tab="interface">
             <svg viewBox="0 0 24 24"><rect x="3" y="4" width="18" height="12" rx="2"/><path d="M8 20h8M12 16v4"/></svg>
-            <span data-i18n="tabInterface">Interface</span>
+            <span data-i18n="tabInterface">Appearance</span>
           </button>
           <button type="button" role="tab" data-tab="data">
             <svg viewBox="0 0 24 24"><ellipse cx="12" cy="5" rx="8" ry="3"/><path d="M4 5v14c0 1.7 3.6 3 8 3s8-1.3 8-3V5M4 12c0 1.7 3.6 3 8 3s8-1.3 8-3"/></svg>
-            <span data-i18n="tabData">Data</span>
+            <span data-i18n="tabData">Database</span>
           </button>
         </nav>
+        <div id="setActions" class="set-actions">
+          <button id="setSave" class="btn save" type="button" disabled>
+            <svg viewBox="0 0 24 24" aria-hidden="true"><path d="M5 12.5 10 17l9-10"/></svg>
+            <span data-i18n="saveSettings">Save</span>
+          </button>
+          <button id="setCancel" class="btn ghost" type="button" disabled data-i18n="cancelSettings">Cancel</button>
+          <p id="setState" class="set-state" role="status"></p>
+        </div>
+        </div>
 
         <div class="set-panels">
           <!-- Localization -->
@@ -215,7 +230,7 @@
             </section>
           </div>
 
-          <!-- Filters -->
+          <!-- Pattern (color tags; the "filters" tab id is kept for saved settings) -->
           <div class="set-panel wide" role="tabpanel" data-panel="filters" hidden>
             <section class="set-section tags-section">
               <div class="sec-head">
@@ -265,6 +280,58 @@
               <p class="muted small" data-i18n="hostHelp"></p>
             </section>
 
+            <section class="set-section">
+              <h3 data-i18n="dmesgTitle">Kernel messages (dmesg)</h3>
+              <p id="dmesgStatus" class="docker-status"></p>
+              <label class="switch-row">
+                <input id="dmesgEnabled" type="checkbox" class="switch">
+                <span data-i18n="dmesgEnabled">Collect the kernel messages of this machine</span>
+              </label>
+              <p class="muted small" data-i18n="dmesgHelp"></p>
+            </section>
+
+            <section class="set-section">
+              <h3 data-i18n="auditTitle">Linux audit (auditd)</h3>
+              <p id="auditStatus" class="docker-status"></p>
+              <label class="switch-row">
+                <input id="auditEnabled" type="checkbox" class="switch">
+                <span data-i18n="auditEnabled">Collect the audit events of this machine</span>
+              </label>
+              <div class="sl-grid">
+                <label class="lbl" for="auditPath" data-i18n="auditPath">File</label>
+                <input id="auditPath" type="text" class="dk-filter" autocomplete="off" spellcheck="false">
+              </div>
+              <p class="muted small src-help" data-i18n="auditHelp"></p>
+            </section>
+
+            <section class="set-section">
+              <h3 data-i18n="selfTitle">LogStream internal logs</h3>
+              <p id="selfStatus" class="docker-status"></p>
+              <label class="switch-row">
+                <input id="selfEnabled" type="checkbox" class="switch">
+                <span data-i18n="selfEnabled">Record the events of LogStream itself</span>
+              </label>
+              <div class="sl-grid">
+                <label class="lbl" for="selfLevel" data-i18n="selfLevel">Minimum level</label>
+                <select id="selfLevel" class="field self-level">
+                  <option value="err" data-i18n="sev3">Errors and above</option>
+                  <option value="warning" data-i18n="sev4">Warnings and above</option>
+                  <option value="notice" data-i18n="sev5">Notice and above</option>
+                  <option value="info" data-i18n="sev6">Info and above</option>
+                </select>
+              </div>
+              <p class="muted small src-help" data-i18n="selfHelp"></p>
+            </section>
+
+            <section class="set-section span">
+              <h3 data-i18n="filesTitle">Log files</h3>
+              <div id="fileList" class="file-list"></div>
+              <div class="file-toolbar">
+                <button id="addFile" class="btn primary" type="button" data-i18n="addFile">+ Add a file</button>
+              </div>
+              <p class="muted small src-help" data-i18n="filesHelp"></p>
+            </section>
+
             <section class="set-section span">
               <h3 data-i18n="dockerTitle">Docker containers</h3>
               <p id="dockerStatus" class="docker-status"></p>

web/style.css

@@ -22,6 +22,10 @@
   --hit: rgba(99, 102, 241, .18);
   --shadow: 0 10px 30px rgba(15, 23, 42, .12);
   --chip-l: 38%;            /* lightness of project-colored text (labels, Docker apps) */
+  --src-audit: #7c4dbd;     /* audit events (auditd) */
+  --src-ls: #1f9aa3;        /* LogStream internal logs */
+  --src-file: #5f8a1f;      /* log files (Settings > Sources > Log files) */
+  --src-tint: 6%;           /* background tint of their rows */
 
   --sev-crit: #be123c;
   --sev-err: #dc2626;
@@ -70,6 +74,10 @@
   --hit: rgba(129, 140, 248, .28);
   --shadow: 0 10px 30px rgba(0, 0, 0, .5);
   --chip-l: 70%;
+  --src-audit: #a983e6;
+  --src-ls: #3cc3cc;
+  --src-file: #9ccc54;
+  --src-tint: 9%;
   --sev-crit: #fb7185;
   --sev-err: #f87171;
   --sev-warning: #fbbf24;
@@ -102,6 +110,11 @@
     --hit: rgba(129, 140, 248, .28);
     --shadow: 0 10px 30px rgba(0, 0, 0, .5);
   --chip-l: 70%;
+    --src-audit: #a983e6;
+    --src-ls: #3cc3cc;
+    --src-file: #9ccc54;
+  --src-file: #9ccc54;
+    --src-tint: 9%;
     --sev-crit: #fb7185;
     --sev-err: #f87171;
     --sev-warning: #fbbf24;
@@ -180,6 +193,16 @@ svg { fill: none; stroke: currentColor; stroke-width: 2; stroke-linecap: round;
 .chip.on { color: var(--accent); border-color: var(--accent); background: var(--accent-soft); }
 
 .actions { display: flex; align-items: center; gap: 6px; margin-left: auto; }
+/* Logged-in user, with how they logged in (oidc / local) */
+.whoami { display: flex; align-items: center; gap: 6px; min-width: 0; margin-left: 6px; font-size: 13px; color: var(--muted); }
+.whoami .who-name { font-weight: 600; color: var(--text); max-width: 18ch; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
+.whoami .who-badge {
+  flex: none; border: 1px solid var(--border); border-radius: 6px;
+  padding: 1px 6px; font-size: 10.5px; font-weight: 650; letter-spacing: .02em;
+}
+.whoami .who-badge.oidc { color: var(--accent); border-color: var(--accent); background: var(--accent-soft); }
+.whoami .who-badge:empty { display: none; }
+.whoami.none .who-name { font-weight: 400; color: var(--muted); }
 .icon-btn {
   width: 36px; height: 36px; display: grid; place-items: center;
   border: 1px solid transparent; background: transparent; border-radius: 9px; color: var(--muted);
@@ -401,6 +424,26 @@ body.col-resizing, body.col-resizing * { cursor: col-resize !important; user-sel
 .row.sev-crit, .row.sev-err, .row.sev-warning { border-left-color: var(--sev); }
 .row.sev-crit, .row.sev-err { background: color-mix(in srgb, var(--sev) 7%, transparent); }
 .row .host[data-act], .row .app[data-act] { cursor: pointer; }
+/* Audit events and LogStream internal logs: colored edge (over the severity
+   one), light tint, badge before the app name. */
+.row.src-audit { --src: var(--src-audit); }
+.row.src-logstream { --src: var(--src-ls); }
+.row.src-file { --src: var(--src-file); }
+.row.src-audit, .row.src-logstream, .row.src-file { border-left-color: var(--src); background: color-mix(in srgb, var(--src) var(--src-tint), transparent); }
+.row.src-audit:hover, .row.src-logstream:hover, .row.src-file:hover { background: color-mix(in srgb, var(--src) 13%, var(--row-hover)); }
+.src-badge {
+  display: inline-block; margin-right: 5px; padding: 0 4px; border-radius: 4px; vertical-align: 1px;
+  font: 700 9.5px/1.45 var(--sans); letter-spacing: .04em;
+  color: var(--src); background: color-mix(in srgb, var(--src) 16%, transparent);
+  box-shadow: inset 0 0 0 1px color-mix(in srgb, var(--src) 40%, transparent);
+}
+.row .raw {
+  margin-top: 2px; font-size: .85em; line-height: 1.35; color: var(--muted);
+  white-space: pre-wrap; word-break: break-all;
+}
+:root[data-density="compact"] .row:not(.open) .raw { display: none; }
+.src-help { margin-top: 10px !important; }
+.self-level { width: auto; justify-self: start; }
 /* Codes of the tags found in the message: grey badges, fixed size */
 .row .codes { white-space: nowrap; overflow: hidden; }
 .row .codes > span { display: inline-flex; gap: 3px; vertical-align: 1px; }
@@ -500,12 +543,25 @@ dialog.settings {
 }
 dialog.settings .dlg { display: flex; flex-direction: column; height: 100%; padding: 0; }
 dialog.settings .dlg > header { padding: 14px 16px 14px 22px; border-bottom: 1px solid var(--border); }
-.set-body { flex: 1; min-height: 0; display: grid; grid-template-columns: 176px minmax(0, 1fr); }
+.set-body { flex: 1; min-height: 0; display: grid; grid-template-columns: minmax(176px, max-content) minmax(0, 1fr); }
-.set-nav {
+.set-side {
-  display: flex; flex-direction: column; gap: 2px;
+  display: flex; flex-direction: column; min-height: 0; overflow-y: auto;
-  padding: 12px 10px; border-right: 1px solid var(--border);
+  border-right: 1px solid var(--border);
   background: color-mix(in srgb, var(--panel-2) 50%, transparent);
 }
+.set-nav { display: flex; flex-direction: column; gap: 2px; padding: 12px 10px; }
+/* Save / Cancel, under the section labels: Save takes the accent color while changes wait. */
+.set-actions {
+  display: flex; flex-wrap: wrap; gap: 6px; margin: 4px 10px 12px; padding-top: 12px;
+  border-top: 1px solid var(--border);
+}
+.set-actions .btn { flex: 1; justify-content: center; }
+.set-actions .btn.save { color: var(--muted); }
+.set-actions .btn svg { width: 16px; height: 16px; }
+.set-actions.dirty .btn.save { background: var(--accent); border-color: var(--accent); color: #fff; box-shadow: 0 0 0 3px var(--accent-soft); }
+.set-actions.dirty .btn.save:hover { filter: brightness(1.08); }
+.set-state { flex-basis: 100%; margin: 2px 2px 0; font-size: 12px; color: var(--muted); }
+.set-actions.dirty .set-state { color: var(--accent); font-weight: 550; }
 .set-nav button {
   display: flex; align-items: center; gap: 10px;
   height: 38px; padding: 0 12px; border: 0; border-radius: 9px;
@@ -657,7 +713,7 @@ input.switch:checked::after { transform: translateX(14px); }
 input.switch:disabled { cursor: not-allowed; }
 input.switch:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; }
 
-/* Settings > Filters: toolbar kept in view while the list scrolls, dense rows. */
+/* Settings > Pattern: toolbar kept in view while the list scrolls, dense rows. */
 .tags-section { padding-top: 12px !important; }
 .tag-toolbar {
   position: sticky; top: -16px; z-index: 2;
@@ -713,15 +769,45 @@ input.switch:focus-visible { outline: 2px solid var(--accent); outline-offset: 2
 .tag-row .icon-btn svg { width: 15px; height: 15px; }
 .opt input { display: none; }
 .opt:has(input:checked) { color: var(--accent); border-color: var(--accent); background: var(--accent-soft); }
+/* Log files (Settings > Sources) */
+.file-list { display: flex; flex-direction: column; gap: 4px; }
+.file-row {
+  display: grid; align-items: center; gap: 6px;
+  grid-template-columns: auto minmax(7rem, 1fr) minmax(10rem, 2fr) minmax(9rem, 1.4fr) minmax(6rem, .8fr) 26px;
+  padding: 6px 6px 6px 10px; border: 1px solid var(--border); border-radius: 8px; background: var(--panel);
+}
+.file-row.off > input[type="text"], .file-row.off > select { opacity: .6; }
+.file-row input[type="text"] {
+  height: 28px; padding: 0 8px; min-width: 0;
+  border: 1px solid var(--border); border-radius: 6px; background: var(--panel); font-size: 12.5px; outline: 0;
+}
+.file-row input.mono { font-family: var(--mono); }
+.file-row input[type="text"]:focus { border-color: var(--accent); box-shadow: 0 0 0 3px var(--accent-soft); }
+.file-row select.field { height: 28px; }
+.file-row .icon-btn { width: 26px; height: 26px; border-radius: 6px; }
+.file-row .icon-btn svg { width: 15px; height: 15px; }
+.file-row .fr-regex, .file-row .fr-status, .file-row .fr-sample { grid-column: 2 / -1; }
+.file-row .fr-sample {
+  margin: 0; font: 11.5px var(--mono); color: var(--muted);
+  white-space: nowrap; overflow: hidden; text-overflow: ellipsis;
+}
+.file-row .fr-sample[hidden] { display: none; }
+.file-row .fr-regex[hidden] { display: none; }
+.file-row .fr-status { margin: 0 !important; font-size: 12px; }
+.file-toolbar { margin-top: 10px; }
 .tag-row .err { grid-column: 1 / -1; color: var(--sev-err); font-size: 12px; }
 .tag-row .err:empty { display: none; }
 
 /* ---------- Mobile ---------- */
 @media (max-width: 760px) {
+  .file-row { grid-template-columns: auto minmax(0, 1fr) 26px; }
+  .file-row [data-del] { grid-column: 3; grid-row: 1; }
+  .file-row > [data-f="path"], .file-row > [data-f="format"], .file-row > [data-f="host"] { grid-column: 2 / -1; }
   body { font-size: 13.5px; }
   .topbar { flex-wrap: wrap; padding: 8px 16px; gap: 8px 10px; }
   .search { order: 3; flex-basis: 100%; max-width: none; }
   .live .lbl { display: none; }
+  .whoami .who-name { max-width: 12ch; }
   .filters { padding: 10px 16px 6px; }
   .filters select { flex: 1 1 calc(50% - 8px); max-width: none; min-width: 0; }
   .view-mode { flex: 1 1 100%; }
@@ -758,15 +844,16 @@ input.switch:focus-visible { outline: 2px solid var(--accent); outline-offset: 2
   .field-grid { grid-template-columns: 1fr; gap: 4px; }
   dialog.settings { height: calc(100vh - 24px); width: calc(100vw - 16px); }
   .set-body { grid-template-columns: 1fr; grid-template-rows: auto minmax(0, 1fr); }
-  .set-nav {
+  .set-side { border-right: 0; border-bottom: 1px solid var(--border); overflow: visible; }
-    display: grid; grid-template-columns: repeat(5, minmax(0, 1fr)); padding: 6px;
+  .set-nav { display: grid; grid-template-columns: repeat(5, minmax(0, 1fr)); padding: 6px; }
-    border-right: 0; border-bottom: 1px solid var(--border);
+  .set-actions { margin: 0 6px 6px; padding-top: 6px; align-items: center; }
-  }
+  .set-actions .btn { flex: 0 0 auto; height: 30px; }
+  .set-state { flex: 1; flex-basis: auto; margin: 0; text-align: right; }
   .set-nav button {
     flex-direction: column; justify-content: center; height: auto; padding: 6px 2px; gap: 3px;
     font-size: 11.5px; text-align: center;
   }
-  .set-nav button span { max-width: 100%; overflow: hidden; text-overflow: ellipsis; }
+  .set-nav button span { max-width: 100%; white-space: normal; line-height: 1.15; overflow: hidden; text-overflow: ellipsis; }
   .set-panels { padding: 12px 12px 18px; }
   .set-panel { grid-template-columns: minmax(0, 1fr); gap: 10px; }
   .set-panel > .set-section, .set-panel > .set-section + .set-section { padding: 12px; }