From 2892c60960e82b3127888f7075522bba70a39687 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 08:24:21 +0000 Subject: [PATCH 01/10] test(ingestor,server): reproduce stats tmp FIFO hang, log spam, silent staleness (#160, #161) Red on master: - a FIFO at .tmp blocks writeStatsAtomic and the writer's stop; - a FIFO with a reader is not refused with a clear error; - a broken tmp path logs a failure on every tick and no recovery line; - the foreign-owner error carries no hint; - /api/mqtt/status has no stale/sampleAgeSec marking. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file_fifo_161_unix_test.go | 114 +++++++++++++++ cmd/ingestor/stats_file_tmp_160_test.go | 130 ++++++++++++++++++ cmd/server/mqtt_status_stale_160_test.go | 85 ++++++++++++ 3 files changed, 329 insertions(+) create mode 100644 cmd/ingestor/stats_file_fifo_161_unix_test.go create mode 100644 cmd/ingestor/stats_file_tmp_160_test.go create mode 100644 cmd/server/mqtt_status_stale_160_test.go diff --git a/cmd/ingestor/stats_file_fifo_161_unix_test.go b/cmd/ingestor/stats_file_fifo_161_unix_test.go new file mode 100644 index 000000000..1fb7f8154 --- /dev/null +++ b/cmd/ingestor/stats_file_fifo_161_unix_test.go @@ -0,0 +1,114 @@ +//go:build !windows + +package main + +import ( + "os" + "path/filepath" + "strings" + "syscall" + "testing" + "time" +) + +// mkStatsTmpFIFO plants a named pipe at path+".tmp". +func mkStatsTmpFIFO(t *testing.T, path string) string { + t.Helper() + tmp := path + ".tmp" + if err := syscall.Mkfifo(tmp, 0o600); err != nil { + t.Skipf("mkfifo: %v", err) + } + return tmp +} + +// openFIFOReader opens the read end without blocking. A writer blocked in +// open(2) on the FIFO then gets through. +func openFIFOReader(t *testing.T, fifo string) *os.File { + t.Helper() + r, err := os.OpenFile(fifo, os.O_RDONLY|syscall.O_NONBLOCK, 0) + if err != nil { + t.Fatalf("open FIFO reader: %v", err) + } + return r +} + +// writeStatsAtomicOrRelease runs writeStatsAtomic and returns its error. If +// it blocks for 2s, the FIFO gets a reader so the goroutine can end, and the +// test fails. +func writeStatsAtomicOrRelease(t *testing.T, path, fifo string) error { + t.Helper() + errc := make(chan error, 1) + go func() { errc <- writeStatsAtomic(path, []byte(`{}`)) }() + select { + case err := <-errc: + return err + case <-time.After(2 * time.Second): + r := openFIFOReader(t, fifo) + defer r.Close() + select { + case <-errc: + case <-time.After(2 * time.Second): + } + t.Fatal("writeStatsAtomic blocked on a FIFO at the tmp path") + return nil + } +} + +// assertNotRegularRefused checks the error and that nothing was published. +func assertNotRegularRefused(t *testing.T, err error, path string) { + t.Helper() + if err == nil { + t.Fatal("FIFO at the tmp path accepted") + } + if !strings.Contains(err.Error(), "not a regular file") { + t.Errorf("error %q does not say the tmp is not a regular file", err) + } + if _, err := os.Lstat(path); !os.IsNotExist(err) { + t.Errorf("stats file published from a FIFO: lstat err %v", err) + } +} + +// Opening a FIFO for writing waits for a reader. Without one the writer +// must fail at once, not block forever (#161). +func TestWriteStatsAtomicFIFOWithoutReaderFailsFast_161(t *testing.T) { + path := filepath.Join(t.TempDir(), "stats.json") + fifo := mkStatsTmpFIFO(t, path) + assertNotRegularRefused(t, writeStatsAtomicOrRelease(t, path, fifo), path) +} + +// With a reader the open succeeds; the writer must still refuse the FIFO +// rather than write into it and rename it over the stats file (#161). +func TestWriteStatsAtomicFIFOWithReaderRefused_161(t *testing.T) { + path := filepath.Join(t.TempDir(), "stats.json") + fifo := mkStatsTmpFIFO(t, path) + r := openFIFOReader(t, fifo) + defer r.Close() + assertNotRegularRefused(t, writeStatsAtomicOrRelease(t, path, fifo), path) +} + +// The writer's stop waits for its goroutine. A FIFO at the tmp path must +// not keep that goroutine, and so the stop, from returning (#161). +func TestStatsFileWriterStopsWithFIFOAtTmp_161(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "stats.json") + t.Setenv("CORESCOPE_INGESTOR_STATS", path) + fifo := mkStatsTmpFIFO(t, path) + captureStdLog(t) // keep the failure line out of the test output + store := openStatsTestStore(t, dir) + stop := StartStatsFileWriter(store, 5*time.Millisecond) + + time.Sleep(100 * time.Millisecond) // several ticks against the FIFO + stopped := make(chan struct{}) + go func() { stop(); close(stopped) }() + select { + case <-stopped: + case <-time.After(3 * time.Second): + r := openFIFOReader(t, fifo) + defer r.Close() + select { + case <-stopped: + case <-time.After(3 * time.Second): + } + t.Fatal("stop hung: the writer is blocked on the FIFO") + } +} diff --git a/cmd/ingestor/stats_file_tmp_160_test.go b/cmd/ingestor/stats_file_tmp_160_test.go new file mode 100644 index 000000000..b4da9e768 --- /dev/null +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -0,0 +1,130 @@ +package main + +import ( + "bytes" + "log" + "os" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" +) + +// syncLogBuffer collects the standard logger's output; the stats writer +// logs from its own goroutine. +type syncLogBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +// linesWith returns the logged lines that contain sub. +func (b *syncLogBuffer) linesWith(sub string) []string { + b.mu.Lock() + defer b.mu.Unlock() + var out []string + for _, l := range strings.Split(b.buf.String(), "\n") { + if strings.Contains(l, sub) { + out = append(out, l) + } + } + return out +} + +// captureStdLog sends the standard logger to a buffer until the test ends. +// Register it before the writer's stop, so the stop runs first (LIFO). +func captureStdLog(t *testing.T) *syncLogBuffer { + t.Helper() + b := &syncLogBuffer{} + oldOut, oldFlags := log.Writer(), log.Flags() + log.SetOutput(b) + t.Cleanup(func() { log.SetOutput(oldOut); log.SetFlags(oldFlags) }) + return b +} + +// openStatsTestStore opens a Store for a stats-writer test; it is closed +// after the writer's stop, which the caller registers later (LIFO). +func openStatsTestStore(t *testing.T, dir string) *Store { + t.Helper() + store, err := OpenStore(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + t.Cleanup(func() { store.Close() }) + return store +} + +// A tmp path the writer cannot use fails every tick (1 Hz in production). +// The failure is logged once, not once per tick, and the first successful +// write after it is logged once (#160). +func TestStatsFileWriterLogsWriteFailureOnce_160(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "stats.json") + t.Setenv("CORESCOPE_INGESTOR_STATS", path) + // A directory at the tmp path makes every write fail. + if err := os.Mkdir(path+".tmp", 0o700); err != nil { + t.Fatal(err) + } + logs := captureStdLog(t) + store := openStatsTestStore(t, dir) + stop := StartStatsFileWriter(store, 5*time.Millisecond) + t.Cleanup(stop) + + time.Sleep(300 * time.Millisecond) // dozens of failing ticks + if got := logs.linesWith("[stats-file] write "); len(got) != 1 { + t.Fatalf("%d failure lines while the tmp stayed broken, want 1:\n%s", len(got), strings.Join(got, "\n")) + } + + if err := os.Remove(path + ".tmp"); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(3 * time.Second) + for { + if _, err := os.Stat(path); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("no stats file after the tmp path was freed") + } + time.Sleep(10 * time.Millisecond) + } + time.Sleep(100 * time.Millisecond) // more successful ticks + stop() + + if got := logs.linesWith("[stats-file] write "); len(got) != 2 { + t.Fatalf("%d write lines, want the failure and one recovery:\n%s", len(got), strings.Join(got, "\n")) + } + if got := logs.linesWith("ok again"); len(got) != 1 { + t.Fatalf("%d recovery lines, want 1", len(got)) + } +} + +// A tmp file that belongs to another user cannot be fixed by the ingestor; +// the error says what the operator has to do (#160). +func TestWriteStatsAtomicForeignTmpErrorNamesTheFix_160(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("no Unix file owners") + } + old := statsFileEUID + t.Cleanup(func() { statsFileEUID = old }) + statsFileEUID = func() int { return os.Geteuid() + 1 } // the tmp is someone else's + + path := filepath.Join(t.TempDir(), "stats.json") + if err := os.WriteFile(path+".tmp", nil, 0o600); err != nil { + t.Fatal(err) + } + err := writeStatsAtomic(path, []byte(`{}`)) + if err == nil { + t.Fatal("foreign tmp accepted") + } + if want := "remove " + path + ".tmp or fix its owner"; !strings.Contains(err.Error(), want) { + t.Fatalf("error %q lacks the hint %q", err, want) + } +} diff --git a/cmd/server/mqtt_status_stale_160_test.go b/cmd/server/mqtt_status_stale_160_test.go new file mode 100644 index 000000000..91f0a4e99 --- /dev/null +++ b/cmd/server/mqtt_status_stale_160_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" +) + +// mqttStatusFor serves /api/mqtt/status from a stats file holding +// sampledAt (none when sampledAt is nil) and returns the decoded body. +func mqttStatusFor(t *testing.T, sampledAt *string) map[string]any { + t.Helper() + statsPath := filepath.Join(t.TempDir(), "ingestor-stats.json") + t.Setenv("CORESCOPE_INGESTOR_STATS", statsPath) + if sampledAt != nil { + data, err := json.Marshal(map[string]any{"sampledAt": *sampledAt, "source_statuses": []any{}}) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(statsPath, data, 0o600); err != nil { + t.Fatal(err) + } + } + rec := httptest.NewRecorder() + (&Server{}).handleMqttStatus(rec, httptest.NewRequest(http.MethodGet, "/api/mqtt/status", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("status %d; body %s", rec.Code, rec.Body) + } + var body map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("unmarshal: %v; body %s", err, rec.Body) + } + return body +} + +func stamp(ts time.Time) *string { + s := ts.UTC().Format(time.RFC3339) + return &s +} + +// When the ingestor stops updating the stats file (#160: a tmp it cannot +// write), /api/mqtt/status marks the frozen data as stale and gives its +// age. The rule is the one /api/perf/io applies to the same file. +func TestMqttStatusMarksStaleStatsFile_160(t *testing.T) { + old := mqttStatusFor(t, stamp(time.Now().Add(-time.Hour))) + if old["stale"] != true { + t.Errorf("hour-old sample: stale = %v, want true; body %v", old["stale"], old) + } + if age, _ := old["sampleAgeSec"].(float64); age < 3600 || age > 3700 { + t.Errorf("hour-old sample: sampleAgeSec = %v, want ~3600", old["sampleAgeSec"]) + } + + justPast := mqttStatusFor(t, stamp(time.Now().Add(-2*IngestorStatsStaleThreshold))) + if justPast["stale"] != true { + t.Errorf("sample 2x the /api/perf/io threshold old: stale = %v, want true", justPast["stale"]) + } + + fresh := mqttStatusFor(t, stamp(time.Now())) + if fresh["stale"] != false { + t.Errorf("fresh sample: stale = %v, want false; body %v", fresh["stale"], fresh) + } + if age, ok := fresh["sampleAgeSec"].(float64); !ok || age > 2 { + t.Errorf("fresh sample: sampleAgeSec = %v, want ~0", fresh["sampleAgeSec"]) + } + + // A stamp that cannot be read cannot vouch for fresh data. + bad := "not-a-time" + garbled := mqttStatusFor(t, &bad) + if garbled["stale"] != true { + t.Errorf("unparseable sampledAt: stale = %v, want true", garbled["stale"]) + } + if _, ok := garbled["sampleAgeSec"]; ok { + t.Errorf("unparseable sampledAt: sampleAgeSec present: %v", garbled["sampleAgeSec"]) + } + + // No stats file: no data, which the empty sampleAt already says. + none := mqttStatusFor(t, nil) + if none["stale"] != false { + t.Errorf("no stats file: stale = %v, want false", none["stale"]) + } +} From e3789e4c936ade274ba872ad5ab349af7c8ca199 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 09:27:08 +0000 Subject: [PATCH 02/10] fix(ingestor): rate-limited stats tmp errors, FIFO-safe writer, stale status (#160, #161) #161: writeStatsAtomic opens the tmp with O_NONBLOCK, so a FIFO without a reader fails with ENXIO instead of blocking the writer (and its stop, which waits on the goroutine) forever. Anything that is not a regular file is then refused via the fstat the owner check already did, with " is not a regular file (mode ...); remove it". #160: the owner error names the fix ("remove or fix its owner"). statsWriteLog logs the first failure at once, a persisting one at most once a minute with a count, and the first success after it once. A healthy writer still logs nothing. /api/mqtt/status (read-only) gains stale and sampleAgeSec, using the /api/perf/io rule, now shared as ingestorStatsStale. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file.go | 94 ++++++++++++++++++--- cmd/ingestor/stats_file_nofollow_unix.go | 4 + cmd/ingestor/stats_file_nofollow_windows.go | 3 + cmd/ingestor/stats_file_tmp_160_test.go | 42 +++++++++ cmd/server/mqtt_status.go | 22 +++++ cmd/server/openapi.go | 2 +- cmd/server/perf_io.go | 12 ++- 7 files changed, 162 insertions(+), 17 deletions(-) diff --git a/cmd/ingestor/stats_file.go b/cmd/ingestor/stats_file.go index 06a7878de..954655063 100644 --- a/cmd/ingestor/stats_file.go +++ b/cmd/ingestor/stats_file.go @@ -138,18 +138,34 @@ func statsFilePath() string { func writeStatsAtomic(path string, b []byte) error { tmp := path + ".tmp" // O_NOFOLLOW: if tmp is a pre-existing symlink, openat fails with ELOOP - // instead of clobbering the symlink target. 0o600 — no need for - // world-readable. - f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|oNoFollow, 0o600) + // instead of clobbering the symlink target. O_NONBLOCK: a FIFO without + // a reader fails with ENXIO instead of blocking the writer, and its + // stop, forever (#161); regular-file I/O ignores the flag. 0o600 — no + // need for world-readable. + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|oNoFollow|oNonBlock, 0o600) if err != nil { + if fi, lerr := os.Lstat(tmp); lerr == nil && !fi.Mode().IsRegular() { + return fmt.Errorf("%v: %w", errStatsTmpNotRegular(tmp, fi.Mode()), err) + } + return err + } + fi, err := f.Stat() + if err != nil { + f.Close() return err } + // Only a regular file may be published: a FIFO that has a reader, or + // a device, opens fine (#161). + if !fi.Mode().IsRegular() { + f.Close() + return errStatsTmpNotRegular(tmp, fi.Mode()) + } // The mode above applies only to a new file. A stale tmp keeps its // owner and mode, and the rename would publish both. One that belongs // to another user is refused before anything is changed: root's chmod // would succeed on it, and its owner may still hold it open. Ours gets // 0o600 and loses its old content (#118). - if err := checkStatsTmpOwner(f); err != nil { + if err := checkStatsTmpOwner(tmp, fi); err != nil { f.Close() return err } @@ -181,19 +197,65 @@ func writeStatsAtomic(path string, b []byte) error { // tests to model a file owned by someone else. var statsFileEUID = os.Geteuid -// checkStatsTmpOwner fails when f belongs to a user other than -// statsFileEUID. Where files have no Unix owner (Windows) it passes. -func checkStatsTmpOwner(f *os.File) error { - fi, err := f.Stat() - if err != nil { - return err - } +// checkStatsTmpOwner fails when the tmp file (name, fi from its open +// descriptor) belongs to a user other than statsFileEUID. Where files have +// no Unix owner (Windows) it passes. +func checkStatsTmpOwner(name string, fi os.FileInfo) error { if uid, ok := fileOwnerUID(fi); ok && uid != statsFileEUID() { - return fmt.Errorf("%s belongs to uid %d, not %d", f.Name(), uid, statsFileEUID()) + // The ingestor cannot fix this itself; say what the operator must + // do (#160). + return fmt.Errorf("%s belongs to uid %d, not %d; remove %s or fix its owner", name, uid, statsFileEUID(), name) } return nil } +// errStatsTmpNotRegular is the error for a stats tmp path that holds +// something other than a regular file, such as a FIFO or a directory. +// It is left in place for the operator to remove (#161). +func errStatsTmpNotRegular(tmp string, mode os.FileMode) error { + return fmt.Errorf("%s is not a regular file (mode %v); remove it", tmp, mode) +} + +// statsWriteErrLogEvery is how often a failure that persists is logged +// again. The writer ticks every second (#160). +const statsWriteErrLogEvery = time.Minute + +// statsWriteLog rate-limits the writer's failure log (#160): the first +// failure is logged at once, a persisting one at most once per `every` +// with the count of failures in between, and the first success after a +// failure once. Successes otherwise log nothing. Owned by the writer +// goroutine; not safe for concurrent use. +type statsWriteLog struct { + every time.Duration + logf func(format string, args ...any) + failing bool + lastLog time.Time + failures int64 // since the last success + suppressed int64 // since the last failure line +} + +func (l *statsWriteLog) failed(path string, err error, now time.Time) { + l.failures++ + if l.failing && now.Sub(l.lastLog) < l.every { + l.suppressed++ + return + } + if l.suppressed > 0 { + l.logf("[stats-file] write %s: %v (%d more failed writes since the last report)", path, err, l.suppressed) + } else { + l.logf("[stats-file] write %s: %v", path, err) + } + l.failing, l.lastLog, l.suppressed = true, now, 0 +} + +func (l *statsWriteLog) succeeded(path string) { + if !l.failing { + return + } + l.logf("[stats-file] write %s: ok again after %d failed writes", path, l.failures) + l.failing, l.failures, l.suppressed = false, 0, 0 +} + // procIOSnapshot is the raw counter snapshot used to compute per-second rates // across two consecutive ticks of the stats-file writer. type procIOSnapshot struct { @@ -267,7 +329,8 @@ func procIORate(prev, cur procIOSnapshot, stamp string) *PerfIOSample { // StartStatsFileWriter writes the current stats snapshot to disk every // `interval` so the server can serve them at /api/perf/write-sources. -// Failures are logged once-per-interval and never fatal. +// Failures are never fatal; they are logged rate-limited (statsWriteLog, +// #160). // // The stats file path is resolved via statsFilePath() once at writer-loop // start; the env var (CORESCOPE_INGESTOR_STATS) is only re-read on process @@ -300,6 +363,7 @@ func StartStatsFileWriter(s *Store, interval time.Duration) (stop func()) { // The buffer grows once and stays. var buf bytes.Buffer enc := json.NewEncoder(&buf) + writeLog := statsWriteLog{every: statsWriteErrLogEvery, logf: log.Printf} for { select { case <-quit: @@ -349,7 +413,9 @@ func StartStatsFileWriter(s *Store, interval time.Duration) (stop func()) { b = b[:n-1] } if err := writeStatsAtomic(path, b); err != nil { - log.Printf("[stats-file] write %s: %v", path, err) + writeLog.failed(path, err, tickAt) + } else { + writeLog.succeeded(path) } } }() diff --git a/cmd/ingestor/stats_file_nofollow_unix.go b/cmd/ingestor/stats_file_nofollow_unix.go index f55d5f1dd..3561a2f27 100644 --- a/cmd/ingestor/stats_file_nofollow_unix.go +++ b/cmd/ingestor/stats_file_nofollow_unix.go @@ -7,3 +7,7 @@ import "syscall" // oNoFollow is syscall.O_NOFOLLOW on platforms that define it (all non-Windows targets). // On Windows this constant does not exist; see stats_file_nofollow_windows.go. const oNoFollow = syscall.O_NOFOLLOW + +// oNonBlock is syscall.O_NONBLOCK: opening a FIFO for writing without it +// waits for a reader, forever if none comes (#161). +const oNonBlock = syscall.O_NONBLOCK diff --git a/cmd/ingestor/stats_file_nofollow_windows.go b/cmd/ingestor/stats_file_nofollow_windows.go index c26b458c3..6a93a5610 100644 --- a/cmd/ingestor/stats_file_nofollow_windows.go +++ b/cmd/ingestor/stats_file_nofollow_windows.go @@ -6,3 +6,6 @@ package main // package. The ingestor is only deployed on Linux where the flag is enforced; // on Windows the flag is a no-op so the binary compiles and tests run. const oNoFollow = 0 + +// oNonBlock is 0 on Windows, which has no FIFOs that open(2) could block on. +const oNonBlock = 0 diff --git a/cmd/ingestor/stats_file_tmp_160_test.go b/cmd/ingestor/stats_file_tmp_160_test.go index b4da9e768..11b674c88 100644 --- a/cmd/ingestor/stats_file_tmp_160_test.go +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -2,6 +2,8 @@ package main import ( "bytes" + "errors" + "fmt" "log" "os" "path/filepath" @@ -128,3 +130,43 @@ func TestWriteStatsAtomicForeignTmpErrorNamesTheFix_160(t *testing.T) { t.Fatalf("error %q lacks the hint %q", err, want) } } + +// A failure that persists is logged at most once per interval, with the +// failures in between counted; recovery is logged once, and a success +// without a failure before it logs nothing (#160). +func TestStatsWriteLogAtMostOncePerInterval_160(t *testing.T) { + var lines []string + l := statsWriteLog{every: time.Minute, logf: func(f string, a ...any) { + lines = append(lines, fmt.Sprintf(f, a...)) + }} + errTmp := errors.New("x.tmp belongs to uid 1, not 2; remove x.tmp or fix its owner") + t0 := time.Unix(1_700_000_000, 0) + + l.succeeded("x") // healthy start: silent + for s := 0; s < 150; s++ { // 2.5 minutes of 1 Hz failures + l.failed("x", errTmp, t0.Add(time.Duration(s)*time.Second)) + } + want := []string{ + "[stats-file] write x: " + errTmp.Error(), + "[stats-file] write x: " + errTmp.Error() + " (59 more failed writes since the last report)", + "[stats-file] write x: " + errTmp.Error() + " (59 more failed writes since the last report)", + } + if strings.Join(lines, "\n") != strings.Join(want, "\n") { + t.Fatalf("failure lines:\n%s\nwant:\n%s", strings.Join(lines, "\n"), strings.Join(want, "\n")) + } + + lines = nil + l.succeeded("x") + l.succeeded("x") + if len(lines) != 1 || lines[0] != "[stats-file] write x: ok again after 150 failed writes" { + t.Fatalf("recovery lines %q", lines) + } + + // A new failure after recovery is logged at once, counted afresh. + lines = nil + l.failed("x", errTmp, t0.Add(151*time.Second)) + l.succeeded("x") + if len(lines) != 2 || !strings.HasSuffix(lines[1], "ok again after 1 failed writes") { + t.Fatalf("second episode lines %q", lines) + } +} diff --git a/cmd/server/mqtt_status.go b/cmd/server/mqtt_status.go index 3e03a4a02..d3035bc48 100644 --- a/cmd/server/mqtt_status.go +++ b/cmd/server/mqtt_status.go @@ -7,6 +7,7 @@ import ( "sort" "strconv" "strings" + "time" "github.com/meshcore-analyzer/brokerurl" ) @@ -143,6 +144,26 @@ type MqttStatusResponse struct { // 0 / omitted: no drops OR older ingestor build that predates // newAsyncEmit. WatchdogLogDropCount int64 `json:"watchdogLogDropCount,omitempty"` + // Stale (#160) is true when sampleAt is older than + // IngestorStatsStaleThreshold (the rule /api/perf/io applies to the + // same file) or cannot be parsed: the ingestor is down or cannot + // write its stats file, and the rows are frozen. False without a + // stats file; sampleAt is then empty and there are no rows. + Stale bool `json:"stale"` + // SampleAgeSec (#160) is the age of sampleAt in whole seconds when + // the response was built. Omitted when there is no readable sampleAt. + SampleAgeSec *int64 `json:"sampleAgeSec,omitempty"` +} + +// mqttStatusStaleness returns Stale and SampleAgeSec for a sampleAt stamp +// read from the stats file at now. +func mqttStatusStaleness(stamp string, now time.Time) (stale bool, ageSec *int64) { + ts, err := time.Parse(time.RFC3339, stamp) + if err != nil { + return true, nil + } + age := int64(now.Sub(ts) / time.Second) + return ingestorStatsStale(ts, now), &age } // ingestorMqttStatusEnvelope is the partial shape the server decodes from @@ -172,6 +193,7 @@ func (s *Server) handleMqttStatus(w http.ResponseWriter, r *http.Request) { return } resp.SampleAt = env.SampledAt + resp.Stale, resp.SampleAgeSec = mqttStatusStaleness(env.SampledAt, time.Now()) resp.WatchdogLastTickUnix = env.WatchdogLastTickUnix resp.WatchdogPanicCount = env.WatchdogPanicCount resp.WatchdogLogDropCount = env.WatchdogLogDropCount diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 3f6ccb8bf..bd418cf7c 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -47,7 +47,7 @@ func routeDescriptions() map[string]routeMeta { "GET /api/health": {Summary: "Health check", Description: "Returns server health, uptime, and memory stats.", Tag: "admin"}, "GET /api/stats": {Summary: "Network statistics", Description: "Returns aggregate stats (node counts, packet counts, observer counts). Cached for 10s.", Tag: "admin"}, "GET /api/perf": {Summary: "Performance statistics", Description: "Returns per-endpoint request timing and slow query log.", Tag: "admin"}, - "GET /api/mqtt/status": {Summary: "MQTT source status", Description: "Returns per-MQTT-source connection state and counters (lastConnectUnix, lastPacketUnix, packetsTotal, etc.). Broker URL credentials (user-info, query, fragment) are masked in broker and lastError, and in a name that is a raw broker URL. Sourced from the ingestor stats file; empty list when unavailable. (#1043)", Tag: "admin"}, + "GET /api/mqtt/status": {Summary: "MQTT source status", Description: "Returns per-MQTT-source connection state and counters (lastConnectUnix, lastPacketUnix, packetsTotal, etc.). Broker URL credentials (user-info, query, fragment) are masked in broker and lastError, and in a name that is a raw broker URL. Sourced from the ingestor stats file; empty list when unavailable. stale is true when the file's sampleAt is older than the /api/perf/io freshness threshold (5s) or unreadable, i.e. the rows are frozen; sampleAgeSec gives its age. (#1043, #160)", Tag: "admin"}, "POST /api/perf/reset": {Summary: "Reset performance stats", Tag: "admin", Auth: true}, // "POST /api/admin/prune" removed in #1283 (ingestor owns prune). "GET /api/debug/affinity": {Summary: "Debug neighbor affinity scores", Tag: "admin", Auth: true}, diff --git a/cmd/server/perf_io.go b/cmd/server/perf_io.go index d67b31533..1d937615c 100644 --- a/cmd/server/perf_io.go +++ b/cmd/server/perf_io.go @@ -162,6 +162,14 @@ func (s *Server) handlePerfIO(w http.ResponseWriter, r *http.Request) { // #1167 must-fix #1: serving stale procIO as live disguises a dead ingestor. const IngestorStatsStaleThreshold = 5 * time.Second +// ingestorStatsStale reports whether an ingestor stats snapshot sampled at +// ts is too old at now to pass for live data. It is the one staleness rule +// for readers of the stats file: /api/perf/io drops such a sample, and +// /api/mqtt/status marks it stale (#160). +func ingestorStatsStale(ts, now time.Time) bool { + return now.Sub(ts) > IngestorStatsStaleThreshold +} + // ingestorIOPeek is the minimal subset of IngestorStats that // readIngestorIOSample actually needs. Decoding into this instead of the // full IngestorStats avoids allocating BackfillUpdates (a map) and the @@ -201,7 +209,7 @@ func readIngestorIOSample() *PerfIOSample { // file (writer wedged) MUST still drop after the threshold. if s.SampledAt != "" { if ts, err := time.Parse(time.RFC3339, s.SampledAt); err == nil { - if time.Since(ts) > IngestorStatsStaleThreshold { + if ingestorStatsStale(ts, time.Now()) { return nil } } @@ -233,7 +241,7 @@ func readIngestorIOSample() *PerfIOSample { if err != nil { return nil } - if time.Since(ts) > IngestorStatsStaleThreshold { + if ingestorStatsStale(ts, time.Now()) { return nil } From 29c794ed8e5010b8ff0a827237594c2586981396 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:17:59 +0000 Subject: [PATCH 03/10] test(ingestor): the #160 hint is missing when open fails with EACCES (#160) The usual #160 case is a 0600 tmp left by another service user and a non-root ingestor: open(2) fails with EACCES before checkStatsTmpOwner runs, so the log line was the bare "open : permission denied". Pin the hint and the owner uids for a foreign tmp, and a permissions hint for an unopenable tmp of the ingestor's own user. (PR #216 review, F1) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file_tmp_160_test.go | 54 +++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/cmd/ingestor/stats_file_tmp_160_test.go b/cmd/ingestor/stats_file_tmp_160_test.go index 11b674c88..53acc08a9 100644 --- a/cmd/ingestor/stats_file_tmp_160_test.go +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -170,3 +170,57 @@ func TestStatsWriteLogAtMostOncePerInterval_160(t *testing.T) { t.Fatalf("second episode lines %q", lines) } } + +// writeUnopenableStatsTmp plants a regular tmp the ingestor cannot open for +// writing (0400), the way a 0600 tmp of another service user looks to a +// non-root ingestor: open(2) fails with EACCES before any owner check. +func writeUnopenableStatsTmp(t *testing.T) (path string) { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("no Unix file owners") + } + if os.Geteuid() == 0 { + t.Skip("root opens a 0400 file for writing") + } + path = filepath.Join(t.TempDir(), "stats.json") + if err := os.WriteFile(path+".tmp", nil, 0o400); err != nil { + t.Fatal(err) + } + return path +} + +// PR #216 review, F1: the usual #160 case is a foreign 0600 tmp and a non-root +// ingestor, where open itself fails with EACCES. The hint must be there too, +// with the owner the Lstat finds. +func TestWriteStatsAtomicUnopenableForeignTmpNamesTheFix_160(t *testing.T) { + path := writeUnopenableStatsTmp(t) + old := statsFileEUID + t.Cleanup(func() { statsFileEUID = old }) + statsFileEUID = func() int { return os.Geteuid() + 1 } // the tmp is someone else's + + err := writeStatsAtomic(path, []byte(`{}`)) + if err == nil { + t.Fatal("unopenable tmp accepted") + } + if !errors.Is(err, os.ErrPermission) { + t.Errorf("error %q does not wrap the permission error", err) + } + for _, want := range []string{"or fix its owner", fmt.Sprintf("uid %d", os.Geteuid()), fmt.Sprintf("uid %d", os.Geteuid()+1)} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error %q lacks %q", err, want) + } + } +} + +// An unopenable tmp of the ingestor's own user is a mode problem, not an +// owner one; the hint says so. +func TestWriteStatsAtomicUnopenableOwnTmpNamesTheFix_160(t *testing.T) { + path := writeUnopenableStatsTmp(t) + err := writeStatsAtomic(path, []byte(`{}`)) + if err == nil { + t.Fatal("unopenable tmp accepted") + } + if !strings.Contains(err.Error(), "or fix its permissions") || strings.Contains(err.Error(), "owner") { + t.Errorf("error %q: want the permissions hint, not the owner one", err) + } +} From e191d6d6bde3d2029269d59d7b84c45de2db9002 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:18:47 +0000 Subject: [PATCH 04/10] fix(ingestor): give the #160 hint when the stats tmp cannot be opened (#160) When open(2) on the tmp fails with EACCES/EPERM, an Lstat of the tmp chooses the hint: "remove or fix its owner (owned by uid X, ingestor uid Y)" for another user's file, "... or fix its permissions (mode ...)" for the ingestor's own. The permission error stays in the %w chain. The Lstat runs only after a failed open; a healthy tick is unchanged. (PR #216 review, F1) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file.go | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/cmd/ingestor/stats_file.go b/cmd/ingestor/stats_file.go index 954655063..77382fceb 100644 --- a/cmd/ingestor/stats_file.go +++ b/cmd/ingestor/stats_file.go @@ -4,6 +4,7 @@ import ( "bufio" "bytes" "encoding/json" + "errors" "fmt" "log" "os" @@ -144,8 +145,16 @@ func writeStatsAtomic(path string, b []byte) error { // need for world-readable. f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|oNoFollow|oNonBlock, 0o600) if err != nil { - if fi, lerr := os.Lstat(tmp); lerr == nil && !fi.Mode().IsRegular() { - return fmt.Errorf("%v: %w", errStatsTmpNotRegular(tmp, fi.Mode()), err) + if fi, lerr := os.Lstat(tmp); lerr == nil { + if !fi.Mode().IsRegular() { + return fmt.Errorf("%v: %w", errStatsTmpNotRegular(tmp, fi.Mode()), err) + } + // The usual #160 case: a 0600 tmp left by another service + // user, which a non-root ingestor cannot even open, so the + // owner check below never runs. Give the hint here too. + if errors.Is(err, os.ErrPermission) { + return fmt.Errorf("%w; %s", err, statsTmpPermissionHint(tmp, fi)) + } } return err } @@ -209,6 +218,16 @@ func checkStatsTmpOwner(name string, fi os.FileInfo) error { return nil } +// statsTmpPermissionHint says what the operator must do about a regular +// tmp (fi from Lstat) the ingestor may not open: fix the owner when it is +// someone else's, else its permissions (#160). +func statsTmpPermissionHint(name string, fi os.FileInfo) string { + if uid, ok := fileOwnerUID(fi); ok && uid != statsFileEUID() { + return fmt.Sprintf("remove %s or fix its owner (owned by uid %d, ingestor uid %d)", name, uid, statsFileEUID()) + } + return fmt.Sprintf("remove %s or fix its permissions (mode %v)", name, fi.Mode()) +} + // errStatsTmpNotRegular is the error for a stats tmp path that holds // something other than a regular file, such as a FIFO or a directory. // It is left in place for the operator to remove (#161). From b55511ab151b49917de994c6135e305e04e2576d Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:19:33 +0000 Subject: [PATCH 05/10] test(ingestor): a flapping stats write failure must be rate-limited (#160) After a success the next failure was logged at once, so failures that alternate with successes at 1 Hz gave 120 lines in 120 s (a failure line and an "ok again" line each time). Pin at most one failure line and one recovery line per interval, with the failures in between counted, and update the limiter test's second episode to match. (PR #216 review, F2) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file_tmp_160_test.go | 48 +++++++++++++++++++++++-- 1 file changed, 45 insertions(+), 3 deletions(-) diff --git a/cmd/ingestor/stats_file_tmp_160_test.go b/cmd/ingestor/stats_file_tmp_160_test.go index 53acc08a9..c39a2db83 100644 --- a/cmd/ingestor/stats_file_tmp_160_test.go +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -162,12 +162,54 @@ func TestStatsWriteLogAtMostOncePerInterval_160(t *testing.T) { t.Fatalf("recovery lines %q", lines) } - // A new failure after recovery is logged at once, counted afresh. + // A new failure inside the interval of the last failure line (120 s) + // is counted, not logged, and its recovery is not logged either; the + // next failure line, an interval after the last one, reports it. lines = nil l.failed("x", errTmp, t0.Add(151*time.Second)) l.succeeded("x") - if len(lines) != 2 || !strings.HasSuffix(lines[1], "ok again after 1 failed writes") { - t.Fatalf("second episode lines %q", lines) + if len(lines) != 0 { + t.Fatalf("episode inside the interval logged %q", lines) + } + l.failed("x", errTmp, t0.Add(181*time.Second)) + l.succeeded("x") + want = []string{ + "[stats-file] write x: " + errTmp.Error() + " (1 more failed writes since the last report)", + "[stats-file] write x: ok again after 1 failed writes", + } + if strings.Join(lines, "\n") != strings.Join(want, "\n") { + t.Fatalf("second episode lines:\n%s\nwant:\n%s", strings.Join(lines, "\n"), strings.Join(want, "\n")) + } +} + +// PR #216 review, F2: a failure that alternates with successes (flapping) +// is rate-limited too. At 1 Hz, 120 s gave 120 lines (a failure line and an +// "ok again" line every time); now the failure line and the recovery line +// each come at most once per interval, and the failures in between are +// counted. +func TestStatsWriteLogFlappingAtMostOncePerInterval_160(t *testing.T) { + var lines []string + l := statsWriteLog{every: time.Minute, logf: func(f string, a ...any) { + lines = append(lines, fmt.Sprintf(f, a...)) + }} + errTmp := errors.New("x.tmp: broken") + t0 := time.Unix(1_700_000_000, 0) + for s := 0; s < 120; s++ { + at := t0.Add(time.Duration(s) * time.Second) + if s%2 == 0 { + l.failed("x", errTmp, at) + } else { + l.succeeded("x") + } + } + want := []string{ + "[stats-file] write x: " + errTmp.Error(), + "[stats-file] write x: ok again after 1 failed writes", + "[stats-file] write x: " + errTmp.Error() + " (29 more failed writes since the last report)", + "[stats-file] write x: ok again after 1 failed writes", + } + if strings.Join(lines, "\n") != strings.Join(want, "\n") { + t.Fatalf("%d lines:\n%s\nwant:\n%s", len(lines), strings.Join(lines, "\n"), strings.Join(want, "\n")) } } From 1474234bd6d3d1bf8ee98b8e83383db296ccf312 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:20:08 +0000 Subject: [PATCH 06/10] fix(ingestor): rate-limit flapping stats write failures too (#160) statsWriteLog now keeps the last failure line's time across episodes: a failure line comes at most once per interval, whether the failure persists or alternates with successes, and the "ok again" line follows only a logged failure line. A failure episode inside the interval is counted and reported with the next failure line. A logged recovery reports the whole episode and clears that count. Persistent failures log as before (150 s at 1 Hz: 3 lines). (PR #216 review, F2) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file.go | 35 ++++++++++++++++++++++------------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/cmd/ingestor/stats_file.go b/cmd/ingestor/stats_file.go index 77382fceb..06c6f6d52 100644 --- a/cmd/ingestor/stats_file.go +++ b/cmd/ingestor/stats_file.go @@ -239,23 +239,28 @@ func errStatsTmpNotRegular(tmp string, mode os.FileMode) error { // again. The writer ticks every second (#160). const statsWriteErrLogEvery = time.Minute -// statsWriteLog rate-limits the writer's failure log (#160): the first -// failure is logged at once, a persisting one at most once per `every` -// with the count of failures in between, and the first success after a -// failure once. Successes otherwise log nothing. Owned by the writer -// goroutine; not safe for concurrent use. +// statsWriteLog rate-limits the writer's failure log (#160): a failure +// line is logged at most once per `every`, the first one at once, with the +// count of failures since the last failure line. The first success after a +// logged failure line is logged once ("ok again"); a failure episode that +// fell inside the interval, and so was not logged, is not either, which +// keeps a flapping failure to one failure line and one recovery line per +// interval. Successes otherwise log nothing. Owned by the writer goroutine; +// not safe for concurrent use. type statsWriteLog struct { every time.Duration logf func(format string, args ...any) - failing bool - lastLog time.Time - failures int64 // since the last success - suppressed int64 // since the last failure line + failing bool // a failure since the last success + reported bool // and a failure line was logged for it + lastLog time.Time // the last failure line + failures int64 // since the last success + suppressed int64 // not reported since the last failure or recovery line } func (l *statsWriteLog) failed(path string, err error, now time.Time) { l.failures++ - if l.failing && now.Sub(l.lastLog) < l.every { + l.failing = true + if !l.lastLog.IsZero() && now.Sub(l.lastLog) < l.every { l.suppressed++ return } @@ -264,15 +269,19 @@ func (l *statsWriteLog) failed(path string, err error, now time.Time) { } else { l.logf("[stats-file] write %s: %v", path, err) } - l.failing, l.lastLog, l.suppressed = true, now, 0 + l.reported, l.lastLog, l.suppressed = true, now, 0 } func (l *statsWriteLog) succeeded(path string) { if !l.failing { return } - l.logf("[stats-file] write %s: ok again after %d failed writes", path, l.failures) - l.failing, l.failures, l.suppressed = false, 0, 0 + if l.reported { + // The recovery line reports the whole episode. + l.logf("[stats-file] write %s: ok again after %d failed writes", path, l.failures) + l.suppressed = 0 + } + l.failing, l.reported, l.failures = false, false, 0 } // procIOSnapshot is the raw counter snapshot used to compute per-second rates From 050a858d9cc6bbea985903b484468932fdd849b7 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:20:52 +0000 Subject: [PATCH 07/10] test(ingestor): a backward clock step must not silence stats write failures (#160) The writer drove the limiter with time.Now().UTC(), which drops the monotonic reading, and the limiter took a negative difference to the last failure line as "inside the interval": after a 1 h step back a persisting failure was not logged for an hour. Pin a line at the step and then one per interval. (PR #216 review, F3) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file_tmp_160_test.go | 30 +++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/cmd/ingestor/stats_file_tmp_160_test.go b/cmd/ingestor/stats_file_tmp_160_test.go index c39a2db83..f50d372ae 100644 --- a/cmd/ingestor/stats_file_tmp_160_test.go +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -266,3 +266,33 @@ func TestWriteStatsAtomicUnopenableOwnTmpNamesTheFix_160(t *testing.T) { t.Errorf("error %q: want the permissions hint, not the owner one", err) } } + +// PR #216 review, F3: a backward wall-clock step must not silence a +// persisting failure. The limiter compared now with the last failure line, +// and a negative difference counted as "inside the interval", so after a +// 1 h step back nothing was logged for an hour. A step back now counts as +// an interval passed (the writer also passes a monotonic time.Now()). +func TestStatsWriteLogBackwardClockStepStillLogs_160(t *testing.T) { + var lines []string + l := statsWriteLog{every: time.Minute, logf: func(f string, a ...any) { + lines = append(lines, fmt.Sprintf(f, a...)) + }} + errTmp := errors.New("x.tmp: broken") + t0 := time.Unix(1_700_000_000, 0) + for s := 0; s < 10; s++ { + l.failed("x", errTmp, t0.Add(time.Duration(s)*time.Second)) + } + back := t0.Add(-time.Hour) // the wall clock steps back 1 h + for s := 0; s < 180; s++ { + l.failed("x", errTmp, back.Add(time.Duration(s)*time.Second)) + } + want := []string{ + "[stats-file] write x: " + errTmp.Error(), + "[stats-file] write x: " + errTmp.Error() + " (9 more failed writes since the last report)", + "[stats-file] write x: " + errTmp.Error() + " (59 more failed writes since the last report)", + "[stats-file] write x: " + errTmp.Error() + " (59 more failed writes since the last report)", + } + if strings.Join(lines, "\n") != strings.Join(want, "\n") { + t.Fatalf("%d lines:\n%s\nwant:\n%s", len(lines), strings.Join(lines, "\n"), strings.Join(want, "\n")) + } +} From e06847835ef1d0d7148f303f15ba3042eadac1c2 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:21:05 +0000 Subject: [PATCH 08/10] fix(ingestor): monotonic time for the stats write limiter, and survive a step back (#160) The writer passes time.Now() to statsWriteLog.failed instead of the UTC tick time, so the interval runs on the monotonic clock; SampledAt and the source statuses keep the UTC tick time. The limiter also treats a negative difference to the last failure line as an interval passed, so even a wall-clock time cannot silence a persisting failure after a step back. (PR #216 review, F3) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/cmd/ingestor/stats_file.go b/cmd/ingestor/stats_file.go index 06c6f6d52..fe0456f80 100644 --- a/cmd/ingestor/stats_file.go +++ b/cmd/ingestor/stats_file.go @@ -260,7 +260,10 @@ type statsWriteLog struct { func (l *statsWriteLog) failed(path string, err error, now time.Time) { l.failures++ l.failing = true - if !l.lastLog.IsZero() && now.Sub(l.lastLog) < l.every { + // A negative difference is a wall-clock step back, not "inside the + // interval": it counts as an interval passed, so a step cannot + // silence a persisting failure. + if d := now.Sub(l.lastLog); !l.lastLog.IsZero() && d >= 0 && d < l.every { l.suppressed++ return } @@ -441,7 +444,9 @@ func StartStatsFileWriter(s *Store, interval time.Duration) (stop func()) { b = b[:n-1] } if err := writeStatsAtomic(path, b); err != nil { - writeLog.failed(path, err, tickAt) + // time.Now(), not tickAt: UTC() drops the monotonic + // reading the limiter's interval relies on. + writeLog.failed(path, err, time.Now()) } else { writeLog.succeeded(path) } From 90bcd67c0a01ea8455ebff5f81e532402bb7bf9d Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:22:04 +0000 Subject: [PATCH 09/10] test(ingestor): a stats write error names the path once (#160) A failure line named the path up to three times, e.g. "[stats-file] write : is not a regular file (mode ...); remove it: open : ...". Pin that each writeStatsAtomic error (FIFO, directory, foreign owner, unopenable foreign tmp) starts with the tmp path and names it once, and that the writer's failure line adds no second copy. The foreign-owner hint is now "remove it or fix its owner". (PR #216 review, F4) Co-Authored-By: Claude Opus 5.5 --- .../stats_file_errtext_160_unix_test.go | 97 +++++++++++++++++++ cmd/ingestor/stats_file_tmp_160_test.go | 5 +- 2 files changed, 100 insertions(+), 2 deletions(-) create mode 100644 cmd/ingestor/stats_file_errtext_160_unix_test.go diff --git a/cmd/ingestor/stats_file_errtext_160_unix_test.go b/cmd/ingestor/stats_file_errtext_160_unix_test.go new file mode 100644 index 000000000..59f97a5dc --- /dev/null +++ b/cmd/ingestor/stats_file_errtext_160_unix_test.go @@ -0,0 +1,97 @@ +//go:build !windows + +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// PR #216 review, F4: a stats write error named the path up to three times, +// e.g. " is not a regular file (mode …); remove it: open : …" in +// a "[stats-file] write : …" line. Each error now names the tmp once, +// at the start, and the writer's failure line adds no second copy. +func TestStatsWriteErrorNamesThePathOnce_160(t *testing.T) { + type tc struct { + name string + setup func(t *testing.T, path string) + want string + } + foreign := func(t *testing.T) { + old := statsFileEUID + t.Cleanup(func() { statsFileEUID = old }) + statsFileEUID = func() int { return os.Geteuid() + 1 } + } + cases := []tc{ + {"FIFO without a reader", func(t *testing.T, path string) { mkStatsTmpFIFO(t, path) }, "not a regular file"}, + {"directory", func(t *testing.T, path string) { + if err := os.Mkdir(path+".tmp", 0o700); err != nil { + t.Fatal(err) + } + }, "not a regular file"}, + {"foreign owner", func(t *testing.T, path string) { + if err := os.WriteFile(path+".tmp", nil, 0o600); err != nil { + t.Fatal(err) + } + foreign(t) + }, "remove it or fix its owner"}, + {"foreign owner, unopenable", func(t *testing.T, path string) { + if os.Geteuid() == 0 { + t.Skip("root opens a 0400 file for writing") + } + if err := os.WriteFile(path+".tmp", nil, 0o400); err != nil { + t.Fatal(err) + } + foreign(t) + }, "remove it or fix its owner"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "stats.json") + c.setup(t, path) + err := writeStatsAtomic(path, []byte(`{}`)) + if err == nil { + t.Fatal("accepted") + } + msg := err.Error() + if n := strings.Count(msg, dir); n != 1 { + t.Errorf("path named %d times in %q, want once", n, msg) + } + if !strings.HasPrefix(msg, path+".tmp: ") { + t.Errorf("%q does not start with the tmp path", msg) + } + if !strings.Contains(msg, c.want) { + t.Errorf("%q lacks %q", msg, c.want) + } + }) + } +} + +// The writer's failure line names the path once too. +func TestStatsFileWriterFailureLineNamesThePathOnce_160(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "stats.json") + t.Setenv("CORESCOPE_INGESTOR_STATS", path) + mkStatsTmpFIFO(t, path) + logs := captureStdLog(t) + store := openStatsTestStore(t, dir) + stop := StartStatsFileWriter(store, 5*time.Millisecond) + t.Cleanup(stop) + + deadline := time.Now().Add(3 * time.Second) + for len(logs.linesWith("[stats-file] write")) == 0 { + if time.Now().After(deadline) { + t.Fatal("no failure line") + } + time.Sleep(5 * time.Millisecond) + } + stop() + line := logs.linesWith("[stats-file] write")[0] + if n := strings.Count(line, dir); n != 1 { + t.Errorf("path named %d times in %q, want once", n, line) + } +} diff --git a/cmd/ingestor/stats_file_tmp_160_test.go b/cmd/ingestor/stats_file_tmp_160_test.go index f50d372ae..0dbc637d6 100644 --- a/cmd/ingestor/stats_file_tmp_160_test.go +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -126,8 +126,9 @@ func TestWriteStatsAtomicForeignTmpErrorNamesTheFix_160(t *testing.T) { if err == nil { t.Fatal("foreign tmp accepted") } - if want := "remove " + path + ".tmp or fix its owner"; !strings.Contains(err.Error(), want) { - t.Fatalf("error %q lacks the hint %q", err, want) + // The tmp path leads the message (F4), so the hint says "remove it". + if want := "remove it or fix its owner"; !strings.HasPrefix(err.Error(), path+".tmp: ") || !strings.Contains(err.Error(), want) { + t.Fatalf("error %q lacks the tmp path or the hint %q", err, want) } } From 0bc53bc464a7311c131b60db9470f26535421ae7 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:23:09 +0000 Subject: [PATCH 10/10] fix(ingestor): name the stats path once in write errors (#160) writeStatsAtomic now returns a statsWriteError for every failed step. Its message starts with the tmp path and does not repeat it: the failed step with its cause (the path stripped from the *PathError/*LinkError), then what is wrong with the tmp and what to do, e.g. : open: permission denied; owned by uid 1000, ingestor uid 1001; remove it or fix its owner : open: no such device or address; not a regular file (mode prw-------); remove it Unwrap returns the cause, so errors.Is still sees the errno. The writer's line for such an error is "[stats-file] write failed: ", without the path in front; the "ok again" line is unchanged. This replaces errStatsTmpNotRegular and statsTmpPermissionHint. Only the message text changes; what is refused, removed or published is the same. (PR #216 review, F4) Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/stats_file.go | 140 +++++++++++++++++++++++++++---------- 1 file changed, 102 insertions(+), 38 deletions(-) diff --git a/cmd/ingestor/stats_file.go b/cmd/ingestor/stats_file.go index fe0456f80..86cdbfec9 100644 --- a/cmd/ingestor/stats_file.go +++ b/cmd/ingestor/stats_file.go @@ -6,8 +6,11 @@ import ( "encoding/json" "errors" "fmt" + "io/fs" "log" "os" + "path/filepath" + "strings" "sync" "time" @@ -138,6 +141,8 @@ func statsFilePath() string { // fails loudly. func writeStatsAtomic(path string, b []byte) error { tmp := path + ".tmp" + // fail wraps a failed step: the error names the tmp once (#160). + fail := func(op string, err error) error { return newStatsWriteError(tmp, op, err) } // O_NOFOLLOW: if tmp is a pre-existing symlink, openat fails with ELOOP // instead of clobbering the symlink target. O_NONBLOCK: a FIFO without // a reader fails with ENXIO instead of blocking the writer, and its @@ -145,29 +150,32 @@ func writeStatsAtomic(path string, b []byte) error { // need for world-readable. f, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|oNoFollow|oNonBlock, 0o600) if err != nil { + e := newStatsWriteError(tmp, "open", err) if fi, lerr := os.Lstat(tmp); lerr == nil { if !fi.Mode().IsRegular() { - return fmt.Errorf("%v: %w", errStatsTmpNotRegular(tmp, fi.Mode()), err) - } - // The usual #160 case: a 0600 tmp left by another service - // user, which a non-root ingestor cannot even open, so the - // owner check below never runs. Give the hint here too. - if errors.Is(err, os.ErrPermission) { - return fmt.Errorf("%w; %s", err, statsTmpPermissionHint(tmp, fi)) + e.setNotRegular(fi.Mode()) + } else if errors.Is(err, os.ErrPermission) { + // The usual #160 case: a 0600 tmp left by another + // service user, which a non-root ingestor cannot even + // open, so the owner check below never runs. Give the + // hint here too. + e.setPermissionHint(fi) } } - return err + return e } fi, err := f.Stat() if err != nil { f.Close() - return err + return fail("stat", err) } // Only a regular file may be published: a FIFO that has a reader, or // a device, opens fine (#161). if !fi.Mode().IsRegular() { f.Close() - return errStatsTmpNotRegular(tmp, fi.Mode()) + e := newStatsWriteError(tmp, "", nil) + e.setNotRegular(fi.Mode()) + return e } // The mode above applies only to a new file. A stale tmp keeps its // owner and mode, and the rename would publish both. One that belongs @@ -180,61 +188,110 @@ func writeStatsAtomic(path string, b []byte) error { } if err := f.Chmod(0o600); err != nil { f.Close() - return err + return fail("chmod", err) } if err := f.Truncate(0); err != nil { f.Close() - return err + return fail("truncate", err) } if _, err := f.Write(b); err != nil { f.Close() os.Remove(tmp) - return err + return fail("write", err) } if err := f.Close(); err != nil { os.Remove(tmp) - return err + return fail("close", err) } if err := os.Rename(tmp, path); err != nil { os.Remove(tmp) - return err + return fail("rename to "+filepath.Base(path), err) } return nil } +// statsWriteError is a failed stats write (#160). Its message names the +// tmp path once, at the start, then the failed step with its cause, what +// is wrong with the tmp, and what the operator must do, e.g. +// +// /tmp/x.json.tmp: open: permission denied; owned by uid 1001, ingestor uid 1000; remove it or fix its owner +// +// Unwrap gives the cause (an errno where there is one), so errors.Is +// works as on the bare error. +type statsWriteError struct { + tmp string + op string // the failed step, "" for a refusal after a check + err error // the step's cause, without the path it names + detail string // what is wrong with the tmp + hint string // what the operator must do +} + +func newStatsWriteError(tmp, op string, err error) *statsWriteError { + var pe *fs.PathError + var le *os.LinkError + switch { + case errors.As(err, &pe): + err = pe.Err + case errors.As(err, &le): + err = le.Err + } + return &statsWriteError{tmp: tmp, op: op, err: err} +} + +func (e *statsWriteError) Error() string { + parts := make([]string, 0, 3) + if e.op != "" && e.err != nil { + parts = append(parts, e.op+": "+e.err.Error()) + } + for _, s := range []string{e.detail, e.hint} { + if s != "" { + parts = append(parts, s) + } + } + return e.tmp + ": " + strings.Join(parts, "; ") +} + +func (e *statsWriteError) Unwrap() error { return e.err } + +// setNotRegular marks a tmp path that holds something other than a +// regular file, such as a FIFO or a directory. It is left in place for the +// operator to remove (#161). +func (e *statsWriteError) setNotRegular(mode os.FileMode) { + e.detail = fmt.Sprintf("not a regular file (mode %v)", mode) + e.hint = "remove it" +} + +// setPermissionHint says what the operator must do about a regular tmp +// (fi from Lstat) the ingestor may not open: fix the owner when it is +// someone else's, else its permissions (#160). +func (e *statsWriteError) setPermissionHint(fi os.FileInfo) { + if uid, ok := fileOwnerUID(fi); ok && uid != statsFileEUID() { + e.detail = fmt.Sprintf("owned by uid %d, ingestor uid %d", uid, statsFileEUID()) + e.hint = "remove it or fix its owner" + return + } + e.detail = fmt.Sprintf("mode %v", fi.Mode()) + e.hint = "remove it or fix its permissions" +} + // statsFileEUID is the user the stats tmp file must belong to; swapped in // tests to model a file owned by someone else. var statsFileEUID = os.Geteuid // checkStatsTmpOwner fails when the tmp file (name, fi from its open // descriptor) belongs to a user other than statsFileEUID. Where files have -// no Unix owner (Windows) it passes. +// no Unix owner (Windows) it passes. The ingestor cannot fix this itself; +// the error says what the operator must do (#160). func checkStatsTmpOwner(name string, fi os.FileInfo) error { if uid, ok := fileOwnerUID(fi); ok && uid != statsFileEUID() { - // The ingestor cannot fix this itself; say what the operator must - // do (#160). - return fmt.Errorf("%s belongs to uid %d, not %d; remove %s or fix its owner", name, uid, statsFileEUID(), name) + e := newStatsWriteError(name, "", nil) + e.detail = fmt.Sprintf("owned by uid %d, ingestor uid %d", uid, statsFileEUID()) + e.hint = "remove it or fix its owner" + return e } return nil } -// statsTmpPermissionHint says what the operator must do about a regular -// tmp (fi from Lstat) the ingestor may not open: fix the owner when it is -// someone else's, else its permissions (#160). -func statsTmpPermissionHint(name string, fi os.FileInfo) string { - if uid, ok := fileOwnerUID(fi); ok && uid != statsFileEUID() { - return fmt.Sprintf("remove %s or fix its owner (owned by uid %d, ingestor uid %d)", name, uid, statsFileEUID()) - } - return fmt.Sprintf("remove %s or fix its permissions (mode %v)", name, fi.Mode()) -} - -// errStatsTmpNotRegular is the error for a stats tmp path that holds -// something other than a regular file, such as a FIFO or a directory. -// It is left in place for the operator to remove (#161). -func errStatsTmpNotRegular(tmp string, mode os.FileMode) error { - return fmt.Errorf("%s is not a regular file (mode %v); remove it", tmp, mode) -} - // statsWriteErrLogEvery is how often a failure that persists is logged // again. The writer ticks every second (#160). const statsWriteErrLogEvery = time.Minute @@ -267,10 +324,17 @@ func (l *statsWriteLog) failed(path string, err error, now time.Time) { l.suppressed++ return } + // A statsWriteError names the tmp, so the line does not name the + // path again (#160). + what := path + ": " + err.Error() + var we *statsWriteError + if errors.As(err, &we) { + what = "failed: " + err.Error() + } if l.suppressed > 0 { - l.logf("[stats-file] write %s: %v (%d more failed writes since the last report)", path, err, l.suppressed) + l.logf("[stats-file] write %s (%d more failed writes since the last report)", what, l.suppressed) } else { - l.logf("[stats-file] write %s: %v", path, err) + l.logf("[stats-file] write %s", what) } l.reported, l.lastLog, l.suppressed = true, now, 0 }