diff --git a/internal/app/run.go b/internal/app/run.go index 9dfa850..8bcb077 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -51,10 +51,10 @@ func Run(ctx context.Context, webFS fs.FS, version, commit string, args []string } logger := logging.SetupLogger(flags.LogFormat, flags.Debug, w) - logger.Info("Starting Heartbeats", "version", version, "commit", commit) + logging.SystemLogger(logger, nil).Info("Starting Heartbeats", "version", version, "commit", commit) if len(flags.OverriddenValues) > 0 { - logger.Info("CLI Overrides", "overrides", flags.OverriddenValues) + logging.SystemLogger(logger, nil).Info("CLI Overrides", "overrides", flags.OverriddenValues) } histStore, err := history.InitializeHistory(flags.HistorySize) @@ -120,11 +120,10 @@ func Run(ctx context.Context, webFS fs.FS, version, commit string, args []string } // Create server and run forever - router := server.NewRouter( - webFS, - api, - logger, - ) + router, err := server.NewRouter(webFS, flags.RoutePrefix, api, flags.Debug) + if err != nil { + return fmt.Errorf("configure router: %w", err) + } if err := server.Run(ctx, flags.ListenAddr, router, logger); err != nil { return fmt.Errorf("failed to run Heartbeats: %w", err) } diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 74df6e9..2de8de1 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -169,8 +169,8 @@ heartbeats: assert.True(t, waitForLog(t, &buf, `"from":"active","to":"grace"`, 2*time.Second), "expected Active → Grace") assert.True(t, waitForLog(t, &buf, `"from":"grace","to":"missing"`, 2*time.Second), "expected Grace → Missing") - assert.True(t, waitForLog(t, &buf, `"msg":"retrying","attempt":1`, 2*time.Second), "expected retry 1") - assert.True(t, waitForLog(t, &buf, `"msg":"retrying","attempt":2`, 2*time.Second), "expected retry 2") - assert.True(t, waitForLog(t, &buf, `notification failed after 2 retries`, 2*time.Second), "expected failure log") + assert.True(t, waitForLog(t, &buf, `"msg":"retrying","attempt":1`, 5*time.Second), "expected retry 1") + assert.True(t, waitForLog(t, &buf, `"msg":"retrying","attempt":2`, 5*time.Second), "expected retry 2") + assert.True(t, waitForLog(t, &buf, `notification failed after 2 retries`, 5*time.Second), "expected failure log") }) } diff --git a/internal/config/config.go b/internal/config/config.go index 622e63d..cc39e8a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/containeroo/heartbeats/internal/heartbeat" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/notifier" "gopkg.in/yaml.v3" ) @@ -132,7 +133,7 @@ func NewReloadFunc( return err } if res.Added+res.Updated+res.Removed > 0 { - logger.Info("heartbeats reloaded", + logging.SystemLogger(logger, nil).Info("heartbeats reloaded", "added", res.Added, "updated", res.Updated, "removed", res.Removed, @@ -149,11 +150,11 @@ func WatchReload(ctx context.Context, reloadCh <-chan os.Signal, logger *slog.Lo case <-ctx.Done(): return case <-reloadCh: - logger.Info("reload requested") + logging.SystemLogger(logger, nil).Info("reload requested") if err := reloadFn(); err != nil { - logger.Error("reload failed", "err", err) + logging.SystemLogger(logger, nil).Error("reload failed", "err", err) } else { - logger.Info("reload completed") + logging.SystemLogger(logger, nil).Info("reload completed") } } } diff --git a/internal/debugserver/debugserver.go b/internal/debugserver/debugserver.go index 6206149..5742b23 100644 --- a/internal/debugserver/debugserver.go +++ b/internal/debugserver/debugserver.go @@ -6,6 +6,7 @@ import ( "net/http" "github.com/containeroo/heartbeats/internal/handler" + "github.com/containeroo/heartbeats/internal/logging" ) // Run starts the local-only debug server for manual testing. @@ -16,7 +17,7 @@ func Run(ctx context.Context, port int, api *handler.API) { mux.Handle("GET /internal/heartbeat/{id}", api.TestHeartbeatHandler()) addr := fmt.Sprintf("127.0.0.1:%d", port) - api.Logger.Info("starting debug server", "listenAddr", addr) + logging.SystemLogger(api.Logger, nil).Info("starting debug server", "listenAddr", addr) server := &http.Server{ Addr: addr, @@ -32,7 +33,7 @@ func Run(ctx context.Context, port int, api *handler.API) { // Serve requests on 127.0.0.1 until shutdown. go func() { if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { - api.Logger.Error("debug server error", "error", err) + logging.SystemLogger(api.Logger, nil).Error("debug server error", "error", err) } }() } diff --git a/internal/handler/api.go b/internal/handler/api.go index 78d5df9..9ba8d1d 100644 --- a/internal/handler/api.go +++ b/internal/handler/api.go @@ -7,6 +7,7 @@ import ( "github.com/containeroo/heartbeats/internal/heartbeat" "github.com/containeroo/heartbeats/internal/history" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/metrics" "github.com/containeroo/heartbeats/internal/notifier" servicehistory "github.com/containeroo/heartbeats/internal/service/history" @@ -64,10 +65,28 @@ func NewAPI( // respondJSON writes a JSON response. func (a *API) respondJSON(w http.ResponseWriter, status int, v any) { if err := encode(w, status, v); err != nil { - a.Logger.Error("encode response failed", "err", err) + logging.AccessLogger(a.Logger, nil).Error( + "encode response failed", + "event", "encode_response_failed", + "err", err, + ) } } +// businessLogger returns a logger enriched with request context for business events. +func (a *API) businessLogger(r *http.Request) *slog.Logger { + return logging.BusinessLogger(a.Logger, r.Context()) +} + +// logRequestError records a structured error for the current request. +func (a *API) logRequestError(r *http.Request, event, message string, err error) { + logging.AccessLogger(a.Logger, r.Context()).Error( + message, + "event", event, + "err", err, + ) +} + // statusResponse is the standard success payload. type statusResponse struct { Status string `json:"status"` diff --git a/internal/handler/bump.go b/internal/handler/bump.go index c939429..6ae020b 100644 --- a/internal/handler/bump.go +++ b/internal/handler/bump.go @@ -13,19 +13,23 @@ func (a *API) BumpHandler() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") if id == "" { + a.logRequestError(r, "missing_id", "missing id", errors.New("missing id")) a.respondJSON(w, http.StatusBadRequest, errorResponse{Error: "missing id"}) return } if err := bump.Receive(r.Context(), a.mgr, a.rec, a.Logger, id, r.RemoteAddr, r.Method, r.UserAgent()); err != nil { if errors.Is(err, bump.ErrUnknownHeartbeat) { + a.logRequestError(r, "unknown_heartbeat", "unknown heartbeat id", err) a.respondJSON(w, http.StatusNotFound, errorResponse{Error: fmt.Sprintf("unknown heartbeat id %q", id)}) return } + a.logRequestError(r, "failed_to_receive", "failed to receive heartbeat", err) a.respondJSON(w, http.StatusInternalServerError, errorResponse{Error: err.Error()}) return } + a.businessLogger(r).Info("received heartbeat", "id", id) a.respondJSON(w, http.StatusOK, statusResponse{Status: "ok"}) }) } @@ -35,19 +39,23 @@ func (a *API) FailHandler() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") if id == "" { + a.logRequestError(r, "missing_id", "missing id", errors.New("missing id")) a.respondJSON(w, http.StatusBadRequest, errorResponse{Error: "missing id"}) return } if err := bump.Fail(r.Context(), a.mgr, a.rec, a.Logger, id, r.RemoteAddr, r.Method, r.UserAgent()); err != nil { if errors.Is(err, bump.ErrUnknownHeartbeat) { + a.logRequestError(r, "unknown_heartbeat", "unknown heartbeat id", err) a.respondJSON(w, http.StatusNotFound, errorResponse{Error: fmt.Sprintf("unknown heartbeat id %q", id)}) return } + a.logRequestError(r, "failed_to_fail", "failed to fail heartbeat", err) a.respondJSON(w, http.StatusInternalServerError, errorResponse{Error: err.Error()}) return } + a.businessLogger(r).Info("failed heartbeat", "id", id) a.respondJSON(w, http.StatusOK, statusResponse{Status: "ok"}) }) } diff --git a/internal/handler/home.go b/internal/handler/home.go index f6c6d8c..ab79bf8 100644 --- a/internal/handler/home.go +++ b/internal/handler/home.go @@ -26,6 +26,7 @@ func (a *API) HomeHandler() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { // execute the "base" template if err := tmpl.ExecuteTemplate(w, "base", data); err != nil { + a.logRequestError(r, "render_home_failed", "render home failed", err) a.respondJSON(w, http.StatusInternalServerError, errorResponse{Error: err.Error()}) } } diff --git a/internal/handler/partials.go b/internal/handler/partials.go index e5786ee..3e1b7cc 100644 --- a/internal/handler/partials.go +++ b/internal/handler/partials.go @@ -44,7 +44,7 @@ func (a *API) PartialHandler( } if err != nil { - a.Logger.Error("render "+section+" partial", "error", err) + a.logRequestError(r, "render_partial_failed", "render "+section+" partial", err) a.respondJSON(w, http.StatusInternalServerError, errorResponse{Error: "internal error"}) } } diff --git a/internal/handler/test.go b/internal/handler/test.go index fe6cf44..0585e11 100644 --- a/internal/handler/test.go +++ b/internal/handler/test.go @@ -1,6 +1,7 @@ package handler import ( + "errors" "net/http" testservice "github.com/containeroo/heartbeats/internal/service/test" @@ -11,6 +12,7 @@ func (a *API) TestReceiverHandler() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") if id == "" { + a.logRequestError(r, "missing_id", "missing id", errors.New("missing id")) a.respondJSON(w, http.StatusBadRequest, errorResponse{Error: "missing id"}) return } @@ -26,12 +28,13 @@ func (a *API) TestHeartbeatHandler() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") if id == "" { + a.logRequestError(r, "missing_id", "missing id", errors.New("missing id")) a.respondJSON(w, http.StatusBadRequest, errorResponse{Error: "missing id"}) return } if err := testservice.TriggerTestHeartbeat(a.mgr, a.Logger, id); err != nil { - a.Logger.Error("handle test failed", "id", id, "err", err) + a.logRequestError(r, "handle_test_failed", "handle test failed", err) a.respondJSON(w, http.StatusNotFound, errorResponse{Error: err.Error()}) return } diff --git a/internal/heartbeat/actor.go b/internal/heartbeat/actor.go index 2cce89e..6ade716 100644 --- a/internal/heartbeat/actor.go +++ b/internal/heartbeat/actor.go @@ -5,6 +5,7 @@ import ( "log/slog" "time" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/metrics" "github.com/containeroo/heartbeats/internal/notifier" servicehistory "github.com/containeroo/heartbeats/internal/service/history" @@ -142,7 +143,7 @@ func (a *Actor) onReceive() { a.State = HeartbeatStateActive if err := a.recordStateChange(prev, a.State); err != nil { - a.logger.Error("failed to record state change", "err", err) + logging.BusinessLogger(a.logger, a.ctx).Error("failed to record state change", "err", err) } a.LastBump = now a.checkTimer = time.NewTimer(a.Interval) @@ -171,7 +172,7 @@ func (a *Actor) onFail() { a.State = HeartbeatStateFailed if err := a.recordStateChange(prev, a.State); err != nil { - a.logger.Error("failed to record state change", "err", err) + logging.BusinessLogger(a.logger, a.ctx).Error("failed to record state change", "err", err) } a.metrics.SetHeartbeatStatus(a.ID, metrics.DOWN) @@ -186,7 +187,7 @@ func (a *Actor) onEnterGrace() { prev := a.State a.State = HeartbeatStateGrace if err := a.recordStateChange(prev, a.State); err != nil { - a.logger.Error("failed to record state change", "err", err) + logging.BusinessLogger(a.logger, a.ctx).Error("failed to record state change", "err", err) } a.graceTimer = time.NewTimer(a.Grace) @@ -200,7 +201,7 @@ func (a *Actor) onEnterMissing() { prev := a.State a.State = HeartbeatStateMissing if err := a.recordStateChange(prev, a.State); err != nil { - a.logger.Error("failed to record state change", "err", err) + logging.BusinessLogger(a.logger, a.ctx).Error("failed to record state change", "err", err) } // send notification diff --git a/internal/heartbeat/manager.go b/internal/heartbeat/manager.go index 478eb56..62c14d8 100644 --- a/internal/heartbeat/manager.go +++ b/internal/heartbeat/manager.go @@ -7,6 +7,7 @@ import ( "slices" "sync" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/utils" ) @@ -71,7 +72,7 @@ func (m *Manager) StartAll() int { ma.started = true go ma.actor.Run(ctx) started++ - m.logger.Debug("started heartbeat", "id", ma.actor.ID) + logging.SystemLogger(m.logger, nil).Debug("started heartbeat", "id", ma.actor.ID) } if started > 0 { m.started = true diff --git a/internal/heartbeat/utils.go b/internal/heartbeat/utils.go index f09c2c2..e0b3963 100644 --- a/internal/heartbeat/utils.go +++ b/internal/heartbeat/utils.go @@ -3,6 +3,7 @@ package heartbeat import ( "time" + "github.com/containeroo/heartbeats/internal/logging" servicehistory "github.com/containeroo/heartbeats/internal/service/history" ) @@ -48,7 +49,7 @@ func (a *Actor) recordStateChange(prev, next HeartbeatState) error { from := prev.String() to := next.String() - a.logger.Info("state change", + logging.BusinessLogger(a.logger, a.ctx).Info("state change", "heartbeat", a.ID, "from", from, "to", to, diff --git a/internal/logging/categories.go b/internal/logging/categories.go new file mode 100644 index 0000000..2908190 --- /dev/null +++ b/internal/logging/categories.go @@ -0,0 +1,44 @@ +package logging + +import ( + "context" + "log/slog" +) + +const ( + CategoryAccess = "access" + CategoryBusiness = "business" + CategoryDB = "db" + CategorySystem = "system" +) + +// WithCategory adds a category field to the logger. +func WithCategory(logger *slog.Logger, category string) *slog.Logger { + if logger == nil { + logger = slog.Default() + } + if category == "" { + return logger + } + return logger.With("category", category) +} + +// AccessLogger returns a logger tagged for access logs. +func AccessLogger(logger *slog.Logger, ctx context.Context) *slog.Logger { + return WithCategory(WithRequestIDLogger(logger, ctx), CategoryAccess) +} + +// BusinessLogger returns a logger tagged for business logs. +func BusinessLogger(logger *slog.Logger, ctx context.Context) *slog.Logger { + return WithCategory(WithRequestIDLogger(logger, ctx), CategoryBusiness) +} + +// DBLogger returns a logger tagged for db logs. +func DBLogger(logger *slog.Logger, ctx context.Context) *slog.Logger { + return WithCategory(WithRequestIDLogger(logger, ctx), CategoryDB) +} + +// SystemLogger returns a logger tagged for system logs. +func SystemLogger(logger *slog.Logger, ctx context.Context) *slog.Logger { + return WithCategory(WithRequestIDLogger(logger, ctx), CategorySystem) +} diff --git a/internal/logging/logger_test.go b/internal/logging/logger_test.go index 41e746e..b57ac14 100644 --- a/internal/logging/logger_test.go +++ b/internal/logging/logger_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/assert" ) +// TestSetupLogger validates logger output formats and log levels. func TestSetupLogger(t *testing.T) { t.Parallel() diff --git a/internal/logging/request.go b/internal/logging/request.go new file mode 100644 index 0000000..0a1d66c --- /dev/null +++ b/internal/logging/request.go @@ -0,0 +1,51 @@ +package logging + +import ( + "context" + "crypto/rand" + "encoding/hex" + "io" + "log/slog" +) + +const RequestIDHeader = "X-Request-Id" + +type requestIDKey struct{} + +var requestIDReader io.Reader = rand.Reader + +// NewRequestID generates a random request id. +func NewRequestID() string { + var buf [16]byte + if _, err := io.ReadFull(requestIDReader, buf[:]); err != nil { + return "" + } + return hex.EncodeToString(buf[:]) +} + +// WithRequestID stores a request id in the context. +func WithRequestID(ctx context.Context, id string) context.Context { + return context.WithValue(ctx, requestIDKey{}, id) +} + +// RequestID extracts a request id from the context. +func RequestID(ctx context.Context) string { + if ctx == nil { + return "" + } + if v, ok := ctx.Value(requestIDKey{}).(string); ok { + return v + } + return "" +} + +// WithRequestIDLogger adds request_id to the logger if present in ctx. +func WithRequestIDLogger(logger *slog.Logger, ctx context.Context) *slog.Logger { + if logger == nil { + logger = slog.Default() + } + if id := RequestID(ctx); id != "" { + return logger.With("request_id", id) + } + return logger +} diff --git a/internal/logging/request_test.go b/internal/logging/request_test.go new file mode 100644 index 0000000..939953e --- /dev/null +++ b/internal/logging/request_test.go @@ -0,0 +1,161 @@ +package logging + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type errReader struct{} + +func (errReader) Read([]byte) (int, error) { + return 0, errors.New("boom") +} + +func TestNewRequestID(t *testing.T) { + t.Run("generates a hex id", func(t *testing.T) { + t.Parallel() + + id := NewRequestID() + assert.Len(t, id, 32) + }) + + t.Run("returns empty string on rand failure", func(t *testing.T) { + // Cannot run in parallel because it mutates crypto/rand.Reader. + old := requestIDReader + requestIDReader = errReader{} + defer func() { requestIDReader = old }() + + id := NewRequestID() + assert.Equal(t, "", id) + }) +} + +func TestRequestID(t *testing.T) { + t.Parallel() + + t.Run("returns empty on nil context", func(t *testing.T) { + t.Parallel() + + assert.Equal(t, "", RequestID(context.TODO())) + }) + + t.Run("returns stored id", func(t *testing.T) { + t.Parallel() + + ctx := WithRequestID(context.Background(), "req-123") + assert.Equal(t, "req-123", RequestID(ctx)) + }) +} + +func TestWithRequestIDLogger(t *testing.T) { + t.Parallel() + + t.Run("adds request_id when present", func(t *testing.T) { + t.Parallel() + + buf := &bytes.Buffer{} + logger := slog.New(slog.NewJSONHandler(buf, nil)) + ctx := WithRequestID(context.Background(), "req-456") + + WithRequestIDLogger(logger, ctx).Info("hello") + + entry := readLogJSON(t, buf) + assert.Equal(t, "req-456", entry["request_id"]) + }) + + t.Run("does not panic with nil logger", func(t *testing.T) { + t.Parallel() + + require.NotPanics(t, func() { + WithRequestIDLogger(nil, context.Background()).Info("hello") + }) + }) +} + +func TestWithCategory(t *testing.T) { + t.Parallel() + + t.Run("returns same logger when category empty", func(t *testing.T) { + t.Parallel() + + logger := slog.New(slog.NewJSONHandler(io.Discard, nil)) + assert.Same(t, logger, WithCategory(logger, "")) + }) + + t.Run("adds category field", func(t *testing.T) { + t.Parallel() + + buf := &bytes.Buffer{} + logger := slog.New(slog.NewJSONHandler(buf, nil)) + WithCategory(logger, CategoryAccess).Info("hello") + + entry := readLogJSON(t, buf) + assert.Equal(t, CategoryAccess, entry["category"]) + }) + + t.Run("does not panic with nil logger", func(t *testing.T) { + t.Parallel() + + require.NotPanics(t, func() { + WithCategory(nil, CategoryAccess).Info("hello") + }) + }) +} + +func TestCategoryLoggers(t *testing.T) { + t.Parallel() + + t.Run("adds category and request_id", func(t *testing.T) { + t.Parallel() + + buf := &bytes.Buffer{} + logger := slog.New(slog.NewJSONHandler(buf, nil)) + ctx := WithRequestID(context.Background(), "req-789") + + AccessLogger(logger, ctx).Info("access") + entry := readLogJSON(t, buf) + assert.Equal(t, CategoryAccess, entry["category"]) + assert.Equal(t, "req-789", entry["request_id"]) + + buf.Reset() + BusinessLogger(logger, ctx).Info("business") + entry = readLogJSON(t, buf) + assert.Equal(t, CategoryBusiness, entry["category"]) + assert.Equal(t, "req-789", entry["request_id"]) + + buf.Reset() + DBLogger(logger, ctx).Info("db") + entry = readLogJSON(t, buf) + assert.Equal(t, CategoryDB, entry["category"]) + assert.Equal(t, "req-789", entry["request_id"]) + + buf.Reset() + SystemLogger(logger, ctx).Info("system") + entry = readLogJSON(t, buf) + assert.Equal(t, CategorySystem, entry["category"]) + assert.Equal(t, "req-789", entry["request_id"]) + }) +} + +func readLogJSON(t *testing.T, buf *bytes.Buffer) map[string]any { + t.Helper() + + line := strings.TrimSpace(buf.String()) + if idx := strings.IndexByte(line, '\n'); idx != -1 { + line = line[:idx] + } + + var entry map[string]any + err := json.Unmarshal([]byte(line), &entry) + require.NoError(t, err) + return entry +} diff --git a/internal/middleware/logging.go b/internal/middleware/logging.go index 694731f..c1a6a09 100644 --- a/internal/middleware/logging.go +++ b/internal/middleware/logging.go @@ -4,6 +4,8 @@ import ( "log/slog" "net/http" "time" + + "github.com/containeroo/heartbeats/internal/logging" ) // LoggingMiddleware wraps the handler, logs the request and response, and writes the response to the client. @@ -19,7 +21,7 @@ func LoggingMiddleware(logger *slog.Logger) Middleware { duration := time.Since(startTime) // Measure total execution time // Log request details - logger.Debug("HTTP request", + logging.AccessLogger(logger, r.Context()).Debug("HTTP request", "method", r.Method, "url_path", r.URL.Path, "status_code", rec.statusCode, diff --git a/internal/middleware/request_id.go b/internal/middleware/request_id.go new file mode 100644 index 0000000..ec1539e --- /dev/null +++ b/internal/middleware/request_id.go @@ -0,0 +1,24 @@ +package middleware + +import ( + "net/http" + + "github.com/containeroo/heartbeats/internal/logging" +) + +// RequestIDMiddleware ensures each request has a request_id in context and response headers. +func RequestIDMiddleware() Middleware { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestID := r.Header.Get(logging.RequestIDHeader) + if requestID == "" { + requestID = logging.NewRequestID() + } + if requestID != "" { + r = r.WithContext(logging.WithRequestID(r.Context(), requestID)) + w.Header().Set(logging.RequestIDHeader, requestID) + } + next.ServeHTTP(w, r) + }) + } +} diff --git a/internal/notifier/dispatcher.go b/internal/notifier/dispatcher.go index d141ec3..928f41a 100644 --- a/internal/notifier/dispatcher.go +++ b/internal/notifier/dispatcher.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/metrics" servicehistory "github.com/containeroo/heartbeats/internal/service/history" ) @@ -123,7 +124,7 @@ func (d *Dispatcher) sendWithRetry(ctx context.Context, receiverID string, n Not if err != nil { payload.Error = err.Error() - d.logger.Error("notification format error", + logging.SystemLogger(d.logger, nil).Error("notification format error", "receiver", receiverID, "type", n.Type(), "target", n.Target(), @@ -134,7 +135,7 @@ func (d *Dispatcher) sendWithRetry(ctx context.Context, receiverID string, n Not ev := factory.NotificationFailed(data.ID, payload.Receiver, payload.Type, payload.Target, payload.Error) if err := d.history.Append(ctx, ev); err != nil { - d.logger.Error("failed to record state change", "err", err) + logging.SystemLogger(d.logger, nil).Error("failed to record state change", "err", err) } return } @@ -146,7 +147,7 @@ func (d *Dispatcher) sendWithRetry(ctx context.Context, receiverID string, n Not receiverStatus = metrics.ERROR - d.logger.Error("notification error", + logging.SystemLogger(d.logger, nil).Error("notification error", "receiver", receiverID, "type", n.Type(), "target", n.Target(), @@ -163,7 +164,7 @@ func (d *Dispatcher) sendWithRetry(ctx context.Context, receiverID string, n Not ev = factory.NotificationSent(data.ID, payload.Receiver, payload.Type, payload.Target) } if err := d.history.Append(ctx, ev); err != nil { - d.logger.Error("failed to record state change", "err", err) + logging.SystemLogger(d.logger, nil).Error("failed to record state change", "err", err) } } @@ -181,7 +182,7 @@ func (d *Dispatcher) retryNotify( return nil // success } - d.logger.Debug("retrying", "attempt", i+1, "receiver", receiverID) + logging.SystemLogger(d.logger, nil).Debug("retrying", "attempt", i+1, "receiver", receiverID) // Wait unless it's the last attempt if i < d.retries-1 { select { diff --git a/internal/notifier/dispatcher_test.go b/internal/notifier/dispatcher_test.go index 811a702..a3078c9 100644 --- a/internal/notifier/dispatcher_test.go +++ b/internal/notifier/dispatcher_test.go @@ -115,7 +115,7 @@ func TestDispatcher_Dispatch(t *testing.T) { // Let it settle time.Sleep(10 * time.Millisecond) - assert.Contains(t, buf.String(), "level=ERROR msg=\"failed to record state change\" err=fail!\n") + assert.Contains(t, buf.String(), "level=ERROR msg=\"failed to record state change\" category=system err=fail!\n") }) } diff --git a/internal/notifier/msteams.go b/internal/notifier/msteams.go index 7e9b60c..eca0889 100644 --- a/internal/notifier/msteams.go +++ b/internal/notifier/msteams.go @@ -8,6 +8,7 @@ import ( "net/url" "time" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/pkg/notify/msteams" "github.com/containeroo/resolver" @@ -76,7 +77,7 @@ func (m *MSTeamsConfig) Notify(ctx context.Context, data NotificationData) error return fmt.Errorf("cannot send MSTeams notification. %w", err) } - m.logger.Info("Notification sent", + logging.BusinessLogger(m.logger, nil).Info("Notification sent", "receiver", m.id, "type", m.Type(), "webhook_url", m.Target(), diff --git a/internal/notifier/msteamsgraph.go b/internal/notifier/msteamsgraph.go index 72f2cd5..ce655df 100644 --- a/internal/notifier/msteamsgraph.go +++ b/internal/notifier/msteamsgraph.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/pkg/notify/msteamsgraph" "github.com/containeroo/resolver" ) @@ -84,7 +85,7 @@ func (m *MSTeamsGraphConfig) Notify(ctx context.Context, data NotificationData) return fmt.Errorf("send MS Teams message: %w", err) } - m.logger.Info("Notification sent", + logging.BusinessLogger(m.logger, nil).Info("Notification sent", "receiver", m.id, "type", m.Type(), "team", MaskedTail(m.TeamID, 4), diff --git a/internal/notifier/slack.go b/internal/notifier/slack.go index c368d57..e247be4 100644 --- a/internal/notifier/slack.go +++ b/internal/notifier/slack.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/pkg/notify/slack" "github.com/containeroo/resolver" ) @@ -90,7 +91,7 @@ func (s *SlackConfig) Notify(ctx context.Context, data NotificationData) error { return fmt.Errorf("send Slack notification: %w", err) } - s.logger.Info("Notification sent", + logging.BusinessLogger(s.logger, nil).Info("Notification sent", "receiver", s.id, "type", s.Type(), "channel", s.Target(), diff --git a/internal/server/routes.go b/internal/server/routes.go index 174c89e..35fef63 100644 --- a/internal/server/routes.go +++ b/internal/server/routes.go @@ -1,11 +1,12 @@ package server import ( + "fmt" "io/fs" - "log/slog" "net/http" "github.com/containeroo/heartbeats/internal/handler" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/middleware" "github.com/containeroo/heartbeats/internal/service/health" ) @@ -13,13 +14,18 @@ import ( // NewRouter creates a new HTTP router func NewRouter( webFS fs.FS, + routePrefix string, api *handler.API, - logger *slog.Logger, -) http.Handler { + debug bool, +) (http.Handler, error) { root := http.NewServeMux() // Handler for embedded static files - staticContent, _ := fs.Sub(webFS, "web/static") + staticContent, err := fs.Sub(webFS, "web/static") + if err != nil { + return nil, fmt.Errorf("web filesystem: %w", err) + } + fileServer := http.FileServer(http.FS(staticContent)) root.Handle("GET /static/", http.StripPrefix("/static/", fileServer)) @@ -35,16 +41,20 @@ func NewRouter( root.Handle("POST /bump/{id}/fail", api.FailHandler()) root.Handle("GET /bump/{id}/fail", api.FailHandler()) - // Mount the whole app under the prefix if provided - var handler http.Handler = root - if api.RoutePrefix != "" { - handler = mountUnderPrefix(root, api.RoutePrefix) + var h http.Handler = root + if routePrefix != "" { + logging.SystemLogger(api.Logger, nil).Info( + "mounted under prefix", + "event", "routes_mounted", + "prefix", routePrefix, + ) + h = mountUnderPrefix(h, routePrefix) } - // wrap the whole mux in logging if debug - if api.Debug { - return middleware.Chain(handler, middleware.LoggingMiddleware(logger)) + // Optional debug logging middleware. + if debug { + return middleware.Chain(h, middleware.LoggingMiddleware(api.Logger), middleware.RequestIDMiddleware()), nil } - return handler + return middleware.Chain(h, middleware.RequestIDMiddleware()), nil } diff --git a/internal/server/routes_test.go b/internal/server/routes_test.go index acde0b4..87b7a55 100644 --- a/internal/server/routes_test.go +++ b/internal/server/routes_test.go @@ -82,7 +82,8 @@ func TestNewRouter(t *testing.T) { metricsReg, nil, ) - router := NewRouter(webFS, api, logger) + router, err := NewRouter(webFS, "", api, true) + assert.NoError(t, err) t.Run("GET /", func(t *testing.T) { t.Parallel() diff --git a/internal/server/server.go b/internal/server/server.go index fc2d17e..45d9e3f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -6,6 +6,8 @@ import ( "net/http" "sync" "time" + + "github.com/containeroo/heartbeats/internal/logging" ) // Run sets up and manages the reverse proxy HTTP server. @@ -21,9 +23,9 @@ func Run(ctx context.Context, listenAddr string, router http.Handler, logger *sl // Start the server in the background. go func() { - logger.Info("starting server", "listenAddr", server.Addr) + logging.SystemLogger(logger, nil).Info("starting server", "listenAddr", server.Addr) if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { - logger.Error("server error", "err", err) + logging.SystemLogger(logger, nil).Error("server error", "err", err) } }() @@ -32,14 +34,14 @@ func Run(ctx context.Context, listenAddr string, router http.Handler, logger *sl wg.Go(func() { <-ctx.Done() // wait for cancel/timeout - logger.Info("shutting down server") + logging.SystemLogger(logger, nil).Info("shutting down server") // Use a bounded timeout to finish in-flight requests. shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := server.Shutdown(shutdownCtx); err != nil { - logger.Error("shutdown error", "err", err) + logging.SystemLogger(logger, nil).Error("shutdown error", "err", err) } }) diff --git a/internal/service/bump/bump.go b/internal/service/bump/bump.go index 2110e14..eb8feb3 100644 --- a/internal/service/bump/bump.go +++ b/internal/service/bump/bump.go @@ -7,6 +7,7 @@ import ( "log/slog" "github.com/containeroo/heartbeats/internal/heartbeat" + "github.com/containeroo/heartbeats/internal/logging" servicehistory "github.com/containeroo/heartbeats/internal/service/history" ) @@ -29,13 +30,13 @@ func Receive( return fmt.Errorf("%w: %s", ErrUnknownHeartbeat, id) } - logger.Info("received bump", "id", id, "from", source) + logging.BusinessLogger(logger, ctx).Info("received bump", "id", id, "from", source) factory := servicehistory.NewFactory() ev := factory.HeartbeatReceived(id, source, method, userAgent) if err := hist.Append(ctx, ev); err != nil { - logger.Error("failed to record state change", "err", err) + logging.BusinessLogger(logger, ctx).Error("failed to record state change", "err", err) return err } @@ -61,13 +62,13 @@ func Fail( return fmt.Errorf("%w: %s", ErrUnknownHeartbeat, id) } - logger.Info("manual fail", "id", id, "from", source) + logging.BusinessLogger(logger, ctx).Info("manual fail", "id", id, "from", source) factory := servicehistory.NewFactory() ev := factory.HeartbeatFailed(id, source, method, userAgent) if err := hist.Append(ctx, ev); err != nil { - logger.Error("failed to record state change", "err", err) + logging.BusinessLogger(logger, ctx).Error("failed to record state change", "err", err) return err } diff --git a/internal/service/test/test.go b/internal/service/test/test.go index 9fab073..2c2173a 100644 --- a/internal/service/test/test.go +++ b/internal/service/test/test.go @@ -6,12 +6,13 @@ import ( "time" "github.com/containeroo/heartbeats/internal/heartbeat" + "github.com/containeroo/heartbeats/internal/logging" "github.com/containeroo/heartbeats/internal/notifier" ) // SendTestNotification sends a test notification to a specific receiver. func SendTestNotification(dispatcher *notifier.Dispatcher, logger *slog.Logger, id string) { - logger.Info("Test request received", "receiver", id) + logging.BusinessLogger(logger, nil).Info("Test request received", "receiver", id) dispatcher.Mailbox() <- notifier.NotificationData{ ID: fmt.Sprintf("manual-test-%s", time.Now().Format(time.RFC3339)), @@ -23,6 +24,6 @@ func SendTestNotification(dispatcher *notifier.Dispatcher, logger *slog.Logger, // TriggerTestHeartbeat sends a test notification for a specific heartbeat. func TriggerTestHeartbeat(mgr *heartbeat.Manager, logger *slog.Logger, id string) error { - logger.Info("Test request heartbeat", "heartbeat", id) + logging.BusinessLogger(logger, nil).Info("Test request heartbeat", "heartbeat", id) return mgr.Test(id) } diff --git a/pkg/notify/README.md b/pkg/notify/README.md index bda58b5..4e39024 100644 --- a/pkg/notify/README.md +++ b/pkg/notify/README.md @@ -10,6 +10,8 @@ The `notify` package provides pluggable implementations for sending notification - Incoming webhook - Graph API (channel messages) - Configurable headers, TLS settings, and authentication +- Typed error semantics for transient vs permanent failures +- Shared HTTP client defaults and per-request timeouts ## Package Overview @@ -28,6 +30,24 @@ The `notify` package provides pluggable implementations for sending notification - [`msteams`](./msteams.go) – Sends cards via Microsoft Teams webhook - [`msteamsgraph`](./msteamspgrahapi.go) – Sends rich messages using Graph API - [`utils`](./utils.go) – Shared HTTP client abstraction +- [`utils/errors.go`](./utils/errors.go) – Typed error helpers (`transient` vs `permanent`) + +## Error Semantics + +All HTTP-based notifiers return typed errors via `utils.Wrap`: + +- `transient`: safe to retry (timeouts, 5xx, 429, 408, network errors) +- `permanent`: do not retry (4xx, invalid payloads, auth errors) + +Use `utils.IsTransient(err)` to decide whether to retry. + +## Defaults & Timeouts + +HTTP clients share a default per-request timeout of 10s. You can override it via: + +- `slack.WithTimeout(...)` +- `msteams.WithTimeout(...)` +- `msteamsgraph.WithTimeout(...)` ## Usage diff --git a/pkg/notify/email/email.go b/pkg/notify/email/email.go index e1677b1..42db6ef 100644 --- a/pkg/notify/email/email.go +++ b/pkg/notify/email/email.go @@ -5,8 +5,12 @@ import ( "crypto/tls" "fmt" "io" + "net" "net/smtp" "strings" + "time" + + "github.com/containeroo/heartbeats/pkg/notify/utils" ) // Sender defines an interface for sending email messages. @@ -35,6 +39,11 @@ type Dialer interface { Dial(addr string) (Client, error) } +// ContextDialer establishes SMTP connections with a context. +type ContextDialer interface { + DialContext(ctx context.Context, addr string) (Client, error) +} + // Client represents a low-level connection to an SMTP server. type Client interface { Mail(from string) error @@ -57,7 +66,19 @@ type DefaultDialer struct{} // - Client: An active SMTP connection. // - error: If the connection fails. func (DefaultDialer) Dial(addr string) (Client, error) { - return smtp.Dial(addr) + return dialSMTP(context.Background(), addr, utils.DefaultTimeout) +} + +// DialContext connects to the SMTP server with a context. +func (DefaultDialer) DialContext(ctx context.Context, addr string) (Client, error) { + timeout := utils.DefaultTimeout + if deadline, ok := ctx.Deadline(); ok { + timeout = time.Until(deadline) + if timeout <= 0 { + return nil, ctx.Err() + } + } + return dialSMTP(ctx, addr, timeout) } // MailClient sends email messages using SMTP. @@ -92,9 +113,15 @@ func (c *MailClient) Send(ctx context.Context, msg Message) error { raw := buildMIMEMessage(msg, c.SMTPConfig.From) addr := fmt.Sprintf("%s:%d", c.SMTPConfig.Host, c.SMTPConfig.Port) - conn, err := c.Dialer.Dial(addr) + var conn Client + var err error + if d, ok := c.Dialer.(ContextDialer); ok { + conn, err = d.DialContext(ctx, addr) + } else { + conn, err = c.Dialer.Dial(addr) + } if err != nil { - return fmt.Errorf("connect SMTP: %w", err) + return utils.Wrap(utils.ErrorTransient, "smtp connect", err) } defer conn.Close() // nolint:errcheck @@ -104,39 +131,52 @@ func (c *MailClient) Send(ctx context.Context, msg Message) error { InsecureSkipVerify: c.SMTPConfig.SkipInsecureVerify != nil && *c.SMTPConfig.SkipInsecureVerify, } if err := conn.StartTLS(tlsConfig); err != nil { - return fmt.Errorf("start TLS: %w", err) + return utils.Wrap(utils.ErrorTransient, "smtp starttls", err) } } if c.SMTPConfig.Username != "" { auth := smtp.PlainAuth("", c.SMTPConfig.Username, c.SMTPConfig.Password, c.SMTPConfig.Host) if err := conn.Auth(auth); err != nil { - return fmt.Errorf("SMTP auth: %w", err) + return utils.Wrap(utils.ErrorPermanent, "smtp auth", err) } } if err := conn.Mail(c.SMTPConfig.From); err != nil { - return fmt.Errorf("MAIL FROM: %w", err) + return utils.Wrap(utils.ErrorPermanent, "smtp mail from", err) } for _, to := range msg.To { if err := conn.Rcpt(to); err != nil { - return fmt.Errorf("RCPT TO (%s): %w", to, err) + return utils.Wrap(utils.ErrorPermanent, "smtp rcpt", fmt.Errorf("%s: %w", to, err)) } } wc, err := conn.Data() if err != nil { - return fmt.Errorf("DATA open: %w", err) + return utils.Wrap(utils.ErrorTransient, "smtp data", err) } defer wc.Close() // nolint:errcheck if _, err := wc.Write(raw); err != nil { - return fmt.Errorf("write body: %w", err) + return utils.Wrap(utils.ErrorTransient, "smtp write", err) } return nil } +func dialSMTP(ctx context.Context, addr string, timeout time.Duration) (Client, error) { + host, _, err := net.SplitHostPort(addr) + if err != nil { + return nil, err + } + d := net.Dialer{Timeout: timeout} + netConn, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return nil, err + } + return smtp.NewClient(netConn, host) +} + // SMTPConfig contains connection and authentication settings for an SMTP server. type SMTPConfig struct { Host string `yaml:"host,omitempty"` // SMTP server hostname. @@ -187,24 +227,24 @@ func buildMIMEMessage(msg Message, from string) []byte { var buf strings.Builder for k, v := range headers { - buf.WriteString(fmt.Sprintf("%s: %s\r\n", k, v)) + fmt.Fprintf(&buf, "%s: %s\r\n", k, v) } - buf.WriteString(fmt.Sprintf("\r\n--%s\r\n", boundary)) + fmt.Fprintf(&buf, "\r\n--%s\r\n", boundary) if msg.IsHTML { - buf.WriteString("Content-Type: text/html; charset=\"UTF-8\"\r\n\r\n") + fmt.Fprintf(&buf, "Content-Type: text/html; charset=\"UTF-8\"\r\n\r\n") } else { - buf.WriteString("Content-Type: text/plain; charset=\"UTF-8\"\r\n\r\n") + fmt.Fprintf(&buf, "Content-Type: text/plain; charset=\"UTF-8\"\r\n\r\n") } buf.WriteString(msg.Body + "\r\n") for _, att := range msg.Attachments { - buf.WriteString(fmt.Sprintf("\r\n--%s\r\n", boundary)) - buf.WriteString(fmt.Sprintf("Content-Type: application/octet-stream; name=\"%s\"\r\n", att.Filename)) - buf.WriteString("Content-Transfer-Encoding: base64\r\n\r\n") - buf.WriteString(string(att.Data) + "\r\n") + fmt.Fprintf(&buf, "\r\n--%s\r\n", boundary) + fmt.Fprintf(&buf, "Content-Type: application/octet-stream; name=\"%s\"\r\n", att.Filename) + fmt.Fprintf(&buf, "Content-Transfer-Encoding: base64\r\n\r\n") + fmt.Fprintf(&buf, "%s\r\n", string(att.Data)) } - buf.WriteString(fmt.Sprintf("\r\n--%s--\r\n", boundary)) + fmt.Fprintf(&buf, "\r\n--%s--\r\n", boundary) return []byte(buf.String()) } diff --git a/pkg/notify/msteams/integration_test.go b/pkg/notify/msteams/integration_test.go new file mode 100644 index 0000000..f613731 --- /dev/null +++ b/pkg/notify/msteams/integration_test.go @@ -0,0 +1,51 @@ +package msteams + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestClient_Send_Integration(t *testing.T) { + t.Parallel() + + var captured MSTeams + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() // nolint:errcheck + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &captured) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + })) + t.Cleanup(srv.Close) + + client := New() + _, err := client.Send(context.Background(), MSTeams{ + Title: "deploy", + Text: "done", + }, srv.URL) + + assert.NoError(t, err) + assert.Equal(t, "deploy", captured.Title) + assert.Equal(t, "done", captured.Text) +} + +func TestClient_Send_Integration_Non200(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte("bad")) + })) + t.Cleanup(srv.Close) + + client := New() + _, err := client.Send(context.Background(), MSTeams{}, srv.URL) + + assert.EqualError(t, err, "permanent: msteams http status: 400: bad") +} diff --git a/pkg/notify/msteams/msteams.go b/pkg/notify/msteams/msteams.go index dd76ce6..e8de333 100644 --- a/pkg/notify/msteams/msteams.go +++ b/pkg/notify/msteams/msteams.go @@ -4,9 +4,9 @@ import ( "context" "encoding/json" "fmt" - "io" "maps" "net/http" + "time" "github.com/containeroo/heartbeats/pkg/notify/utils" ) @@ -53,6 +53,15 @@ func WithInsecureTLS(skipInsecure bool) Option { } } +// WithTimeout sets the per-request timeout. +func WithTimeout(timeout time.Duration) Option { + return func(c *Client) { + if hc, ok := c.HttpClient.(*utils.HttpClient); ok { + hc.Timeout = timeout + } + } +} + // New creates a new MS Teams client with functional options. // // Use options like WithHeaders or WithInsecureTLS to customize behavior. @@ -80,22 +89,23 @@ func (c *Client) Send(ctx context.Context, message MSTeams, webhookURL string) ( // Serialize the message to JSON. data, err := json.Marshal(message) if err != nil { - return "", fmt.Errorf("error marshalling MS Teams message: %w", err) + return "", utils.Wrap(utils.ErrorPermanent, "msteams marshal", err) } // Execute the POST request using the configured HTTP client. resp, err := c.HttpClient.DoRequest(ctx, "POST", webhookURL, data) if err != nil { - return "", fmt.Errorf("error sending HTTP request: %w", err) + return "", utils.Wrap(utils.ErrorTransient, "msteams request", err) } defer resp.Body.Close() // nolint:errcheck // Read response body regardless of status - body, _ := io.ReadAll(resp.Body) + body, _ := utils.ReadBodyLimited(resp.Body) + bodyStr := utils.RedactSecrets(string(body)) // Verify the HTTP status code is 200 OK. if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("received non-200 response: %d, body: %s", resp.StatusCode, string(body)) + return "", utils.Wrap(utils.KindFromStatus(resp.StatusCode), "msteams http status", fmt.Errorf("%d: %s", resp.StatusCode, bodyStr)) } return "Message sent successfully", nil diff --git a/pkg/notify/msteams/msteams_test.go b/pkg/notify/msteams/msteams_test.go index 6386dfc..e864ce7 100644 --- a/pkg/notify/msteams/msteams_test.go +++ b/pkg/notify/msteams/msteams_test.go @@ -55,7 +55,7 @@ func TestClient_Send_RequestFailure(t *testing.T) { _, err := client.Send(context.Background(), MSTeams{}, "https://webhook") assert.Error(t, err) - assert.EqualError(t, err, "error sending HTTP request: unexpected EOF") + assert.EqualError(t, err, "transient: msteams request: unexpected EOF") } func TestClient_Send_Non200Status(t *testing.T) { @@ -72,5 +72,5 @@ func TestClient_Send_Non200Status(t *testing.T) { _, err := client.Send(context.Background(), MSTeams{}, "https://webhook") assert.Error(t, err) - assert.EqualError(t, err, "received non-200 response: 403, body: forbidden") + assert.EqualError(t, err, "permanent: msteams http status: 403: forbidden") } diff --git a/pkg/notify/msteamsgraph/integration_test.go b/pkg/notify/msteamsgraph/integration_test.go new file mode 100644 index 0000000..6c53184 --- /dev/null +++ b/pkg/notify/msteamsgraph/integration_test.go @@ -0,0 +1,50 @@ +package msteamsgraph + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestClient_SendChannel_Integration(t *testing.T) { + t.Parallel() + + var captured Message + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() // nolint:errcheck + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &captured) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"123"}`)) + })) + t.Cleanup(srv.Close) + + client := New(WithEndpointBase(srv.URL)) + _, err := client.SendChannel(context.Background(), "team", "channel", Message{ + Body: ItemBody{ContentType: "html", Content: "hi"}, + }) + + assert.NoError(t, err) + assert.Equal(t, "html", captured.Body.ContentType) + assert.Equal(t, "hi", captured.Body.Content) +} + +func TestClient_SendChannel_Integration_Non200(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":{"code":"BadRequest","message":"nope"}}`)) + })) + t.Cleanup(srv.Close) + + client := New(WithEndpointBase(srv.URL)) + _, err := client.SendChannel(context.Background(), "team", "channel", Message{}) + + assert.EqualError(t, err, "permanent: msteamsgraph http status: BadRequest: nope") +} diff --git a/pkg/notify/msteamsgraph/msteamsgraph.go b/pkg/notify/msteamsgraph/msteamsgraph.go index af45aee..8c9ad5a 100644 --- a/pkg/notify/msteamsgraph/msteamsgraph.go +++ b/pkg/notify/msteamsgraph/msteamsgraph.go @@ -4,12 +4,14 @@ import ( "context" "encoding/json" "fmt" + "io" "maps" + "time" "github.com/containeroo/heartbeats/pkg/notify/utils" ) -const channelEndpoint string = "https://graph.microsoft.com/v1.0/teams/%s/channels/%s/messages" +const defaultEndpointBase string = "https://graph.microsoft.com/v1.0" // Message represents the payload structure for a Teams message. type Message struct { @@ -45,6 +47,7 @@ type Sender interface { // Client posts messages using the Microsoft Graph API. type Client struct { HttpClient utils.HTTPDoer // HTTP client for sending requests + Endpoint string // Endpoint base URL for the Graph API } // Option configures the Teams client. @@ -69,10 +72,26 @@ func WithInsecureTLS(skipInsecure bool) Option { } } +// WithEndpointBase overrides the default Graph API base URL. +func WithEndpointBase(endpoint string) Option { + return func(c *Client) { + c.Endpoint = endpoint + } +} + +// WithTimeout sets the per-request timeout. +func WithTimeout(timeout time.Duration) Option { + return func(c *Client) { + hc := c.HttpClient.(*utils.HttpClient) + hc.Timeout = timeout + } +} + // New returns a new Client with optional configuration. func New(opts ...Option) *Client { c := &Client{ HttpClient: utils.NewHttpClient(make(map[string]string), false), + Endpoint: defaultEndpointBase, } for _, opt := range opts { opt(c) @@ -94,7 +113,11 @@ func NewWithToken(token string, opts ...Option) *Client { // teamID: ID of the team. // channelID: ID of the channel inside the team. func (c *Client) SendChannel(ctx context.Context, teamID, channelID string, msg Message) (*Response, error) { - endpoint := fmt.Sprintf(channelEndpoint, teamID, channelID) + base := c.Endpoint + if base == "" { + base = defaultEndpointBase + } + endpoint := fmt.Sprintf("%s/teams/%s/channels/%s/messages", base, teamID, channelID) return c.send(ctx, endpoint, msg) } @@ -102,22 +125,22 @@ func (c *Client) SendChannel(ctx context.Context, teamID, channelID string, msg func (c *Client) send(ctx context.Context, url string, msg Message) (*Response, error) { data, err := json.Marshal(msg) if err != nil { - return nil, fmt.Errorf("marshal error: %w", err) + return nil, utils.Wrap(utils.ErrorPermanent, "msteamsgraph marshal", err) } resp, err := c.HttpClient.DoRequest(ctx, "POST", url, data) if err != nil { - return nil, fmt.Errorf("request failed: %w", err) + return nil, utils.Wrap(utils.ErrorTransient, "msteamsgraph request", err) } defer resp.Body.Close() // nolint:errcheck var parsed Response - if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { - return nil, fmt.Errorf("decode error: %w", err) + if err := json.NewDecoder(io.LimitReader(resp.Body, utils.MaxResponseBody)).Decode(&parsed); err != nil { + return nil, utils.Wrap(utils.ErrorPermanent, "msteamsgraph decode", err) } if resp.StatusCode >= 400 { - return &parsed, fmt.Errorf("teams graph error: %s", parsed.ErrorOrMessage()) + return &parsed, utils.Wrap(utils.KindFromStatus(resp.StatusCode), "msteamsgraph http status", fmt.Errorf("%s", parsed.ErrorOrMessage())) } return &parsed, nil diff --git a/pkg/notify/msteamsgraph/msteamsgraph_test.go b/pkg/notify/msteamsgraph/msteamsgraph_test.go index a22dec7..59dbaad 100644 --- a/pkg/notify/msteamsgraph/msteamsgraph_test.go +++ b/pkg/notify/msteamsgraph/msteamsgraph_test.go @@ -75,7 +75,7 @@ func TestClient_SendChannel_GraphError(t *testing.T) { _, err := client.SendChannel(context.Background(), "team", "channel", Message{}) assert.Error(t, err) - assert.EqualError(t, err, "teams graph error: InvalidAuthToken: Access denied.") + assert.EqualError(t, err, "permanent: msteamsgraph http status: InvalidAuthToken: Access denied.") } func TestClient_SendChannel_DecodeError(t *testing.T) { @@ -92,7 +92,7 @@ func TestClient_SendChannel_DecodeError(t *testing.T) { _, err := client.SendChannel(context.Background(), "team", "channel", Message{}) assert.Error(t, err) - assert.EqualError(t, err, "decode error: invalid character 'i' looking for beginning of object key string") + assert.EqualError(t, err, "permanent: msteamsgraph decode: invalid character 'i' looking for beginning of object key string") } func TestClient_SendChannel_RequestError(t *testing.T) { @@ -106,5 +106,5 @@ func TestClient_SendChannel_RequestError(t *testing.T) { _, err := client.SendChannel(context.Background(), "team", "channel", Message{}) assert.Error(t, err) - assert.EqualError(t, err, "request failed: assert.AnError general error for testing") + assert.EqualError(t, err, "transient: msteamsgraph request: assert.AnError general error for testing") } diff --git a/pkg/notify/slack/integration_test.go b/pkg/notify/slack/integration_test.go new file mode 100644 index 0000000..c657b18 --- /dev/null +++ b/pkg/notify/slack/integration_test.go @@ -0,0 +1,59 @@ +package slack + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestClient_Send_Integration(t *testing.T) { + t.Parallel() + + var captured Slack + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() // nolint:errcheck + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &captured) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + t.Cleanup(srv.Close) + + client := New(WithEndpoint(srv.URL)) + _, err := client.Send(context.Background(), Slack{ + Channel: "#alerts", + Attachments: []Attachment{{ + Color: "good", + Title: "ok", + Text: "hello", + }}, + }) + + assert.NoError(t, err) + assert.Equal(t, "#alerts", captured.Channel) + if assert.Len(t, captured.Attachments, 1) { + assert.Equal(t, "ok", captured.Attachments[0].Title) + assert.Equal(t, "hello", captured.Attachments[0].Text) + } +} + +func TestClient_Send_Integration_Non200(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTeapot) + _, _ = io.Copy(w, bytes.NewBufferString("nope")) + })) + t.Cleanup(srv.Close) + + client := New(WithEndpoint(srv.URL)) + _, err := client.Send(context.Background(), Slack{}) + + assert.EqualError(t, err, "permanent: slack http status: 418") +} diff --git a/pkg/notify/slack/slack.go b/pkg/notify/slack/slack.go index 58220b1..ddb050d 100644 --- a/pkg/notify/slack/slack.go +++ b/pkg/notify/slack/slack.go @@ -4,8 +4,10 @@ import ( "context" "encoding/json" "fmt" + "io" "maps" "net/http" + "time" "github.com/containeroo/heartbeats/pkg/notify/utils" ) @@ -48,6 +50,7 @@ type Sender interface { // Client sends Slack messages using the official chat.postMessage API. type Client struct { HttpClient utils.HTTPDoer // HttpClient is used to send HTTP requests (mockable for testing). + Endpoint string // Endpoint is the Slack API endpoint to use. } // Option configures a Slack client. @@ -71,6 +74,20 @@ func WithInsecureTLS(skipInsecure bool) Option { } } +// WithEndpoint overrides the Slack API endpoint. +func WithEndpoint(endpoint string) Option { + return func(c *Client) { + c.Endpoint = endpoint + } +} + +// WithTimeout sets the per-request timeout. +func WithTimeout(timeout time.Duration) Option { + return func(c *Client) { + c.HttpClient.(*utils.HttpClient).Timeout = timeout + } +} + // New creates a Slack API client with optional configuration. // // Options can be used to set headers or disable TLS verification. @@ -78,6 +95,7 @@ func New(opts ...Option) *Client { // start with empty headers + default TLS settings client := &Client{ HttpClient: utils.NewHttpClient(make(map[string]string), false), + Endpoint: slackAPIEndpoint, } for _, opt := range opts { @@ -114,28 +132,32 @@ func (c *Client) Send(ctx context.Context, slackMessage Slack) (*Response, error // Encode the Slack message to JSON data, err := json.Marshal(slackMessage) if err != nil { - return nil, fmt.Errorf("error marshalling Slack message: %w", err) + return nil, utils.Wrap(utils.ErrorPermanent, "slack marshal", err) } // Send the HTTP request using the injected HTTP client - resp, err := c.HttpClient.DoRequest(ctx, "POST", slackAPIEndpoint, data) + endpoint := c.Endpoint + if endpoint == "" { + endpoint = slackAPIEndpoint + } + resp, err := c.HttpClient.DoRequest(ctx, "POST", endpoint, data) if err != nil { - return nil, fmt.Errorf("error sending HTTP request: %w", err) + return nil, utils.Wrap(utils.ErrorTransient, "slack request", err) } defer resp.Body.Close() // nolint:errcheck // Slack returns 200 OK even for some failures—check explicitly if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("received non-200 response: %d", resp.StatusCode) + return nil, utils.Wrap(utils.KindFromStatus(resp.StatusCode), "slack http status", fmt.Errorf("%d", resp.StatusCode)) } var parsed Response - if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { - return nil, fmt.Errorf("error decoding response: %w", err) + if err := json.NewDecoder(io.LimitReader(resp.Body, utils.MaxResponseBody)).Decode(&parsed); err != nil { + return nil, utils.Wrap(utils.ErrorPermanent, "slack decode", err) } if !parsed.Ok { - return &parsed, fmt.Errorf("Slack API error: %s", parsed.Error) + return &parsed, utils.Wrap(utils.ErrorPermanent, "slack api", fmt.Errorf("%s", parsed.Error)) } return &parsed, nil diff --git a/pkg/notify/slack/slack_test.go b/pkg/notify/slack/slack_test.go index dea5a58..65ffd41 100644 --- a/pkg/notify/slack/slack_test.go +++ b/pkg/notify/slack/slack_test.go @@ -78,7 +78,7 @@ func TestClient_Send_Non200(t *testing.T) { _, err := client.Send(context.Background(), Slack{}) assert.Error(t, err) - assert.EqualError(t, err, "received non-200 response: 403") + assert.EqualError(t, err, "permanent: slack http status: 403") } func TestClient_Send_ErrorDecoding(t *testing.T) { @@ -95,7 +95,7 @@ func TestClient_Send_ErrorDecoding(t *testing.T) { _, err := client.Send(context.Background(), Slack{}) assert.Error(t, err) - assert.EqualError(t, err, "error decoding response: invalid character 'i' looking for beginning of object key string") + assert.EqualError(t, err, "permanent: slack decode: invalid character 'i' looking for beginning of object key string") } func TestClient_Send_ErrorAPIResponse(t *testing.T) { @@ -112,5 +112,5 @@ func TestClient_Send_ErrorAPIResponse(t *testing.T) { _, err := client.Send(context.Background(), Slack{}) assert.Error(t, err) - assert.EqualError(t, err, "Slack API error: invalid_auth") + assert.EqualError(t, err, "permanent: slack api: invalid_auth") } diff --git a/pkg/notify/utils/errors.go b/pkg/notify/utils/errors.go new file mode 100644 index 0000000..525068e --- /dev/null +++ b/pkg/notify/utils/errors.go @@ -0,0 +1,57 @@ +package utils + +import ( + "errors" + "fmt" +) + +// ErrorKind identifies whether an error is transient or permanent. +type ErrorKind string + +const ( + ErrorTransient ErrorKind = "transient" + ErrorPermanent ErrorKind = "permanent" +) + +// Error wraps an error with a kind and operation. +type Error struct { + Kind ErrorKind + Op string + Err error +} + +func (e *Error) Error() string { + if e.Op == "" { + return fmt.Sprintf("%s: %v", e.Kind, e.Err) + } + return fmt.Sprintf("%s: %s: %v", e.Kind, e.Op, e.Err) +} + +func (e *Error) Unwrap() error { return e.Err } + +// Wrap annotates err with kind and op. +func Wrap(kind ErrorKind, op string, err error) error { + if err == nil { + return nil + } + return &Error{Kind: kind, Op: op, Err: err} +} + +// IsTransient returns true if err is a transient error. +func IsTransient(err error) bool { + var e *Error + if errors.As(err, &e) { + return e.Kind == ErrorTransient + } + return false +} + +// KindFromStatus maps HTTP status to error kind. +func KindFromStatus(status int) ErrorKind { + switch { + case status == 408 || status == 429 || status >= 500: + return ErrorTransient + default: + return ErrorPermanent + } +} diff --git a/pkg/notify/utils/response.go b/pkg/notify/utils/response.go new file mode 100644 index 0000000..b9e443a --- /dev/null +++ b/pkg/notify/utils/response.go @@ -0,0 +1,46 @@ +package utils + +import ( + "io" + "net/url" + "regexp" + "strings" +) + +// MaxResponseBody limits how much of a response body is read for errors/decoding. +const MaxResponseBody = 64 * 1024 + +// ReadBodyLimited reads at most MaxResponseBody bytes from r. +func ReadBodyLimited(r io.Reader) ([]byte, error) { + return io.ReadAll(io.LimitReader(r, MaxResponseBody)) +} + +var bearerRE = regexp.MustCompile(`(?i)Bearer\s+[^\s"']+`) + +// RedactSecrets removes obvious secrets (tokens, query params) from a string. +func RedactSecrets(s string) string { + s = bearerRE.ReplaceAllString(s, "Bearer ****") + return redactURLs(s) +} + +// redactURLs removes user info and query params from URLs. +func redactURLs(s string) string { + parts := strings.Fields(s) + if len(parts) == 0 { + return s + } + for i, part := range parts { + if !strings.HasPrefix(part, "http://") && !strings.HasPrefix(part, "https://") { + continue + } + u, err := url.Parse(part) + if err != nil { + continue + } + u.User = nil + u.RawQuery = "" + u.Fragment = "" + parts[i] = u.String() + } + return strings.Join(parts, " ") +} diff --git a/pkg/notify/utils/response_test.go b/pkg/notify/utils/response_test.go new file mode 100644 index 0000000..6ed9ebd --- /dev/null +++ b/pkg/notify/utils/response_test.go @@ -0,0 +1,23 @@ +package utils + +import ( + "strings" + "testing" +) + +func TestRedactSecrets(t *testing.T) { + t.Parallel() + + input := "Bearer abc123 https://example.com/path?token=secret#frag ok" + got := RedactSecrets(input) + + if got == input { + t.Fatalf("expected redaction, got unchanged: %q", got) + } + if strings.Contains(got, "abc123") { + t.Fatalf("expected token to be redacted, got: %q", got) + } + if strings.Contains(got, "token=secret") { + t.Fatalf("expected query to be stripped, got: %q", got) + } +} diff --git a/pkg/notify/utils/utils.go b/pkg/notify/utils/utils.go index b4b92d5..db6edbf 100644 --- a/pkg/notify/utils/utils.go +++ b/pkg/notify/utils/utils.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "fmt" "net/http" + "time" ) // HTTPDoer defines the interface required to perform an HTTP request. @@ -30,8 +31,12 @@ type HTTPDoer interface { type HttpClient struct { Headers map[string]string // Headers are added to each outbound HTTP request. SkipInsecure bool // SkipInsecure disables TLS certificate validation when true. + Timeout time.Duration // Timeout is the per-request timeout. } +// DefaultTimeout is the default per-request timeout. +const DefaultTimeout = 10 * time.Second + // NewHttpClient creates a configured HTTP client for issuing requests. // // Parameters: @@ -44,6 +49,7 @@ func NewHttpClient(headers map[string]string, skipInsecure bool) *HttpClient { return &HttpClient{ Headers: headers, SkipInsecure: skipInsecure, + Timeout: DefaultTimeout, } } @@ -61,6 +67,14 @@ func NewHttpClient(headers map[string]string, skipInsecure bool) *HttpClient { func (hc *HttpClient) DoRequest(ctx context.Context, method, url string, body []byte) (*http.Response, error) { client := hc.createHTTPClient() + if hc.Timeout > 0 { + if _, ok := ctx.Deadline(); !ok { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, hc.Timeout) + defer cancel() + } + } + // Build the HTTP request using the provided context. req, err := http.NewRequestWithContext(ctx, method, url, bytes.NewBuffer(body)) if err != nil { @@ -95,5 +109,6 @@ func (hc *HttpClient) createHTTPClient() *http.Client { return &http.Client{ Transport: transport, + Timeout: hc.Timeout, } } diff --git a/pkg/notify/utils/utils_test.go b/pkg/notify/utils/utils_test.go index a1d9a14..3d9fc0e 100644 --- a/pkg/notify/utils/utils_test.go +++ b/pkg/notify/utils/utils_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/stretchr/testify/assert" ) @@ -54,4 +55,20 @@ func TestDoRequest(t *testing.T) { assert.Error(t, err) assert.EqualError(t, err, "error creating INVALID METHOD request: net/http: invalid method \"INVALID METHOD\"") }) + + t.Run("Timeout", func(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(200 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + client := NewHttpClient(nil, false) + client.Timeout = 50 * time.Millisecond + + _, err := client.DoRequest(context.Background(), "GET", server.URL, nil) + assert.Error(t, err) + }) }