metrics

pkg/actor/disk_storage.go

@@ -56,9 +56,57 @@ func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[
 		writerIdleTTL: defaultWriterIdleTTL,
 	}
 	go s.purgeLoop()
+	go s.filesExistingLoop()
 	return s
 }
 
+// filesExistingLoop refreshes the EventLogFilesExisting gauge every
+// 30 seconds. The gauge represents the count of .events.log files on
+// disk, which diverges from the open-writer count whenever the idle
+// purge closes a writer but the file is still kept (the .log file is
+// not deleted by the idle purge). Periodic re-listing is cheap (one
+// os.ReadDir per tick) and is the only way the gauge can pick up files
+// created or removed out of band (e.g. GDPR purge, or the init-time
+// rebuild of a grain that already has a log).
+func (s *DiskStorage[V]) filesExistingLoop() {
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-s.done:
+			return
+		case <-ticker.C:
+			s.refreshFilesExisting()
+		}
+	}
+}
+
+// refreshFilesExisting counts .events.log files in the storage dir
+// and writes the value to the EventLogFilesExisting gauge. Safe to
+// call concurrently — os.ReadDir is atomic for our purposes (the dir
+// is owned by this pod, mutations go through the writersMu lock).
+// No-op when the dir does not exist yet (cold start) or when metrics
+// are disabled (s.StateStorage.metrics == nil).
+func (s *DiskStorage[V]) refreshFilesExisting() {
+	if s.StateStorage.metrics == nil {
+		return
+	}
+	entries, err := os.ReadDir(s.path)
+	if err != nil {
+		// Dir may not exist yet on a cold start, or briefly during a
+		// retention-driven directory recreation. Leave the gauge
+		// untouched in that case — the next tick will catch up.
+		return
+	}
+	count := 0
+	for _, e := range entries {
+		if !e.IsDir() && strings.HasSuffix(e.Name(), ".events.log") {
+			count++
+		}
+	}
+	s.StateStorage.metrics.EventLogFilesExisting.Set(float64(count))
+}
+
 func (s *DiskStorage[V]) ensureDir() error {
 	s.dirOnce.Do(func() {
 		s.dirErr = os.MkdirAll(s.path, 0o755)
@@ -70,6 +118,27 @@ func (s *DiskStorage[V]) ensureDir() error {
 // when finished writing.
 
 
+// SetMetrics attaches a Metrics instance to the embedded StateStorage
+// so every append / replay / unknown-type path has access to the
+// same event-log counters. The metrics struct is shared by reference;
+// do not mutate its fields from outside. Pass nil to disable
+// instrumentation (used by tests that do not care about Prometheus).
+func (s *DiskStorage[V]) SetMetrics(m *Metrics) {
+	s.StateStorage.metrics = m
+}
+
+func (s *DiskStorage[V]) recordAppend(n int) {
+	m := s.StateStorage.metrics
+	if m == nil {
+		return
+	}
+	m.EventLogAppends.Inc()
+	if n > 0 {
+		m.EventLogBytesWritten.Add(float64(n))
+	}
+	m.EventLogLastAppendUnix.Set(float64(time.Now().Unix()))
+}
+
 func (s *DiskStorage[V]) writeEvents(id uint64, events []QueueEvent) error {
 	if len(events) == 0 {
 		return nil
@@ -81,9 +150,15 @@ func (s *DiskStorage[V]) writeEvents(id uint64, events []QueueEvent) error {
 	defer w.mu.Unlock()
 
 	for _, evt := range events {
-		if err := s.Append(w.file, evt.Message, evt.TimeStamp); err != nil {
+		n, err := s.Append(w.file, evt.Message, evt.TimeStamp)
+		if err != nil {
+			// A partial write still counts the bytes that DID land on
+			// disk, so the throughput gauge is honest about a mid-write
+			// failure.
+			s.recordAppend(n)
 			return err
 		}
+		s.recordAppend(n)
 	}
 	w.lastUsed = time.Now()
 	return nil
@@ -101,9 +176,12 @@ func (s *DiskStorage[V]) writeMutations(id uint64, msgs ...proto.Message) error
 
 	now := time.Now()
 	for _, m := range msgs {
-		if err := s.Append(w.file, m, now); err != nil {
+		n, err := s.Append(w.file, m, now)
+		if err != nil {
+			s.recordAppend(n)
 			return err
 		}
+		s.recordAppend(n)
 	}
 	w.lastUsed = time.Now()
 	return nil
@@ -232,6 +310,43 @@ func (s *DiskStorage[V]) logPath(id uint64) string {
 	return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id))
 }
 
+// loadReplayMetrics times the replay, records duration, and tracks
+// handler errors. It is shared by LoadEvents and LoadEventsFunc. The
+// returned `onMessage` closure bumps EventLogMutationErrors on any
+// non-nil error from registry.Apply, and the deferred function bumps
+// EventLogReplayTotal / Failures / Duration on the overall return.
+func (s *DiskStorage[V]) loadReplayMetrics() (onMessage func(err error), done func(err error)) {
+	m := s.StateStorage.metrics
+	if m == nil {
+		noop := func(error) {}
+		return noop, noop
+	}
+	start := time.Now()
+	onMessage = func(err error) {
+		if err != nil {
+			m.EventLogMutationErrors.Inc()
+		}
+	}
+	done = func(err error) {
+		if err != nil {
+			m.EventLogReplayFailures.Inc()
+			return
+		}
+		m.EventLogReplayTotal.Inc()
+		m.EventLogReplayDuration.Observe(time.Since(start).Seconds())
+	}
+	return onMessage, done
+}
+
+// recordReplayFailure bumps EventLogReplayFailures when the open
+// step fails before the Load loop runs (and therefore before the
+// loadReplayMetrics closure is in scope). Nil-safe.
+func (s *DiskStorage[V]) recordReplayFailure() {
+	if m := s.StateStorage.metrics; m != nil {
+		m.EventLogReplayFailures.Inc()
+	}
+}
+
 func (s *DiskStorage[V]) LoadEventsFunc(ctx context.Context, id uint64, grain Grain[V], condition func(msg proto.Message, index int, timeStamp time.Time) bool) error {
 	path := s.logPath(id)
 	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
@@ -241,16 +356,25 @@ func (s *DiskStorage[V]) LoadEventsFunc(ctx context.Context, id uint64, grain Gr
 
 	fh, err := os.Open(path)
 	if err != nil {
+		s.recordReplayFailure()
 		return fmt.Errorf("open replay file: %w", err)
 	}
 	defer fh.Close()
+	trackApply, done := s.loadReplayMetrics()
 	index := 0
-	return s.Load(fh, func(msg proto.Message, when time.Time) {
+	loadErr := s.Load(fh, func(msg proto.Message, when time.Time) {
 		if condition(msg, index, when) {
-			s.registry.Apply(ctx, grain, msg)
+			results, _ := s.registry.Apply(ctx, grain, msg)
+			for _, r := range results {
+				if r.Error != nil {
+					trackApply(r.Error)
+				}
+			}
 		}
 		index++
 	})
+	done(loadErr)
+	return loadErr
 }
 
 func (s *DiskStorage[V]) LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error {
@@ -262,12 +386,21 @@ func (s *DiskStorage[V]) LoadEvents(ctx context.Context, id uint64, grain Grain[
 
 	fh, err := os.Open(path)
 	if err != nil {
+		s.recordReplayFailure()
 		return fmt.Errorf("open replay file: %w", err)
 	}
 	defer fh.Close()
-	return s.Load(fh, func(msg proto.Message, _ time.Time) {
-		s.registry.Apply(ctx, grain, msg)
+	trackApply, done := s.loadReplayMetrics()
+	loadErr := s.Load(fh, func(msg proto.Message, _ time.Time) {
+		results, _ := s.registry.Apply(ctx, grain, msg)
+		for _, r := range results {
+			if r.Error != nil {
+				trackApply(r.Error)
+			}
+		}
 	})
+	done(loadErr)
+	return loadErr
 }
 
 func (s *DiskStorage[V]) Close() {