diff --git a/CODEOWNERS b/CODEOWNERS index 4923660..21b2025 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -1 +1 @@ -* @brexhq/substation +* @LiveRamp/seceng diff --git a/cmd/gcp/function/substation/pubsub.go b/cmd/gcp/function/substation/pubsub.go index ae8f04b..3ab4a6f 100644 --- a/cmd/gcp/function/substation/pubsub.go +++ b/cmd/gcp/function/substation/pubsub.go @@ -78,7 +78,7 @@ func (ps *processingState) log() { } } -// nolint: gocognit // Ignore cognitive complexity. +//nolint:gocyclo,cyclop,gocognit // Ignore cyclomatic and cognitive complexity. func pubSubHandler(ctx context.Context, e cloudevents.Event) error { // Set up signal handling for graceful shutdown var state processingState diff --git a/transform/send_http_post.go b/transform/send_http_post.go index 34592b8..f4069e2 100644 --- a/transform/send_http_post.go +++ b/transform/send_http_post.go @@ -1,6 +1,7 @@ package transform import ( + "bytes" "context" "encoding/json" "fmt" @@ -19,6 +20,10 @@ import ( "github.com/brexhq/substation/v2/internal/secrets" ) +// errorBodyLimit bounds how much of a non-2xx response body is read for error +// context. Servers may echo request data, so this is deliberately small. +const errorBodyLimit = 512 + type sendHTTPPostConfig struct { // URL is the HTTP(S) endpoint that data is sent to. URL string `json:"url"` @@ -188,7 +193,11 @@ func (tf *sendHTTPPost) send(ctx context.Context, key string) error { return err } - //nolint:errcheck // Response body is discarded to avoid resource leaks. + // A bounded prefix of the body is retained so that non-2xx responses can + // explain themselves. The limit keeps payload data echoed by the server + // out of logs and errors. + body, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit)) + //nolint:errcheck // Remainder is discarded to avoid resource leaks. io.Copy(io.Discard, resp.Body) resp.Body.Close() @@ -200,6 +209,16 @@ func (tf *sendHTTPPost) send(ctx context.Context, key string) error { WithField("event_count", eventCount). WithField("duration_ms", duration.Milliseconds()). Debug("Sent HTTP POST request") + + // Responses that the HTTP client does not retry (any 4xx except 429) are + // returned with a nil error, so the status must be checked explicitly. + // Without this the batch is silently discarded. + // + // The URL is deliberately omitted: it is interpolated from secrets and may + // carry credentials, and errors propagate further than the debug log above. + if resp.StatusCode < 200 || resp.StatusCode > 299 { + return fmt.Errorf("transform %s: http post: status %d: %s", tf.conf.ID, resp.StatusCode, bytes.TrimSpace(body)) + } } return nil diff --git a/transform/send_http_post_test.go b/transform/send_http_post_test.go new file mode 100644 index 0000000..bbe5efe --- /dev/null +++ b/transform/send_http_post_test.go @@ -0,0 +1,76 @@ +package transform + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/brexhq/substation/v2/config" + "github.com/brexhq/substation/v2/message" +) + +var _ Transformer = &sendHTTPPost{} + +// Statuses that the HTTP client does not retry are returned with a nil error, +// so the transform must inspect the status code itself. Retried statuses (429 +// and 5xx) are deliberately excluded here because exercising them would incur +// the client's full backoff and exceed the test timeout. +var sendHTTPPostTests = []struct { + name string + statusCode int + body string + expectErr bool +}{ + {"200 succeeds", http.StatusOK, "", false}, + {"201 succeeds", http.StatusCreated, "", false}, + {"400 errors", http.StatusBadRequest, "invalid JSON at offset 12", true}, + {"401 errors", http.StatusUnauthorized, "invalid ingest token", true}, + {"404 errors", http.StatusNotFound, "no such endpoint", true}, +} + +func TestSendHTTPPost(t *testing.T) { + ctx := context.TODO() + + for _, test := range sendHTTPPostTests { + t.Run(test.name, func(t *testing.T) { + serv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(test.statusCode) + //nolint:errcheck // Test server write. + w.Write([]byte(test.body)) + })) + defer serv.Close() + + tf, err := newSendHTTPPost(ctx, config.Config{ + Settings: map[string]interface{}{"url": serv.URL}, + }) + if err != nil { + t.Fatal(err) + } + + if _, err := tf.Transform(ctx, message.New().SetData([]byte(`{"a":1}`))); err != nil { + t.Fatal(err) + } + + // The batch is sent when the control message is received. + _, err = tf.Transform(ctx, message.New().AsControl()) + + if !test.expectErr { + if err != nil { + t.Errorf("expected no error, got %v", err) + } + + return + } + + if err == nil { + t.Fatalf("expected an error for status %d, got nil", test.statusCode) + } + + if !strings.Contains(err.Error(), test.body) { + t.Errorf("expected error to contain response body %q, got %q", test.body, err) + } + }) + } +}