diff --git a/cmd/ingestor/stats_file.go b/cmd/ingestor/stats_file.go index 06a7878de..86cdbfec9 100644 --- a/cmd/ingestor/stats_file.go +++ b/cmd/ingestor/stats_file.go @@ -4,9 +4,13 @@ import ( "bufio" "bytes" "encoding/json" + "errors" "fmt" + "io/fs" "log" "os" + "path/filepath" + "strings" "sync" "time" @@ -137,63 +141,216 @@ 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. 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 { - return err + e := newStatsWriteError(tmp, "open", err) + if fi, lerr := os.Lstat(tmp); lerr == nil { + if !fi.Mode().IsRegular() { + 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 e + } + fi, err := f.Stat() + if err != nil { + f.Close() + 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() + 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 // 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 } 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 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. 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() { - return fmt.Errorf("%s belongs to uid %d, not %d", f.Name(), uid, statsFileEUID()) + 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 } +// 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): 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 // 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++ + l.failing = true + // 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 + } + // 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 (%d more failed writes since the last report)", what, l.suppressed) + } else { + l.logf("[stats-file] write %s", what) + } + l.reported, l.lastLog, l.suppressed = true, now, 0 +} + +func (l *statsWriteLog) succeeded(path string) { + if !l.failing { + return + } + 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 // across two consecutive ticks of the stats-file writer. type procIOSnapshot struct { @@ -267,7 +424,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 +458,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 +508,11 @@ 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) + // 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) } } }() 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_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_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 new file mode 100644 index 000000000..0dbc637d6 --- /dev/null +++ b/cmd/ingestor/stats_file_tmp_160_test.go @@ -0,0 +1,299 @@ +package main + +import ( + "bytes" + "errors" + "fmt" + "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") + } + // 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) + } +} + +// 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 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) != 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")) + } +} + +// 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) + } +} + +// 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")) + } +} 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/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"]) + } +} 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 }