From 53a5de75e574104e2c9ea576bed06d40ec0d428e Mon Sep 17 00:00:00 2001 From: gi8 Date: Tue, 20 Jan 2026 09:13:13 +0100 Subject: [PATCH 1/2] refactor(notify): standardize errors and HTTP client defaults --- pkg/notify/README.md | 20 +++++++ pkg/notify/email/email.go | 34 +++++------ pkg/notify/msteams/integration_test.go | 51 +++++++++++++++++ pkg/notify/msteams/msteams.go | 16 +++++- pkg/notify/msteams/msteams_test.go | 4 +- pkg/notify/msteamsgraph/integration_test.go | 50 +++++++++++++++++ pkg/notify/msteamsgraph/msteamsgraph.go | 34 +++++++++-- pkg/notify/msteamsgraph/msteamsgraph_test.go | 6 +- pkg/notify/slack/integration_test.go | 59 ++++++++++++++++++++ pkg/notify/slack/slack.go | 33 +++++++++-- pkg/notify/slack/slack_test.go | 6 +- pkg/notify/utils/errors.go | 57 +++++++++++++++++++ pkg/notify/utils/utils.go | 15 +++++ 13 files changed, 346 insertions(+), 39 deletions(-) create mode 100644 pkg/notify/msteams/integration_test.go create mode 100644 pkg/notify/msteamsgraph/integration_test.go create mode 100644 pkg/notify/slack/integration_test.go create mode 100644 pkg/notify/utils/errors.go 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..03b3a60 100644 --- a/pkg/notify/email/email.go +++ b/pkg/notify/email/email.go @@ -7,6 +7,8 @@ import ( "io" "net/smtp" "strings" + + "github.com/containeroo/heartbeats/pkg/notify/utils" ) // Sender defines an interface for sending email messages. @@ -94,7 +96,7 @@ func (c *MailClient) Send(ctx context.Context, msg Message) error { 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,34 +106,34 @@ 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 @@ -187,24 +189,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..19c5d84 100644 --- a/pkg/notify/msteams/msteams.go +++ b/pkg/notify/msteams/msteams.go @@ -7,6 +7,7 @@ import ( "io" "maps" "net/http" + "time" "github.com/containeroo/heartbeats/pkg/notify/utils" ) @@ -53,6 +54,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,13 +90,13 @@ 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 @@ -95,7 +105,7 @@ func (c *Client) Send(ctx context.Context, message MSTeams, webhookURL string) ( // 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, string(body))) } 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..7cd7e2c 100644 --- a/pkg/notify/msteamsgraph/msteamsgraph.go +++ b/pkg/notify/msteamsgraph/msteamsgraph.go @@ -5,11 +5,12 @@ import ( "encoding/json" "fmt" "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 +46,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 +71,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 +112,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 +124,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) + 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..29bb5a6 100644 --- a/pkg/notify/slack/slack.go +++ b/pkg/notify/slack/slack.go @@ -6,6 +6,7 @@ import ( "fmt" "maps" "net/http" + "time" "github.com/containeroo/heartbeats/pkg/notify/utils" ) @@ -48,6 +49,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 +73,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 +94,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 +131,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) + 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/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, } } From 3277cdfd1bab3fbb20c4473666012d6453c406d3 Mon Sep 17 00:00:00 2001 From: gi8 Date: Tue, 20 Jan 2026 09:55:03 +0100 Subject: [PATCH 2/2] improve notify package --- internal/app/run_test.go | 6 ++-- pkg/notify/email/email.go | 42 ++++++++++++++++++++-- pkg/notify/msteams/msteams.go | 6 ++-- pkg/notify/msteamsgraph/msteamsgraph.go | 3 +- pkg/notify/slack/slack.go | 3 +- pkg/notify/utils/response.go | 46 +++++++++++++++++++++++++ pkg/notify/utils/response_test.go | 23 +++++++++++++ pkg/notify/utils/utils_test.go | 17 +++++++++ 8 files changed, 136 insertions(+), 10 deletions(-) create mode 100644 pkg/notify/utils/response.go create mode 100644 pkg/notify/utils/response_test.go 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/pkg/notify/email/email.go b/pkg/notify/email/email.go index 03b3a60..42db6ef 100644 --- a/pkg/notify/email/email.go +++ b/pkg/notify/email/email.go @@ -5,8 +5,10 @@ import ( "crypto/tls" "fmt" "io" + "net" "net/smtp" "strings" + "time" "github.com/containeroo/heartbeats/pkg/notify/utils" ) @@ -37,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 @@ -59,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. @@ -94,7 +113,13 @@ 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 utils.Wrap(utils.ErrorTransient, "smtp connect", err) } @@ -139,6 +164,19 @@ func (c *MailClient) Send(ctx context.Context, msg Message) error { 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. diff --git a/pkg/notify/msteams/msteams.go b/pkg/notify/msteams/msteams.go index 19c5d84..e8de333 100644 --- a/pkg/notify/msteams/msteams.go +++ b/pkg/notify/msteams/msteams.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "io" "maps" "net/http" "time" @@ -101,11 +100,12 @@ func (c *Client) Send(ctx context.Context, message MSTeams, webhookURL string) ( 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 "", utils.Wrap(utils.KindFromStatus(resp.StatusCode), "msteams http status", fmt.Errorf("%d: %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/msteamsgraph/msteamsgraph.go b/pkg/notify/msteamsgraph/msteamsgraph.go index 7cd7e2c..8c9ad5a 100644 --- a/pkg/notify/msteamsgraph/msteamsgraph.go +++ b/pkg/notify/msteamsgraph/msteamsgraph.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "io" "maps" "time" @@ -134,7 +135,7 @@ func (c *Client) send(ctx context.Context, url string, msg Message) (*Response, defer resp.Body.Close() // nolint:errcheck var parsed Response - if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { + if err := json.NewDecoder(io.LimitReader(resp.Body, utils.MaxResponseBody)).Decode(&parsed); err != nil { return nil, utils.Wrap(utils.ErrorPermanent, "msteamsgraph decode", err) } diff --git a/pkg/notify/slack/slack.go b/pkg/notify/slack/slack.go index 29bb5a6..ddb050d 100644 --- a/pkg/notify/slack/slack.go +++ b/pkg/notify/slack/slack.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "io" "maps" "net/http" "time" @@ -151,7 +152,7 @@ func (c *Client) Send(ctx context.Context, slackMessage Slack) (*Response, error } var parsed Response - if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { + if err := json.NewDecoder(io.LimitReader(resp.Body, utils.MaxResponseBody)).Decode(&parsed); err != nil { return nil, utils.Wrap(utils.ErrorPermanent, "slack decode", err) } 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_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) + }) }