fix(ingestor): rate-limited stats tmp errors, FIFO-safe writer, stale status (#160, #161)

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)
 			}
 		}
 	}()

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. "<tmp> is not a regular file (mode …); remove it: open <tmp>: …" in
+// a "[stats-file] write <path>: …" 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)
+	}
+}