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_test.go b/internal/handler/heartbeat_test.go index abeebd1..36a8cb5 100644 --- a/internal/handler/heartbeat_test.go +++ b/internal/handler/heartbeat_test.go @@ -1 +1,89 @@ package handler + +import ( + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/containeroo/heartbeats/internal/heartbeat/service" + "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) *API { + api := NewAPI( + "test", + "test", + "http://example.com", + slog.New(slog.NewTextHandler(&strings.Builder{}, nil)), + ) + api.SetService(svc) + 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 reaches the service", func(t *testing.T) { + t.Parallel() + + svc := &fakeService{} + api := newHeartbeatAPI(svc) + + rec := bump(t, api, "api") + require.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, []string{"api"}, svc.updated) + }) + + t.Run("service error maps to 404", func(t *testing.T) { + t.Parallel() + + svc := &fakeService{updateErr: errors.New("heartbeat \"api\" not found")} + api := newHeartbeatAPI(svc) + + rec := bump(t, api, "api") + require.Equal(t, http.StatusNotFound, rec.Code) + }) + + t.Run("missing id is rejected before the service", func(t *testing.T) { + t.Parallel() + + svc := &fakeService{} + api := newHeartbeatAPI(svc) + + rec := bump(t, api, "") + require.Equal(t, http.StatusBadRequest, rec.Code) + assert.Empty(t, svc.updated) + }) +} 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) {