From adab18d6b2e7cfdafe89352e5bda5a137e3ed9fe Mon Sep 17 00:00:00 2001 From: Christopher Plieger <917744+cplieger@users.noreply.github.com> Date: Thu, 23 Jul 2026 13:22:00 +0200 Subject: [PATCH 1/2] fix: increment heartbeats_heartbeat_received_total on accepted bumps --- internal/handler/heartbeat.go | 3 + internal/handler/heartbeat_test.go | 108 +++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+) diff --git a/internal/handler/heartbeat.go b/internal/handler/heartbeat.go index 8a267de..d418b42 100644 --- a/internal/handler/heartbeat.go +++ b/internal/handler/heartbeat.go @@ -23,6 +23,9 @@ func (a *API) Heartbeat() http.HandlerFunc { a.respondJSON(w, http.StatusNotFound, errorResponse{Error: err.Error()}) return } + if a.metrics != nil { + a.metrics.IncHeartbeatReceived(heartbeatID) + } a.respondJSON(w, http.StatusOK, statusResponse{Status: "ok"}) } } diff --git a/internal/handler/heartbeat_test.go b/internal/handler/heartbeat_test.go index abeebd1..e02c68a 100644 --- a/internal/handler/heartbeat_test.go +++ b/internal/handler/heartbeat_test.go @@ -1 +1,109 @@ package handler + +import ( + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/containeroo/heartbeats/internal/heartbeat/service" + "github.com/containeroo/heartbeats/internal/metrics" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeService struct { + updateErr error + updated []string +} + +func (f *fakeService) HeartbeatSummaries() []service.HeartbeatSummary { return nil } +func (f *fakeService) ReceiverSummaries() []service.ReceiverSummary { return nil } +func (f *fakeService) StatusAll() []service.Status { return nil } +func (f *fakeService) StatusByID(id string) (service.Status, error) { return service.Status{}, nil } + +func (f *fakeService) Update(id string, payload string, now time.Time) error { + if f.updateErr != nil { + return f.updateErr + } + f.updated = append(f.updated, id) + return nil +} + +func newHeartbeatAPI(svc ServiceProvider, metricsReg *metrics.Registry) *API { + api := NewAPI( + "test", + "test", + "http://example.com", + slog.New(slog.NewTextHandler(&strings.Builder{}, nil)), + ) + api.SetService(svc) + api.SetMetrics(metricsReg) + return api +} + +func bump(t *testing.T, api *API, id string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest("POST", "/api/heartbeat/"+id, strings.NewReader("payload")) + req.SetPathValue("id", id) + rec := httptest.NewRecorder() + api.Heartbeat().ServeHTTP(rec, req) + return rec +} + +func TestHeartbeatHandler(t *testing.T) { + t.Parallel() + + t.Run("accepted bump increments received counter", func(t *testing.T) { + t.Parallel() + + svc := &fakeService{} + metricsReg := metrics.NewRegistry() + api := newHeartbeatAPI(svc, metricsReg) + + rec := bump(t, api, "api") + require.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, []string{"api"}, svc.updated) + + metricsRec := httptest.NewRecorder() + api.Metrics().ServeHTTP(metricsRec, httptest.NewRequest("GET", "/metrics", nil)) + assert.Contains(t, + metricsRec.Body.String(), + `heartbeats_heartbeat_received_total{heartbeat="api"} 1`, + "Expected an accepted bump to increment 'heartbeats_heartbeat_received_total'", + ) + }) + + t.Run("rejected bump does not increment received counter", func(t *testing.T) { + t.Parallel() + + svc := &fakeService{updateErr: errors.New("heartbeat \"api\" not found")} + metricsReg := metrics.NewRegistry() + api := newHeartbeatAPI(svc, metricsReg) + + rec := bump(t, api, "api") + require.Equal(t, http.StatusNotFound, rec.Code) + + metricsRec := httptest.NewRecorder() + api.Metrics().ServeHTTP(metricsRec, httptest.NewRequest("GET", "/metrics", nil)) + assert.NotContains(t, + metricsRec.Body.String(), + "heartbeats_heartbeat_received_total{", + "Expected a rejected bump to leave 'heartbeats_heartbeat_received_total' untouched", + ) + }) + + t.Run("missing id is rejected", func(t *testing.T) { + t.Parallel() + + svc := &fakeService{} + api := newHeartbeatAPI(svc, metrics.NewRegistry()) + + rec := bump(t, api, "") + require.Equal(t, http.StatusBadRequest, rec.Code) + assert.Empty(t, svc.updated) + }) +} From 0eeb120513a2a607ee6aed903e94381fc61f718e Mon Sep 17 00:00:00 2001 From: Christopher Plieger <917744+cplieger@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:03:03 +0200 Subject: [PATCH 2/2] fix: count received bumps in the service layer --- internal/app/run.go | 2 +- internal/handler/heartbeat.go | 3 -- internal/handler/heartbeat_test.go | 34 ++++-------------- internal/heartbeat/service/service.go | 20 ++++++++++- internal/heartbeat/service/service_test.go | 41 +++++++++++++++++++++- 5 files changed, 67 insertions(+), 33 deletions(-) diff --git a/internal/app/run.go b/internal/app/run.go index 9b75e0f..9046a5b 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -138,7 +138,7 @@ func Run(ctx context.Context, appFS fs.FS, version, commit string, args []string } manager.StartAll(ctx) - svc := service.NewService(manager, notifyManager, historyRecorder) + svc := service.NewService(manager, notifyManager, historyRecorder, metricsReg) api.SetService(svc) reloadConfigFn := reconcile.NewReloadFunc( diff --git a/internal/handler/heartbeat.go b/internal/handler/heartbeat.go index d418b42..8a267de 100644 --- a/internal/handler/heartbeat.go +++ b/internal/handler/heartbeat.go @@ -23,9 +23,6 @@ func (a *API) Heartbeat() http.HandlerFunc { a.respondJSON(w, http.StatusNotFound, errorResponse{Error: err.Error()}) return } - if a.metrics != nil { - a.metrics.IncHeartbeatReceived(heartbeatID) - } a.respondJSON(w, http.StatusOK, statusResponse{Status: "ok"}) } } diff --git a/internal/handler/heartbeat_test.go b/internal/handler/heartbeat_test.go index e02c68a..36a8cb5 100644 --- a/internal/handler/heartbeat_test.go +++ b/internal/handler/heartbeat_test.go @@ -10,7 +10,6 @@ import ( "time" "github.com/containeroo/heartbeats/internal/heartbeat/service" - "github.com/containeroo/heartbeats/internal/metrics" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -33,7 +32,7 @@ func (f *fakeService) Update(id string, payload string, now time.Time) error { return nil } -func newHeartbeatAPI(svc ServiceProvider, metricsReg *metrics.Registry) *API { +func newHeartbeatAPI(svc ServiceProvider) *API { api := NewAPI( "test", "test", @@ -41,7 +40,6 @@ func newHeartbeatAPI(svc ServiceProvider, metricsReg *metrics.Registry) *API { slog.New(slog.NewTextHandler(&strings.Builder{}, nil)), ) api.SetService(svc) - api.SetMetrics(metricsReg) return api } @@ -57,50 +55,32 @@ func bump(t *testing.T, api *API, id string) *httptest.ResponseRecorder { func TestHeartbeatHandler(t *testing.T) { t.Parallel() - t.Run("accepted bump increments received counter", func(t *testing.T) { + t.Run("accepted bump reaches the service", func(t *testing.T) { t.Parallel() svc := &fakeService{} - metricsReg := metrics.NewRegistry() - api := newHeartbeatAPI(svc, metricsReg) + api := newHeartbeatAPI(svc) rec := bump(t, api, "api") require.Equal(t, http.StatusOK, rec.Code) assert.Equal(t, []string{"api"}, svc.updated) - - metricsRec := httptest.NewRecorder() - api.Metrics().ServeHTTP(metricsRec, httptest.NewRequest("GET", "/metrics", nil)) - assert.Contains(t, - metricsRec.Body.String(), - `heartbeats_heartbeat_received_total{heartbeat="api"} 1`, - "Expected an accepted bump to increment 'heartbeats_heartbeat_received_total'", - ) }) - t.Run("rejected bump does not increment received counter", func(t *testing.T) { + t.Run("service error maps to 404", func(t *testing.T) { t.Parallel() svc := &fakeService{updateErr: errors.New("heartbeat \"api\" not found")} - metricsReg := metrics.NewRegistry() - api := newHeartbeatAPI(svc, metricsReg) + api := newHeartbeatAPI(svc) rec := bump(t, api, "api") require.Equal(t, http.StatusNotFound, rec.Code) - - metricsRec := httptest.NewRecorder() - api.Metrics().ServeHTTP(metricsRec, httptest.NewRequest("GET", "/metrics", nil)) - assert.NotContains(t, - metricsRec.Body.String(), - "heartbeats_heartbeat_received_total{", - "Expected a rejected bump to leave 'heartbeats_heartbeat_received_total' untouched", - ) }) - t.Run("missing id is rejected", func(t *testing.T) { + t.Run("missing id is rejected before the service", func(t *testing.T) { t.Parallel() svc := &fakeService{} - api := newHeartbeatAPI(svc, metrics.NewRegistry()) + api := newHeartbeatAPI(svc) rec := bump(t, api, "") require.Equal(t, http.StatusBadRequest, rec.Code) diff --git a/internal/heartbeat/service/service.go b/internal/heartbeat/service/service.go index 9f6cdbf..e94b7c8 100644 --- a/internal/heartbeat/service/service.go +++ b/internal/heartbeat/service/service.go @@ -13,6 +13,7 @@ import ( htypes "github.com/containeroo/heartbeats/internal/heartbeat/types" "github.com/containeroo/heartbeats/internal/history" + "github.com/containeroo/heartbeats/internal/metrics" "github.com/containeroo/heartbeats/internal/runner" ) @@ -32,6 +33,7 @@ type Service struct { manager Store receivers ReceiverStore history history.Recorder + metrics *metrics.Registry } // Status is the public heartbeat state payload. @@ -69,11 +71,12 @@ type ReceiverSummary struct { } // NewService constructs a Service from a Manager. -func NewService(manager Store, receivers ReceiverStore, historyStore history.Recorder) *Service { +func NewService(manager Store, receivers ReceiverStore, historyStore history.Recorder, metricsReg *metrics.Registry) *Service { return &Service{ manager: manager, receivers: receivers, history: historyStore, + metrics: metricsReg, } } @@ -81,16 +84,31 @@ func NewService(manager Store, receivers ReceiverStore, historyStore history.Rec func (s *Service) Update(id string, payload string, now time.Time) error { hb, ok := s.manager.Get(id) if !ok { + // Unknown ids are deliberately not counted: the id becomes a metric + // label, and counting arbitrary request paths would let callers mint + // unbounded label cardinality. return fmt.Errorf("heartbeat %q not found", id) } if ok := hb.State.UpdateSeen(now, payload); !ok { + // The state (last seen, payload) was still updated; only the runner + // wake-up was dropped. The bump was received, so it counts. s.recordHeartbeatReceived(id, now, len(payload), false) + s.incHeartbeatReceived(id) return errors.New("heartbeat mailbox full") } s.recordHeartbeatReceived(id, now, len(payload), true) + s.incHeartbeatReceived(id) return nil } +// incHeartbeatReceived bumps the received counter when metrics are wired. +func (s *Service) incHeartbeatReceived(id string) { + if s.metrics == nil { + return + } + s.metrics.IncHeartbeatReceived(id) +} + // StatusByID returns the status for a heartbeat. func (s *Service) StatusByID(id string) (Status, error) { hb, ok := s.manager.Get(id) diff --git a/internal/heartbeat/service/service_test.go b/internal/heartbeat/service/service_test.go index ad5e35a..41ebbbe 100644 --- a/internal/heartbeat/service/service_test.go +++ b/internal/heartbeat/service/service_test.go @@ -1,6 +1,7 @@ package service import ( + "net/http/httptest" "testing" "time" @@ -11,6 +12,7 @@ import ( "github.com/containeroo/heartbeats/internal/config" "github.com/containeroo/heartbeats/internal/heartbeat/types" "github.com/containeroo/heartbeats/internal/history" + "github.com/containeroo/heartbeats/internal/metrics" "github.com/containeroo/heartbeats/internal/runner" ) @@ -87,10 +89,18 @@ func newTestService(t *testing.T) (*Service, *fakeStore, *history.Store) { t.Helper() store := newFakeStore() hist := history.NewStore(10) - svc := NewService(store, newReceiverStore(), hist) + svc := NewService(store, newReceiverStore(), hist, metrics.NewRegistry()) return svc, store, hist } +// scrapeMetrics renders the service's metrics registry as Prometheus text. +func scrapeMetrics(t *testing.T, svc *Service) string { + t.Helper() + rec := httptest.NewRecorder() + svc.metrics.Metrics().ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil)) + return rec.Body.String() +} + func TestServiceUpdate(t *testing.T) { t.Parallel() @@ -106,6 +116,35 @@ func TestServiceUpdate(t *testing.T) { events := hist.List() require.Len(t, events, 2) require.Equal(t, false, events[1].Fields["enqueued"]) + + // Both deliveries were received (the mailbox-full one still updated + // state), so both count. + require.Contains(t, scrapeMetrics(t, svc), + `heartbeats_heartbeat_received_total{heartbeat="api"} 2`) +} + +func TestServiceUpdateUnknownID(t *testing.T) { + t.Parallel() + + svc, _, _ := newTestService(t) + + err := svc.Update("ghost", "payload", time.Now()) + require.Error(t, err) + require.Contains(t, err.Error(), "not found") + + // Unknown ids must not mint counter series (label cardinality guard). + require.NotContains(t, scrapeMetrics(t, svc), + "heartbeats_heartbeat_received_total{") +} + +func TestServiceUpdateWithoutMetrics(t *testing.T) { + t.Parallel() + + store := newFakeStore() + svc := NewService(store, newReceiverStore(), history.NewStore(10), nil) + store.s["api"] = newHeartbeat(t, "api", time.Second, 2*time.Second, runner.StageLate, time.Now()) + + require.NoError(t, svc.Update("api", "payload", time.Now())) } func TestServiceStatus(t *testing.T) {