macos: notable-events collector health stats

Author: guohdd

cmd/system-probe/modules/notable_events_darwin.go

@@ -34,6 +34,7 @@ type notableEventsCollector interface {
 	Close() error
 	Pending() []notableevents.Event
 	Ack([]string) error
+	Stats() notableevents.CollectorStats
 }
 
 // newNotableEventsCollector provides the platform collector constructor and a test seam.
@@ -157,9 +158,34 @@ func writeNotableEventsJSON(req *http.Request, w http.ResponseWriter, status int
 	utils.WriteAsJSON(req, w, body, utils.CompactOutput)
 }
 
-// GetStats returns the module's currently empty runtime statistics.
+// GetStats returns aggregate collector health for the system-probe stats
+// endpoint, which feeds `agent status` and flares. Every value is a scalar:
+// report directories are user-specific, so no path or per-directory breakdown
+// may appear here.
 func (m *notableEventsModule) GetStats() map[string]interface{} {
-	return map[string]interface{}{}
+	stats := m.collector.Stats()
+	return map[string]interface{}{
+		"pending_events":                       stats.PendingEvents,
+		"pending_events_max":                   stats.PendingEventsMax,
+		"tracked_files":                        stats.TrackedFiles,
+		"tracked_files_max":                    stats.TrackedFilesMax,
+		"tracked_directories":                  stats.TrackedDirectories,
+		"tracked_directories_max":              stats.TrackedDirectoriesMax,
+		"saturated_directories":                stats.SaturatedDirectories,
+		"retry_directories":                    stats.RetryDirectories,
+		"acknowledged_identities":              stats.AcknowledgedIdentities,
+		"acknowledged_identities_max":          stats.AcknowledgedIdentitiesMax,
+		"bookmark_unsaved":                     stats.BookmarkUnsaved,
+		"bookmark_stage_pending":               stats.BookmarkStagePending,
+		"watcher_active":                       stats.WatcherActive,
+		"persistence_errors":                   stats.PersistenceErrors,
+		"capacity_deferrals":                   stats.CapacityDeferrals,
+		"baseline_suppressed_first_run":        stats.BaselineSuppressedFirstRun,
+		"baseline_suppressed_after_saturation": stats.BaselineSuppressedAfterSaturation,
+		"fsevents_drops":                       stats.FSEventsDrops,
+		"watcher_errors":                       stats.WatcherErrors,
+		"watcher_restarts":                     stats.WatcherRestarts,
+	}
 }
 
 // Close releases the collector and its filesystem monitoring resources.

cmd/system-probe/modules/notable_events_darwin_test.go

@@ -9,6 +9,7 @@ package modules
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"net/http"
 	"net/http/httptest"
