From 2e2060da5cb8e1fa325efea48d91954917ed092e Mon Sep 17 00:00:00 2001 From: matst80 Date: Thu, 2 Jul 2026 10:37:36 +0200 Subject: [PATCH] metrics --- cmd/cart/main.go | 14 +-- cmd/cart/pool-server.go | 23 ++-- cmd/checkout/main.go | 14 +-- cmd/checkout/pool-server.go | 19 ++- cmd/checkout/utils.go | 1 - cmd/profile/main.go | 9 ++ grafana_dashboard_cart.json | 2 +- pkg/actor/disk_storage.go | 145 +++++++++++++++++++++- pkg/actor/metrics.go | 216 +++++++++++++++++++++++++++++++++ pkg/actor/simple_grain_pool.go | 67 ++++++++-- pkg/actor/state.go | 36 ++++-- 11 files changed, 481 insertions(+), 65 deletions(-) create mode 100644 pkg/actor/metrics.go diff --git a/cmd/cart/main.go b/cmd/cart/main.go index 63846ca..4761a35 100644 --- a/cmd/cart/main.go +++ b/cmd/cart/main.go @@ -27,18 +27,17 @@ import ( "git.k6n.net/mats/platform/config" "git.k6n.net/mats/platform/event" "git.k6n.net/mats/platform/rabbit" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/redis/go-redis/v9" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" ) var ( - grainSpawns = promauto.NewCounter(prometheus.CounterOpts{ - Name: "cart_grain_spawned_total", - Help: "The total number of spawned grains", - }) + // cartMetrics is the shared prometheus instrumentation for the + // cart actor pool and event log. Built once at process start; nil + // is not expected in production but the actor package's + // touch sites are nil-safe so a test binary can pass nil. + cartMetrics = actor.NewMetrics("cart") ) func init() { @@ -258,13 +257,14 @@ func main() { ) diskStorage := actor.NewDiskStorage[cart.CartGrain](config.EnvString("CART_DIR", "data"), reg) + diskStorage.SetMetrics(cartMetrics) poolConfig := actor.GrainPoolConfig[cart.CartGrain]{ MutationRegistry: reg, Storage: diskStorage, + Metrics: cartMetrics, Spawn: func(ctx context.Context, id uint64) (actor.Grain[cart.CartGrain], error) { _, span := tracer.Start(ctx, fmt.Sprintf("Spawn cart id %d", id)) defer span.End() - grainSpawns.Inc() ret := cart.NewCartGrain(id, time.Now()) // Set baseline lastChange at spawn; replay may update it to last event timestamp. diff --git a/cmd/cart/pool-server.go b/cmd/cart/pool-server.go index e194096..b554333 100644 --- a/cmd/cart/pool-server.go +++ b/cmd/cart/pool-server.go @@ -16,8 +16,6 @@ import ( messages "git.k6n.net/mats/go-cart-actor/proto/cart" "git.k6n.net/mats/go-cart-actor/pkg/voucher" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" @@ -28,16 +26,13 @@ import ( "go.opentelemetry.io/otel/trace" ) -var ( - grainMutations = promauto.NewCounter(prometheus.CounterOpts{ - Name: "cart_grain_mutations_total", - Help: "The total number of mutations", - }) - grainLookups = promauto.NewCounter(prometheus.CounterOpts{ - Name: "cart_grain_lookups_total", - Help: "The total number of lookups", - }) -) +// Pool metrics (cart_grain_spawned_total, cart_grain_lookups_total, +// cart_mutations_total, cart_remote_negotiation_total, +// cart_connected_remotes, …) are bumped centrally by SimpleGrainPool +// — see pkg/actor/simple_grain_pool.go. The cart Metrics is +// registered once in cmd/cart/main.go and shared with DiskStorage +// via SetMetrics, so the per-handler counters this file used to +// maintain are no longer needed. type PoolServer struct { actor.GrainPool[cart.CartGrain] @@ -122,7 +117,6 @@ func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, if err != nil { return err } - grainMutations.Add(float64(len(data.Mutations))) return s.WriteResult(w, data) } @@ -496,9 +490,8 @@ func (s *PoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http.Request hostAttr := attribute.String("other host", ownerHost.Name()) span.SetAttributes(hostAttr) proxyCalls.Add(ctx, 1, metric.WithAttributes(hostAttr)) - handled, err := ownerHost.Proxy(uint64(cartId), w, r, nil) + handled, err := ownerHost.Proxy(uint64(cartId), w, r, nil) - grainLookups.Inc() if err == nil && handled { return nil } diff --git a/cmd/checkout/main.go b/cmd/checkout/main.go index d4c74af..2c52129 100644 --- a/cmd/checkout/main.go +++ b/cmd/checkout/main.go @@ -21,17 +21,16 @@ import ( "git.k6n.net/mats/platform/tax" "github.com/adyen/adyen-go-api-library/v21/src/adyen" "github.com/adyen/adyen-go-api-library/v21/src/common" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" "github.com/redis/go-redis/v9" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" ) var ( - grainSpawns = promauto.NewCounter(prometheus.CounterOpts{ - Name: "checkout_grain_spawned_total", - Help: "The total number of spawned checkout grains", - }) + // checkoutMetrics is the shared prometheus instrumentation for the + // checkout actor pool and event log. Built once at process start; + // the actor package's touch sites are nil-safe so a test binary + // could pass nil. + checkoutMetrics = actor.NewMetrics("checkout") ) func init() { @@ -114,14 +113,15 @@ func main() { checkoutDir = "data" } diskStorage := actor.NewDiskStorage[checkout.CheckoutGrain](checkoutDir, reg) + diskStorage.SetMetrics(checkoutMetrics) var syncedServer *CheckoutPoolServer poolConfig := actor.GrainPoolConfig[checkout.CheckoutGrain]{ MutationRegistry: reg, Storage: diskStorage, + Metrics: checkoutMetrics, Spawn: func(ctx context.Context, id uint64) (actor.Grain[checkout.CheckoutGrain], error) { _, span := tracer.Start(ctx, fmt.Sprintf("Spawn checkout id %d", id)) defer span.End() - grainSpawns.Inc() ret := checkout.NewCheckoutGrain(id, 0, 0, time.Now(), nil) // version to be set later // Load persisted events/state for this checkout if present diff --git a/cmd/checkout/pool-server.go b/cmd/checkout/pool-server.go index b252db0..7d36f28 100644 --- a/cmd/checkout/pool-server.go +++ b/cmd/checkout/pool-server.go @@ -19,8 +19,6 @@ import ( "git.k6n.net/mats/platform/inventory" adyen "github.com/adyen/adyen-go-api-library/v21/src/adyen" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" "go.opentelemetry.io/contrib/bridges/otelslog" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "google.golang.org/protobuf/proto" @@ -33,16 +31,13 @@ import ( "go.opentelemetry.io/otel/trace" ) -var ( - grainMutations = promauto.NewCounter(prometheus.CounterOpts{ - Name: "checkout_grain_mutations_total", - Help: "The total number of mutations", - }) - grainLookups = promauto.NewCounter(prometheus.CounterOpts{ - Name: "checkout_grain_lookups_total", - Help: "The total number of lookups", - }) -) +// Pool metrics (checkout_grain_spawned_total, checkout_grain_lookups_total, +// checkout_mutations_total, checkout_remote_negotiation_total, +// checkout_connected_remotes, …) are bumped centrally by +// SimpleGrainPool — see pkg/actor/simple_grain_pool.go. The checkout +// Metrics is registered once in cmd/checkout/main.go and shared with +// DiskStorage via SetMetrics, so the per-handler counters this file +// used to maintain are no longer needed. type CheckoutPoolServer struct { actor.GrainPool[checkout.CheckoutGrain] diff --git a/cmd/checkout/utils.go b/cmd/checkout/utils.go index f52729f..5ee9217 100644 --- a/cmd/checkout/utils.go +++ b/cmd/checkout/utils.go @@ -167,7 +167,6 @@ func (s *CheckoutPoolServer) ProxyHandler(fn func(w http.ResponseWriter, r *http proxyCalls.Add(ctx, 1, metric.WithAttributes(hostAttr)) handled, err := ownerHost.Proxy(uint64(checkoutId), w, r, nil) - grainLookups.Inc() if err == nil && handled { return nil } diff --git a/cmd/profile/main.go b/cmd/profile/main.go index 22c2748..eaa3e2c 100644 --- a/cmd/profile/main.go +++ b/cmd/profile/main.go @@ -98,6 +98,13 @@ func authSecret() []byte { var podIp = os.Getenv("POD_IP") var name = os.Getenv("POD_NAME") +// profileMetrics is the shared prometheus instrumentation for the +// profile actor pool and event log. Built once at process start; the +// actor package's touch sites are nil-safe so a test binary could pass +// nil. Package-level so tests in this package can construct a pool +// without re-registering (cart/checkout use the same pattern). +var profileMetrics = actor.NewMetrics("profile") + // loadUCPProfileSigner loads the UCP ECDSA signing key from the path specified in // the UCP_SIGNING_KEY_PATH environment variable. Returns nil if unset or unreadable. func loadUCPProfileSigner() *ucp.SigningConfig { @@ -126,10 +133,12 @@ func main() { log.Printf("warning: could not create profile data directory %s: %v", profileDir, err) } diskStorage := actor.NewDiskStorage[profile.ProfileGrain](profileDir, reg) + diskStorage.SetMetrics(profileMetrics) poolConfig := actor.GrainPoolConfig[profile.ProfileGrain]{ MutationRegistry: reg, Storage: diskStorage, + Metrics: profileMetrics, Spawn: func(ctx context.Context, id uint64) (actor.Grain[profile.ProfileGrain], error) { ret := profile.NewProfileGrain(id, time.Now()) err := diskStorage.LoadEvents(ctx, id, ret) diff --git a/grafana_dashboard_cart.json b/grafana_dashboard_cart.json index 8379c13..73e2b5f 100644 --- a/grafana_dashboard_cart.json +++ b/grafana_dashboard_cart.json @@ -71,7 +71,7 @@ "gridPos": { "x": 18, "y": 1, "w": 6, "h": 4 }, "datasource": "${DS_PROMETHEUS}", "targets": [ - { "refId": "A", "expr": "connected_remotes" } + { "refId": "A", "expr": "cart_connected_remotes" } ], "options": { "colorMode": "value", diff --git a/pkg/actor/disk_storage.go b/pkg/actor/disk_storage.go index f90d7e8..c5db18d 100644 --- a/pkg/actor/disk_storage.go +++ b/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() { diff --git a/pkg/actor/metrics.go b/pkg/actor/metrics.go new file mode 100644 index 0000000..0f8bd66 --- /dev/null +++ b/pkg/actor/metrics.go @@ -0,0 +1,216 @@ +package actor + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +// Metrics bundles every prometheus metric the actor package emits. It is +// constructed per service (cart / checkout / profile / order) with a +// service-specific prefix so the metric names do not collide when more +// than one actor-pool service is scraped into the same Prometheus. +// +// All touch sites are nil-safe: the pool / storage / registry code wraps +// each counter / histogram call in a nil-check on the *Metrics pointer, +// so the existing tests (which construct a pool or storage without +// metrics) keep working without a forced dependency on this struct. +// +// Metric names match the panels in grafana_dashboard_cart.json exactly. +// The dashboard had `connected_remotes` (no prefix) as a typo; the +// source-of-truth metric exported here is `_connected_remotes` +// and the dashboard JSON is updated in the same change to match. +type Metrics struct { + // Pool — set by SimpleGrainPool. + ActiveGrains prometheus.Gauge + GrainsInPool prometheus.Gauge + PoolUsage prometheus.Gauge + ConnectedRemotes prometheus.Gauge + GrainSpawnedTotal prometheus.Counter + GrainLookupsTotal prometheus.Counter + MutationsTotal prometheus.Counter + MutationFailuresTotal prometheus.Counter + MutationLatency prometheus.Histogram + RemoteNegotiationTotal prometheus.Counter + + // Event log — set by DiskStorage and StateStorage. + EventLogAppends prometheus.Counter + EventLogBytesWritten prometheus.Counter + EventLogFilesExisting prometheus.Gauge + EventLogLastAppendUnix prometheus.Gauge + EventLogReplayTotal prometheus.Counter + EventLogReplayFailures prometheus.Counter + EventLogReplayDuration prometheus.Histogram + EventLogUnknownTypes prometheus.Counter + EventLogMutationErrors prometheus.Counter +} + +// NewMetrics registers every actor metric in the default Prometheus +// registry (the one promhttp.Handler() exposes on /metrics) with the +// service-specific prefix. The prefix is the same string the binary uses +// as its service name (e.g. "cart", "checkout", "profile", "order"), +// which keeps the metric names aligned with the per-service labels in +// grafana_dashboard_cart.json. +func NewMetrics(prefix string) *Metrics { + return &Metrics{ + ActiveGrains: promauto.NewGauge(prometheus.GaugeOpts{ + Name: prefix + "_active_grains", + Help: "Number of grains currently held in the local pool.", + }), + GrainsInPool: promauto.NewGauge(prometheus.GaugeOpts{ + Name: prefix + "_grains_in_pool", + Help: "Alias of active_grains: the dashboard shows both as a sanity check.", + }), + PoolUsage: promauto.NewGauge(prometheus.GaugeOpts{ + Name: prefix + "_grain_pool_usage", + Help: "Local pool fill ratio: len(grains) / poolSize (range 0..1).", + }), + ConnectedRemotes: promauto.NewGauge(prometheus.GaugeOpts{ + Name: prefix + "_connected_remotes", + Help: "Number of remote actor-pool hosts currently connected via gRPC.", + }), + GrainSpawnedTotal: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_grain_spawned_total", + Help: "Total number of grains spawned (loaded from disk or freshly created).", + }), + GrainLookupsTotal: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_grain_lookups_total", + Help: "Total number of pool.Get calls (state reads).", + }), + MutationsTotal: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_mutations_total", + Help: "Total number of mutations applied across all grains (one per proto message, not per call).", + }), + MutationFailuresTotal: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_mutation_failures_total", + Help: "Total number of pool.Apply calls that returned an error.", + }), + // Latency buckets cover the typical pool.Apply (handler + registry + // apply + storage enqueue): sub-ms in the hot path, hundreds of ms + // under load. Tweak if a hot mutation type moves outside this range. + MutationLatency: promauto.NewHistogram(prometheus.HistogramOpts{ + Name: prefix + "_mutation_latency_seconds", + Help: "Wall-clock duration of a pool.Apply call (lock + spawn + handler + storage enqueue).", + Buckets: prometheus.ExponentialBuckets(0.0005, 2, 14), // 0.5ms ... ~8s + }), + RemoteNegotiationTotal: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_remote_negotiation_total", + Help: "Total number of cluster-negotiation rounds initiated by this pod.", + }), + + EventLogAppends: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_event_log_appends_total", + Help: "Total number of event-log line appends (one per persisted mutation message).", + }), + EventLogBytesWritten: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_event_log_bytes_written_total", + Help: "Total bytes written to per-grain .events.log files.", + }), + EventLogFilesExisting: promauto.NewGauge(prometheus.GaugeOpts{ + Name: prefix + "_event_log_files_existing", + Help: "Number of per-grain .events.log files currently on disk in the storage directory.", + }), + EventLogLastAppendUnix: promauto.NewGauge(prometheus.GaugeOpts{ + Name: prefix + "_event_log_last_append_unix", + Help: "Unix timestamp of the last successful event-log append.", + }), + EventLogReplayTotal: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_event_log_replay_total", + Help: "Total number of successful event-log replays (one per grain loaded from disk).", + }), + EventLogReplayFailures: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_event_log_replay_failures_total", + Help: "Total number of failed event-log replays (open error, JSON parse error, or handler error).", + }), + EventLogReplayDuration: promauto.NewHistogram(prometheus.HistogramOpts{ + Name: prefix + "_event_log_replay_duration_seconds", + Help: "Wall-clock time to replay a single grain's event log from disk.", + Buckets: prometheus.ExponentialBuckets(0.001, 2, 14), // 1ms ... ~16s + }), + EventLogUnknownTypes: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_event_log_unknown_types_total", + Help: "Total number of event-log entries with a type the registry does not recognise.", + }), + EventLogMutationErrors: promauto.NewCounter(prometheus.CounterOpts{ + Name: prefix + "_event_log_mutation_errors_total", + Help: "Total number of mutation-handler errors observed during event-log replay.", + }), + } +} + +// MutationStarted returns a closure to be deferred at the top of +// pool.Apply so the latency observation / counters fire exactly once +// per call regardless of which return path is taken. mutationCount +// is the number of proto messages in the Apply call (a single call +// can batch several messages — SetCartItems, for example, sends +// ClearCart + N AddItem together); the counter is incremented by +// that count so the dashboard's `rate(cart_mutations_total[1m])` +// reports "messages per second" rather than "calls per second", +// matching the original `grainMutations.Add(len(data.Mutations))` +// semantics. The returned closure is nil-safe — pass a nil *Metrics +// in and the call is a no-op. +func (m *Metrics) MutationStarted(mutationCount int) func(err error) { + if m == nil { + return func(error) {} + } + start := time.Now() + return func(err error) { + if mutationCount > 0 { + m.MutationsTotal.Add(float64(mutationCount)) + } + if err != nil { + m.MutationFailuresTotal.Inc() + } + m.MutationLatency.Observe(time.Since(start).Seconds()) + } +} + +// SetPoolStats updates the three pool-level gauges (grains_in_pool, +// active_grains, pool_usage) in one call. Safe to invoke under the +// pool's localMu — it does not take any lock itself, only prometheus +// atomic counters. +func (m *Metrics) SetPoolStats(grains, poolSize int) { + if m == nil { + return + } + m.GrainsInPool.Set(float64(grains)) + m.ActiveGrains.Set(float64(grains)) + if poolSize > 0 { + m.PoolUsage.Set(float64(grains) / float64(poolSize)) + } else { + m.PoolUsage.Set(0) + } +} + +// IncSpawn bumps the grain-spawned counter. Nil-safe. +func (m *Metrics) IncSpawn() { + if m == nil { + return + } + m.GrainSpawnedTotal.Inc() +} + +// IncLookup bumps the grain-lookups counter. Nil-safe. +func (m *Metrics) IncLookup() { + if m == nil { + return + } + m.GrainLookupsTotal.Inc() +} + +// IncNegotiation bumps the remote-negotiation counter. Nil-safe. +func (m *Metrics) IncNegotiation() { + if m == nil { + return + } + m.RemoteNegotiationTotal.Inc() +} + +// SetConnectedRemotes updates the connected-remotes gauge. Nil-safe. +func (m *Metrics) SetConnectedRemotes(n int) { + if m == nil { + return + } + m.ConnectedRemotes.Set(float64(n)) +} diff --git a/pkg/actor/simple_grain_pool.go b/pkg/actor/simple_grain_pool.go index 08afc3a..5edb117 100644 --- a/pkg/actor/simple_grain_pool.go +++ b/pkg/actor/simple_grain_pool.go @@ -28,6 +28,12 @@ type SimpleGrainPool[V any] struct { // grain processes a single message at a time (the actor guarantee). grainLocks *keyedMutex + // metrics is the prometheus instrumentation wired in by NewMetrics + // at composition time. Nil is fine — every touch site nil-checks + // first, so the existing tests (which construct a pool without + // metrics) keep working. + metrics *Metrics + // Cluster coordination -------------------------------------------------- hostname string remoteMu sync.RWMutex @@ -51,6 +57,13 @@ type GrainPoolConfig[V any] struct { PoolSize int MutationRegistry MutationRegistry Storage LogStorage[V] + // Metrics is the prometheus instrumentation to register pool-level + // counters, gauges, and histograms with. Nil-safe: when unset, all + // touch sites are no-ops. Use actor.NewMetrics("cart") (or the + // matching service prefix) to construct the per-service metrics + // struct, and pair it with DiskStorage.SetMetrics so the same + // Metrics instance covers both pool and event-log metrics. + Metrics *Metrics } func NewSimpleGrainPool[V any](config GrainPoolConfig[V]) (*SimpleGrainPool[V], error) { @@ -64,6 +77,7 @@ func NewSimpleGrainPool[V any](config GrainPoolConfig[V]) (*SimpleGrainPool[V], ttl: config.TTL, poolSize: config.PoolSize, hostname: config.Hostname, + metrics: config.Metrics, remoteOwners: make(map[uint64]Host[V]), remoteHosts: make(map[string]Host[V]), grainLocks: newKeyedMutex(), @@ -76,6 +90,11 @@ func NewSimpleGrainPool[V any](config GrainPoolConfig[V]) (*SimpleGrainPool[V], } }() + // Publish the initial pool stats so the dashboard has a number to + // draw on first scrape, rather than waiting for the first grain + // spawn / eviction to bump the gauges. + p.metrics.SetPoolStats(0, p.poolSize) + return p, nil } @@ -96,6 +115,7 @@ func (p *SimpleGrainPool[V]) purge() { purgeLimit := time.Now().Add(-p.ttl) purgedIds := make([]uint64, 0, len(p.grains)) p.localMu.Lock() + evicted := 0 for id, grain := range p.grains { if grain.GetLastAccess().Before(purgeLimit) { purgedIds = append(purgedIds, id) @@ -106,9 +126,14 @@ func (p *SimpleGrainPool[V]) purge() { } delete(p.grains, id) + evicted++ } } + remaining := len(p.grains) p.localMu.Unlock() + if evicted > 0 { + p.metrics.SetPoolStats(remaining, p.poolSize) + } p.forAllHosts(func(remote Host[V]) { remote.AnnounceExpiry(purgedIds) }) @@ -156,15 +181,20 @@ func (p *SimpleGrainPool[V]) HandleOwnershipChange(host string, ids []uint64) er } remoteHost = createdHost } + // Lock order: remoteMu before localMu. Unlock in reverse order so + // the localMu gauge read below cannot deadlock against a concurrent + // p.localMu holder. p.remoteMu.Lock() - defer p.remoteMu.Unlock() p.localMu.Lock() - defer p.localMu.Unlock() for _, id := range ids { log.Printf("Handling ownership change for cart %d to host %s", id, host) delete(p.grains, id) p.remoteOwners[id] = remoteHost } + remaining := len(p.grains) + p.localMu.Unlock() + p.remoteMu.Unlock() + p.metrics.SetPoolStats(remaining, p.poolSize) return nil } @@ -209,8 +239,9 @@ func (p *SimpleGrainPool[V]) AddRemote(host string) (Host[V], error) { return existing, nil } p.remoteHosts[host] = remote + count := len(p.remoteHosts) p.remoteMu.Unlock() - // connectedRemotes.Set(float64(p.RemoteCount())) + p.metrics.SetConnectedRemotes(count) log.Printf("Connected to remote host %s", host) go p.pingLoop(remote) @@ -250,6 +281,7 @@ func (p *SimpleGrainPool[V]) RemoveHost(host string) { delete(p.remoteOwners, id) } } + remoteCount := len(p.remoteHosts) log.Printf("Removing host %s, grains: %d", host, count) p.remoteMu.Unlock() @@ -257,7 +289,7 @@ func (p *SimpleGrainPool[V]) RemoveHost(host string) { if exists { remote.Close() } - // connectedRemotes.Set(float64(p.RemoteCount())) + p.metrics.SetConnectedRemotes(remoteCount) } func (p *SimpleGrainPool[V]) RemoteCount() int { @@ -332,7 +364,7 @@ func (p *SimpleGrainPool[V]) Negotiate(otherHosts []string) { } func (p *SimpleGrainPool[V]) SendNegotiation() { - //negotiationCount.Inc() + p.metrics.IncNegotiation() p.remoteMu.RLock() hosts := make([]string, 0, len(p.remoteHosts)+1) @@ -405,10 +437,13 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(ctx context.Context, id uint64) (Gr if err != nil { return nil, err } + p.metrics.IncSpawn() go p.broadcastOwnership([]uint64{id}) p.localMu.Lock() p.grains[id] = grain + remaining := len(p.grains) p.localMu.Unlock() + p.metrics.SetPoolStats(remaining, p.poolSize) return grain, nil } @@ -417,7 +452,20 @@ func (p *SimpleGrainPool[V]) getOrClaimGrain(ctx context.Context, id uint64) (Gr // var ErrNotOwner = fmt.Errorf("not owner") // Apply applies a mutation to a grain. -func (p *SimpleGrainPool[V]) Apply(ctx context.Context, id uint64, mutation ...proto.Message) (*MutationResult[V], error) { +func (p *SimpleGrainPool[V]) Apply(ctx context.Context, id uint64, mutation ...proto.Message) (result *MutationResult[V], err error) { + // The metric closure fires exactly once on every return path: the + // latency / counters cover the full Apply (lock acquire + spawn + + // handler + storage enqueue), and the failure counter only bumps + // when the call returned a non-nil error. `len(mutation)` is the + // number of proto messages in this Apply call — a single call can + // batch several messages (SetCartItems sends ClearCart + N AddItem + // together), and the counter is incremented by that count so the + // dashboard's `rate(cart_mutations_total[1m])` reports messages + // per second rather than calls per second, matching the original + // `grainMutations.Add(len(data.Mutations))` semantics. + done := p.metrics.MutationStarted(len(mutation)) + defer func() { done(err) }() + // Serialize all access to this grain: spawn, mutation handlers and the // final state read happen atomically with respect to other callers of the // same id. Different ids never contend. @@ -443,19 +491,20 @@ func (p *SimpleGrainPool[V]) Apply(ctx context.Context, id uint64, mutation ...p for _, listener := range p.listeners { go listener.AppendMutations(id, mutations...) } - result, err := grain.GetCurrentState() + state, err := grain.GetCurrentState() if err != nil { return nil, err } return &MutationResult[V]{ - Result: *result, + Result: *state, Mutations: mutations, }, nil } // Get returns the current state of a grain. -func (p *SimpleGrainPool[V]) Get(ctx context.Context, id uint64) (*V, error) { +func (p *SimpleGrainPool[V]) Get(ctx context.Context, id uint64) (result *V, err error) { + p.metrics.IncLookup() unlock := p.grainLocks.lock(id) defer unlock() diff --git a/pkg/actor/state.go b/pkg/actor/state.go index 17483b7..c7b210c 100644 --- a/pkg/actor/state.go +++ b/pkg/actor/state.go @@ -12,6 +12,12 @@ import ( type StateStorage struct { registry MutationRegistry + // metrics is set by the owning DiskStorage via SetMetrics. Nil is fine — + // every method that touches it nil-checks first. The pointer is + // intentionally unexported: callers set it through DiskStorage.SetMetrics + // so the embedded StateStorage always tracks the parent disk storage's + // metrics, even if the embedded value is replaced. + metrics *Metrics } type StorageEvent struct { @@ -50,10 +56,20 @@ func (s *StateStorage) Load(r io.Reader, onMessage func(msg proto.Message, timeS return err } -func (s *StateStorage) Append(io io.Writer, mutation proto.Message, timeStamp time.Time) error { +// Append serialises a mutation as a StorageEvent and writes it (plus a +// trailing newline) to io. The first return value is the number of +// bytes written, which the caller (DiskStorage) adds to the +// event_log_bytes_written_total counter; the second is the underlying +// write error, if any. ErrUnknownType is returned (with 0 bytes) when +// the mutation's type is not registered, and is also counted in +// event_log_unknown_types_total. +func (s *StateStorage) Append(io io.Writer, mutation proto.Message, timeStamp time.Time) (int, error) { typeName, ok := s.registry.GetTypeName(mutation) if !ok { - return ErrUnknownType + if s.metrics != nil { + s.metrics.EventLogUnknownTypes.Inc() + } + return 0, ErrUnknownType } event := &StorageEvent{ Type: typeName, @@ -62,13 +78,16 @@ func (s *StateStorage) Append(io io.Writer, mutation proto.Message, timeStamp ti } jsonBytes, err := json.Marshal(event) if err != nil { - return err + return 0, err } - if _, err := io.Write(jsonBytes); err != nil { - return err + n, err := io.Write(jsonBytes) + total := n + if err != nil { + return total, err } - io.Write([]byte("\n")) - return nil + n, err = io.Write([]byte("\n")) + total += n + return total, err } func (s *StateStorage) Read(r *bufio.Scanner) (*StorageEvent, error) { @@ -83,6 +102,9 @@ func (s *StateStorage) Read(r *bufio.Scanner) (*StorageEvent, error) { typeName := event.Type mutation, ok := s.registry.Create(typeName) if !ok { + if s.metrics != nil { + s.metrics.EventLogUnknownTypes.Inc() + } return nil, ErrUnknownType } if err := json.Unmarshal(event.Mutation, mutation); err != nil {