diff --git a/.env.example b/.env.example index 41d51c9..b5d5437 100644 --- a/.env.example +++ b/.env.example @@ -53,3 +53,10 @@ 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 diff --git a/README.fr.md b/README.fr.md index f08db38..2b28650 100644 --- a/README.fr.md +++ b/README.fr.md @@ -203,6 +203,43 @@ configurer sur l'hôte. La source est **désactivée par défaut** : activez-la 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 > 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 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. + ## Export CSV Le bouton **Exporter** (à côté du nombre de logs) télécharge tous les logs stockés qui @@ -396,6 +433,8 @@ 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 » | | `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) | @@ -456,6 +495,7 @@ Pour mettre à jour l'une d'elles : | `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 | | `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 | | `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/*` | diff --git a/README.md b/README.md index 96a7c52..42d2309 100644 --- a/README.md +++ b/README.md @@ -187,6 +187,39 @@ configuring anything on the host. The source is **off by default**: turn it on i 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 > 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 > 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. + ## CSV export The **Export** button (next to the log count) downloads every stored log matching the @@ -363,6 +396,8 @@ 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 | | `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) | @@ -421,6 +456,7 @@ To update one of them: | `docker.go` | Docker container logs (API, followers, positions, level detection) | | `syslogserver.go` | syslog listeners opened and closed from Settings > 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 | | `tags.go` | color tag storage | | `presets.go`, `presets.json` | color tag presets (`/api/presets`), built-in list | | `api.go` | `/api/*` HTTP routes | diff --git a/api.go b/api.go index e034298..6cd6303 100644 --- a/api.go +++ b/api.go @@ -23,6 +23,7 @@ type API struct { docker *DockerManager // nil when DOCKER_LOGS is off syslog *SyslogServer host *HostLogs + dmesg *Dmesg dbstats dbStatsCache } @@ -49,6 +50,28 @@ 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) +} + +// 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). diff --git a/dmesg.go b/dmesg.go new file mode 100644 index 0000000..63ec261 --- /dev/null +++ b/dmesg.go @@ -0,0 +1,482 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "io/fs" + "log" + "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 { + log.Printf("dmesg: 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 != "" { + log.Printf("dmesg: %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, + } +} diff --git a/dmesg_test.go b/dmesg_test.go new file mode 100644 index 0000000..b90a22a --- /dev/null +++ b/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) + } +} diff --git a/docker-compose.yml b/docker-compose.yml index d4d9089..857464a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -37,8 +37,19 @@ 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) 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) diff --git a/main.go b/main.go index 571368f..f87ad33 100644 --- a/main.go +++ b/main.go @@ -38,6 +38,8 @@ type config struct { backfill time.Duration hostRoot string hostFill time.Duration + kmsgDevice string + kmsgFill time.Duration batchSize int queueSize int flushEvery time.Duration @@ -115,6 +117,8 @@ 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), batchSize: getenvInt("BATCH_SIZE", 1000), queueSize: getenvInt("QUEUE_SIZE", 100000), flushEvery: time.Duration(getenvInt("FLUSH_MS", 1000)) * time.Millisecond, @@ -186,6 +190,9 @@ 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) api.Routes(mux) mux.Handle("GET /", http.FileServer(http.FS(static))) diff --git a/query.go b/query.go index 457d65f..aea8147 100644 --- a/query.go +++ b/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": 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"))`) } if f.Severity >= 0 && f.Severity < 7 { names := make([]string, 0, 8) diff --git a/syslog.go b/syslog.go index e63e151..2145ebd 100644 --- a/syslog.go +++ b/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" or "kernel" Extra map[string]string // additional fields (docker: container, image, …) // Not stored. Wait: the producer can be slowed down when the queue is full diff --git a/web/app.js b/web/app.js index 2280f6e..ec61169 100644 --- a/web/app.js +++ b/web/app.js @@ -77,6 +77,7 @@ const I18N = { clickHost: 'Click to filter on this host', clickApp: 'Click to filter on this app', sourceAria: 'Source', srcAll: 'All sources', tabSources: 'Sources', srcHost: 'Host system', + srcKernel: 'Kernel (dmesg)', hostTitle: 'Host system logs', hostEnabled: 'Collect the system logs of this machine', hostOff: 'Off: the system logs of the machine hosting LogStream are not collected.', @@ -89,6 +90,15 @@ const I18N = { host_empty: 'No systemd journal and no log file found in /var/log.', host_read: 'Cannot read the host logs: ', hostHelp: 'Reads the systemd journal of the host (/var/log/journal, /run/log/journal), or the text files of /var/log (syslog, messages, *.log) when there is no journal. Directories are mounted read-only. When turned on, the last hour is read first (HOST_LOGS_BACKFILL).', + dmesgTitle: 'Kernel messages (dmesg)', + dmesgEnabled: 'Collect the kernel messages of this machine', + dmesgOff: 'Off: the kernel messages of the machine hosting LogStream are not collected.', + dmesgOk: ({ d, r }) => `Reading ${d}: ${r} messages since startup.`, + dmesgLost: (n) => ` ${n} messages were overwritten in the kernel buffer before being read.`, + dmesg_not_found: 'Device /dev/kmsg not found in the container: uncomment its devices section in docker-compose.yml (see the README).', + dmesg_permission: 'Permission denied on /dev/kmsg: set kernel.dmesg_restrict=0 on the host, or run the container as root with the SYSLOG capability (see the README). ', + dmesg_read: 'Cannot read the kernel messages: ', + dmesgHelp: 'Reads the kernel ring buffer of the host through /dev/kmsg, like dmesg: hardware, drivers, OOM killer, firewall… When turned on, the last hour is read first (DMESG_BACKFILL); after a restart, reading resumes where it stopped.', syslogEnabled: 'Receive syslog messages', syslogProtocols: 'Protocols', syslogPort: 'Port', syslogListening: ({ port, protos }) => `Listening on port ${port} (${protos}).`, syslogOff: 'Syslog reception is off: syslog messages are ignored (Docker logs are still collected).', @@ -251,6 +261,7 @@ const I18N = { clickHost: 'Cliquer pour filtrer sur cet hôte', clickApp: 'Cliquer pour filtrer sur cette appli', sourceAria: 'Source', srcAll: 'Toutes les sources', tabSources: 'Sources', srcHost: 'Système hôte', + srcKernel: 'Noyau (dmesg)', hostTitle: 'Logs système de l\'hôte', hostEnabled: 'Collecter les logs système de cette machine', hostOff: 'Désactivé : les logs système de la machine qui héberge LogStream ne sont pas collectés.', @@ -263,6 +274,15 @@ const I18N = { host_empty: 'Ni journal systemd ni fichier de log trouvé dans /var/log.', host_read: 'Lecture des logs de l\'hôte impossible : ', hostHelp: 'Lit le journal systemd de l\'hôte (/var/log/journal, /run/log/journal), ou les fichiers texte de /var/log (syslog, messages, *.log) s\'il n\'y a pas de journal. Les répertoires sont montés en lecture seule. À l\'activation, la dernière heure est lue d\'abord (HOST_LOGS_BACKFILL).', + dmesgTitle: 'Messages du noyau (dmesg)', + dmesgEnabled: 'Collecter les messages du noyau de cette machine', + dmesgOff: 'Désactivé : les messages du noyau de la machine qui héberge LogStream ne sont pas collectés.', + dmesgOk: ({ d, r }) => `Lecture de ${d} : ${r} messages depuis le démarrage.`, + dmesgLost: (n) => ` ${n} messages ont été écrasés dans le buffer du noyau avant d'être lus.`, + dmesg_not_found: 'Périphérique /dev/kmsg absent du conteneur : décommentez sa section devices dans docker-compose.yml (voir le README).', + dmesg_permission: 'Accès refusé à /dev/kmsg : réglez kernel.dmesg_restrict=0 sur l\'hôte, ou lancez le conteneur en root avec la capacité SYSLOG (voir le README). ', + dmesg_read: 'Lecture des messages du noyau impossible : ', + dmesgHelp: 'Lit le buffer circulaire du noyau de l\'hôte via /dev/kmsg, comme dmesg : matériel, pilotes, OOM killer, pare-feu… À l\'activation, la dernière heure est lue d\'abord (DMESG_BACKFILL) ; après un redémarrage, la lecture reprend où elle s\'était arrêtée.', syslogEnabled: 'Recevoir les messages syslog', syslogProtocols: 'Protocoles', syslogPort: 'Port', syslogListening: ({ port, protos }) => `Écoute sur le port ${port} (${protos}).`, syslogOff: 'Réception syslog désactivée : les messages syslog sont ignorés (les logs Docker sont toujours collectés).', @@ -543,6 +563,7 @@ function setLang(next) { renderInterface(); renderSyslog(); renderHost(); + renderDmesg(); renderDocker(); } @@ -958,7 +979,7 @@ function updateCount(tookMs) { } function detailsHTML(r) { - const order = ['received', 'msg_time', '_time', 'host', 'host_name', 'host_ip', 'source_type', 'container', 'image', 'compose_project', 'compose_service', 'stream', 'unit', 'log_file', 'app', 'procid', 'severity', 'facility', 'source', 'proto', 'sd', '_msg']; + const order = ['received', 'msg_time', '_time', 'host', 'host_name', 'host_ip', 'source_type', 'container', 'image', 'compose_project', 'compose_service', 'stream', 'unit', 'log_file', 'subsystem', 'device', 'app', 'procid', 'severity', 'facility', 'source', 'proto', 'sd', '_msg']; const hidden = new Set(['_stream_id', '_stream', 'sevnum']); if (r.msg_time) hidden.add('_time'); // same as the reception time const keys = order.filter((k) => r[k] != null && r[k] !== '' && !hidden.has(k)) @@ -2000,6 +2021,45 @@ async function configureHost(enabled) { $('#hostEnabled').addEventListener('change', (ev) => configureHost(ev.target.checked)); +/* ================= Kernel messages (dmesg) ================= */ + +let dmesgState = null; + +async function loadDmesg() { + try { dmesgState = await api('/api/dmesg'); } catch { dmesgState = null; } + renderDmesg(); +} + +function renderDmesg() { + const s = dmesgState; + const status = $('#dmesgStatus'); + if (!s) { status.textContent = ''; return; } + $('#dmesgEnabled').checked = s.enabled; + let text = t('hostWaiting'); + let cls = 'muted'; + if (!s.enabled) { + text = t('dmesgOff'); + } else if (s.code) { + text = t('dmesg_' + s.code) + (s.code === 'read' || s.code === 'permission' ? s.error || '' : ''); + cls = 'bad'; + } else if (s.open) { + text = t('dmesgOk', { d: s.device, r: fmtNum(s.read) }) + (s.lost ? t('dmesgLost', fmtNum(s.lost)) : ''); + cls = 'good'; + } + status.textContent = text; + status.className = 'docker-status ' + cls; +} + +async function configureDmesg(enabled) { + dmesgState = { ...(dmesgState || {}), enabled }; + renderDmesg(); + try { dmesgState = await api('/api/dmesg', { method: 'PUT', body: { enabled } }); } catch (e) { toast(e.message); } + renderDmesg(); + setTimeout(loadDmesg, 1500); // the device is opened in the background +} + +$('#dmesgEnabled').addEventListener('change', (ev) => configureDmesg(ev.target.checked)); + /* ================= Docker source ================= */ let dockerState = null; @@ -2116,6 +2176,7 @@ setInterval(() => { if ($('#settingsDlg').open && !document.querySelector('[data-panel="sources"]').hidden) { loadDocker(); loadHost(); + loadDmesg(); } }, 4000); @@ -2409,6 +2470,7 @@ $('#settingsBtn').addEventListener('click', () => { loadPurgeStatus(); loadSyslog(); loadHost(); + loadDmesg(); loadDocker(); renderDbStats(); loadDbStats(); diff --git a/web/index.html b/web/index.html index e95218c..7960fe0 100644 --- a/web/index.html +++ b/web/index.html @@ -102,6 +102,7 @@ + @@ -265,6 +266,16 @@
+