@@ -26,6 +27,7 @@ import (
 type fakeNotableEventsCollector struct {
 	events       []notableevents.Event
 	acked        []string
+	stats        notableevents.CollectorStats
 	startErr     error
 	closeErr     error
 	ackErr       error
@@ -58,6 +60,11 @@ func (f *fakeNotableEventsCollector) Ack(ids []string) error {
 	return f.ackErr
 }
 
+// Stats returns the configured health snapshot.
+func (f *fakeNotableEventsCollector) Stats() notableevents.CollectorStats {
+	return f.stats
+}
+
 // notableEventsTestHandler constructs the module routes around a supplied fake collector.
 func notableEventsTestHandler(t *testing.T, collector notableEventsCollector) http.Handler {
 	t.Helper()
@@ -202,6 +209,60 @@ func TestNotableEventsAckFailure(t *testing.T) {
 	assert.JSONEq(t, `{"error":"failed to acknowledge notable events"}`, recorder.Body.String())
 }
 
+// TestNotableEventsGetStatsExposesCollectorHealth verifies every collector
+// health field reaches the stats endpoint under a stable key, and that the
+// payload stays flat and JSON-serializable for `agent status` and flares.
+func TestNotableEventsGetStatsExposesCollectorHealth(t *testing.T) {
+	collector := &fakeNotableEventsCollector{stats: notableevents.CollectorStats{
+		PendingEvents:                     3,
+		PendingEventsMax:                  128,
+		TrackedFiles:                      7,
+		TrackedFilesMax:                   2048,
+		TrackedDirectories:                2,
+		TrackedDirectoriesMax:             256,
+		SaturatedDirectories:              1,
+		RetryDirectories:                  4,
+		AcknowledgedIdentities:            9,
+		AcknowledgedIdentitiesMax:         4096,
+		BookmarkUnsaved:                   true,
+		BookmarkStagePending:              true,
+		WatcherActive:                     true,
+		PersistenceErrors:                 5,
+		CapacityDeferrals:                 6,
+		BaselineSuppressedFirstRun:        11,
+		BaselineSuppressedAfterSaturation: 12,
+		FSEventsDrops:                     13,
+		WatcherErrors:                     14,
+		WatcherRestarts:                   15,
+	}}
+
+	encoded, err := json.Marshal((&notableEventsModule{collector: collector}).GetStats())
+
+	require.NoError(t, err)
+	assert.JSONEq(t, `{
+		"pending_events":3,
+		"pending_events_max":128,
+		"tracked_files":7,
+		"tracked_files_max":2048,
+		"tracked_directories":2,
+		"tracked_directories_max":256,
+		"saturated_directories":1,
+		"retry_directories":4,
+		"acknowledged_identities":9,
+		"acknowledged_identities_max":4096,
+		"bookmark_unsaved":true,
+		"bookmark_stage_pending":true,
+		"watcher_active":true,
+		"persistence_errors":5,
+		"capacity_deferrals":6,
+		"baseline_suppressed_first_run":11,
+		"baseline_suppressed_after_saturation":12,
+		"fsevents_drops":13,
+		"watcher_errors":14,
+		"watcher_restarts":15
+	}`, string(encoded))
+}
+
 // TestNotableEventsRoutesAreMethodSpecific verifies endpoints reject unsupported HTTP methods.
 func TestNotableEventsRoutesAreMethodSpecific(t *testing.T) {
 	handler := notableEventsTestHandler(t, &fakeNotableEventsCollector{})

pkg/notableevents/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "collector_darwin.go",
         "format_darwin.go",
         "reader_darwin.go",
+        "stats_darwin.go",
         "store_darwin.go",
         "watcher_darwin.go",
         "watcher_fsevents_darwin.c",

pkg/notableevents/collector_darwin.go

@@ -122,9 +122,12 @@ type Collector struct {
 	store             darwinBookmarkStore
 	createWatcher     darwinReportWatcherFactory
 	watcher           darwinReportWatcher
-	knownDirs         map[string]reportDirectory
-	retryDirs         map[string]reportDirectory
-	stagedRuntime     *stagedDarwinScanRuntime
+	// watcherEverAttached distinguishes the initial attach from later ones so
+	// only the latter count as restarts.
+	watcherEverAttached bool
+	knownDirs           map[string]reportDirectory
+	retryDirs           map[string]reportDirectory
+	stagedRuntime       *stagedDarwinScanRuntime
 
 	state             *darwinBookmarkState
 	stagedScan        *stagedDarwinScan
@@ -137,6 +140,8 @@ type Collector struct {
 	identityRetention time.Duration
 	maxAcknowledged   int
 
+	counters collectorCounters
+
 	started   bool
 	closed    bool
 	cancel    context.CancelFunc
@@ -286,6 +291,45 @@ func (c *Collector) Pending() []Event {
 	return events
 }
 
+// Stats returns an aggregate snapshot of collector health. It never acquires
+// scanMu, which a scan holds across report I/O, so it stays responsive while
+// scanning. Gauges owned by scanMu are read from their atomic mirrors.
+func (c *Collector) Stats() CollectorStats {
+	stats := CollectorStats{
+		PendingEventsMax:                  maxDarwinPendingEvents,
+		TrackedFilesMax:                   maxDarwinTotalFiles,
+		TrackedDirectoriesMax:             maxDarwinDirectories,
+		AcknowledgedIdentitiesMax:         c.maxAcknowledged,
+		RetryDirectories:                  int(c.counters.retryDirectories.Load()),
+		WatcherActive:                     c.counters.watcherActive.Load(),
+		PersistenceErrors:                 c.counters.persistenceErrors.Load(),
+		CapacityDeferrals:                 c.counters.capacityDeferrals.Load(),
+		BaselineSuppressedFirstRun:        c.counters.baselineSuppressedFirstRun.Load(),
+		BaselineSuppressedAfterSaturation: c.counters.baselineSuppressedAfterSaturation.Load(),
+		FSEventsDrops:                     c.counters.fseventsDrops.Load(),
+		WatcherErrors:                     c.counters.watcherErrors.Load(),
+		WatcherRestarts:                   c.counters.watcherRestarts.Load(),
+	}
+
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+	stats.PendingEvents = len(c.state.Pending)
+	stats.AcknowledgedIdentities = len(c.state.Acknowledged)
+	stats.TrackedDirectories = len(c.state.Directories)
+	stats.BookmarkUnsaved = c.unsaved
+	stats.BookmarkStagePending = c.stagedScan != nil
+	for _, dirState := range c.state.Directories {
+		if dirState == nil {
+			continue
+		}
+		stats.TrackedFiles += len(dirState.Files)
+		if dirState.Saturated {
+			stats.SaturatedDirectories++
+		}
+	}
+	return stats
+}
+
 // Ack atomically moves pending IDs to retained acknowledgement state. Unknown
 // and already acknowledged IDs are no-ops. If persistence fails, no in-memory
 // event is removed and callers can safely retry.
@@ -337,6 +381,7 @@ func (c *Collector) Ack(ids []string) error {
 	c.stateMu.Unlock()
 
 	if err := c.store.Save(next); err != nil {
+		c.counters.persistenceErrors.Add(1)
 		c.stateMu.Lock()
 		c.releaseCommitLocked(reservation)
 		c.stateMu.Unlock()
@@ -412,11 +457,23 @@ func (c *Collector) run(ctx context.Context) {
 				c.scanMu.Unlock()
 				continue
 			}
+			c.recordWatcherError(err)
 			log.Warnf("macOS DiagnosticReports watcher error: %v", err)
 		}
 	}
 }
 
+// recordWatcherError separates dropped-event notifications, which imply lost
+// change notifications, from other asynchronous watcher failures.
+func (c *Collector) recordWatcherError(err error) {
+	var dropped *fseventsDroppedError
+	if errors.As(err, &dropped) {
+		c.counters.fseventsDrops.Add(1)
+		return
+	}
+	c.counters.watcherErrors.Add(1)
+}
+
 // watcherChannelsLocked returns nil-safe watcher channels while scanMu is held.
 func (c *Collector) watcherChannelsLocked() (<-chan string, <-chan error) {
 	if c.watcher == nil {
@@ -429,6 +486,7 @@ func (c *Collector) watcherChannelsLocked() (<-chan string, <-chan error) {
 func (c *Collector) closeWatcherLocked() error {
 	watcher := c.watcher
 	c.watcher = nil
+	c.counters.watcherActive.Store(false)
 	if watcher == nil {
 		return nil
 	}
@@ -447,6 +505,11 @@ type directoryScanResult struct {
 	BaselineIncidentIDs map[string]struct{}
 	BaselineCompletions []darwinBaselineReportCompletion
 	Deliverables        map[string]darwinScanDeliverable
+	// BaselineSuppressedFirstRun and BaselineSuppressedAfterSaturation count
+	// reports recorded without delivery, split by why the directory was being
+	// baselined.
+	BaselineSuppressedFirstRun        int
+	BaselineSuppressedAfterSaturation int
 }
 
 type darwinBaselineReportCompletion struct {
@@ -469,6 +532,9 @@ type darwinReportSource struct {
 type darwinScanIncidents struct {
 	baselineIDs  map[string]struct{}
 	deliverables map[string]darwinScanDeliverable
+	// capacityDeferrals counts events reconcile could not queue because
+	// pending delivery was full.
+	capacityDeferrals int
 }
 
 // scanOnce refreshes report directories and reconciles their current contents.
@@ -503,6 +569,11 @@ func (c *Collector) ensureWatcherLocked() {
 		return
 	}
 	c.watcher = watcher
+	c.counters.watcherActive.Store(true)
+	if c.watcherEverAttached {
+		c.counters.watcherRestarts.Add(1)
+	}
+	c.watcherEverAttached = true
 }
 
 // processWatcherEvent coalesces queued changes and scans only affected known directories.
@@ -588,9 +659,10 @@ func (c *Collector) retryDirectoriesLocked() []reportDirectory {
 // restoreWatcherLocked recreates watcher coverage after an asynchronous watcher failure.
 func (c *Collector) restoreWatcherLocked() {
 	c.ensureWatcherLocked()
-	if c.watcher != nil {
-		c.refreshWatchedDirectoriesLocked(c.knownDirectoriesLocked())
+	if c.watcher == nil {
+		return
 	}
+	c.refreshWatchedDirectoriesLocked(c.knownDirectoriesLocked())
 }
 
 // knownDirectoriesLocked returns a stable snapshot of currently discovered report directories.
@@ -637,6 +709,8 @@ func (c *Collector) scanDirectoriesLocked(ctx context.Context, dirs []reportDire
 
 		result, err := c.scanDirectory(ctx, dir, candidate)
 		dirty = dirty || result.StateChanged
+		addUint64(&c.counters.baselineSuppressedFirstRun, result.BaselineSuppressedFirstRun)
+		addUint64(&c.counters.baselineSuppressedAfterSaturation, result.BaselineSuppressedAfterSaturation)
 		retryResults[directoryRuntimeKey(dir.path)] = result.ShouldRetry
 		results = append(results, result)
 		incidents.addDirectoryResult(result)
@@ -652,6 +726,7 @@ func (c *Collector) scanDirectoriesLocked(ctx context.Context, dirs []reportDire
 	if incidents.reconcile(candidate, retryResults, c.currentTime()) {
 		dirty = true
 	}
+	addUint64(&c.counters.capacityDeferrals, incidents.capacityDeferrals)
 	for index := range results {
 		result := &results[index]
 		dirState := candidate.Directories[result.DirectoryKey]
@@ -737,6 +812,7 @@ func (i *darwinScanIncidents) reconcile(state *darwinBookmarkState, retryResults
 			continue
 		}
 		if len(state.Pending) >= maxDarwinPendingEvents {
+			i.capacityDeferrals++
 			for key := range deliverable.ContributingDirKeys {
 				retryResults[key] = true
 			}
@@ -814,6 +890,7 @@ func (c *Collector) persistCandidateLocked(
 	c.stateMu.Unlock()
 
 	if err := c.store.Save(cloneDarwinBookmarkState(staged.candidate)); err != nil {
+		c.counters.persistenceErrors.Add(1)
 		c.stagedRuntime = runtime
 		c.stateMu.Lock()
 		c.stagedScan = staged
@@ -851,6 +928,7 @@ func (c *Collector) persistStagedScanLocked() ([]reportDirectory, bool) {
 	c.stateMu.Unlock()
 
 	if err := c.store.Save(cloneDarwinBookmarkState(staged.candidate)); err != nil {
+		c.counters.persistenceErrors.Add(1)
 		c.stateMu.Lock()
 		c.releaseCommitLocked(reservation)
 		c.stateMu.Unlock()
@@ -966,6 +1044,7 @@ func (c *Collector) refreshKnownDirectoriesLocked(dirs []reportDirectory) {
 			delete(c.retryDirs, key)
 		}
 	}
+	c.counters.retryDirectories.Store(int64(len(c.retryDirs)))
 }
 
 // refreshWatchedDirectoriesLocked synchronizes FSEvents coverage with watchable directories.
@@ -994,6 +1073,7 @@ func (c *Collector) setDirectoryRetryLocked(dir reportDirectory, shouldRetry boo
 	} else {
 		delete(c.retryDirs, key)
 	}
+	c.counters.retryDirectories.Store(int64(len(c.retryDirs)))
 }
 
 // scanDirectoryInternal securely reads reports and updates only the private
@@ -1060,6 +1140,10 @@ func (c *Collector) scanDirectoryInternal(ctx context.Context, dir reportDirecto
 	}
 
 	baselineScan := !dirState.Initialized || dirState.Saturated
+	// A directory that never completed a baseline has never delivered anything,
+	// so attribute its suppressions to first run even when it is also
+	// saturated. Saturation only costs real events once delivery was working.
+	saturationBaseline := dirState.Initialized && dirState.Saturated
 	presentFiles := make(map[string]struct{})
 
 	for _, entry := range entries {
@@ -1126,6 +1210,15 @@ func (c *Collector) scanDirectoryInternal(ctx context.Context, dir reportDirecto
 		}
 
 		if baselineScan || suppressOnSuccess {
+			// Only a baseline scan counts. The suppressOnSuccess-only path
+			// completes a report that an earlier baseline already counted.
+			if baselineScan {
+				if saturationBaseline {
+					result.BaselineSuppressedAfterSaturation++
+				} else {
+					result.BaselineSuppressedFirstRun++
+				}
+			}
 			if suppressOnSuccess && !baselineScan {
 				result.BaselineCompletions = append(result.BaselineCompletions, darwinBaselineReportCompletion{
 					Name:  name,