Linux audit (auditd) source and LogStream internal logs
Audit source (source_type=audit), off until enabled in Settings > Sources: - follows /var/log/audit/audit.log (path configurable, AUDIT_LOG), with rotation (end of audit.log.1 read first) and an offset checkpoint in /data/audit-state.json; last AUDIT_BACKFILL (1h) read on first enable - groups the records of an event by msg=audit(time:serial), closed on EOE, after 2 s without a record, or once the log has moved 2 s past it - fields audit_type, audit_serial, auid, user (enriched name, host /etc/passwd or uid), exe, res, key, audit_raw; app = audit_type, host = node= / AUDIT_HOST / /host/etc/hostname; summary "who: what [key] → result"; warning when res=failed / success=no, info otherwise - hexadecimal values (proctitle, cmd, exec arguments…) decoded Internal logs (source_type=logstream, host=logstream, app=component): - logEvent() for VictoriaLogs unreachable / back, lost batches, spooled batches refused, syslog listener errors, Docker API and container streams lost, failing sources, start and stop - on by default from warning; switch and minimum level in Settings > Sources (/data/selflog.json) - bounded queue, never waits for the Store, same event at most once per 30 s with a count of the repeats: a failing Store cannot loop UI: violet (audit) and turquoise (LogStream) left edge and tint, AUDIT / LS badge before the app, raw audit record under the summary (folded in compact density until the row is opened), Audit and LogStream in the source filter, FR/EN translations. docker-compose: optional /var/log/audit, /etc/passwd and /etc/hostname mounts. README (EN/FR): host prerequisites (log_group, ACL for UID 10001). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
f3373e1f66
commit
076c01aac4
20 files changed
+1948
-36
No files matched your search
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user