collect synthetic http requests

collector/body.go

@@ -34,6 +34,20 @@ func NewBody(rc io.ReadCloser, limit int) *Body {
 		reader: rc,
 		buffer: NewLimitedBuffer(limit),
 	}
+	if rc == nil {
+		b.isFullyCaptured = true
+	}
+	return b
+}
+
+// NewBodyFromBytes creates a Body pre-populated with the given content, for
+// callers that already hold the complete payload.
+func NewBodyFromBytes(content []byte, limit int) *Body {
+	b := &Body{
+		buffer: NewLimitedBuffer(limit),
+	}
+	b.buffer.Write(content)
+	b.isFullyCaptured = !b.buffer.IsTruncated()
 	return b
 }
 

collector/body_test.go

@@ -44,6 +44,35 @@ func TestBody_PartialRead(t *testing.T) {
 	assert.True(t, body.IsFullyCaptured())
 }
 
+func TestNewBodyFromBytes(t *testing.T) {
+	content := []byte("This is test data for a synthetic body")
+
+	body := collector.NewBodyFromBytes(content, 100)
+
+	assert.Equal(t, content, body.Bytes())
+	assert.Equal(t, string(content), body.String())
+	assert.Equal(t, uint64(len(content)), body.Size())
+	assert.False(t, body.IsTruncated())
+	assert.True(t, body.IsFullyCaptured())
+}
+
+func TestNewBodyFromBytes_Truncated(t *testing.T) {
+	content := []byte("This is test data that exceeds the configured limit")
+
+	body := collector.NewBodyFromBytes(content, 10)
+
+	assert.Equal(t, content[:10], body.Bytes())
+	assert.Equal(t, uint64(10), body.Size())
+	assert.True(t, body.IsTruncated())
+	assert.False(t, body.IsFullyCaptured())
+}
+
+func TestNewBody_NilReaderIsFullyCaptured(t *testing.T) {
+	body := collector.NewBody(nil, 100)
+
+	assert.True(t, body.IsFullyCaptured())
+}
+
 // Fix for TestBody_ReadAfterClose
 func TestBody_ReadAfterClose(t *testing.T) {
 	// Create test data

collector/http_client.go

@@ -5,6 +5,8 @@ import (
 	"net/http"
 	"strconv"
 	"time"
+
+	"github.com/gofrs/uuid"
 )
 
 // HTTPClientOptions configures the HTTP client collector
@@ -82,11 +84,49 @@ func (c *HTTPClientCollector) Subscribe(ctx context.Context) <-chan HTTPClientRe
 	return c.notifier.Subscribe(ctx)
 }
 
-// Add adds an HTTP request to the collector and notifies subscribers
+// Add notifies subscribers of an HTTP request. It does not apply transformers
+// or involve the event aggregator. For a synthetic request not performed through
+// Transport, use Collect instead.
 func (c *HTTPClientCollector) Add(req HTTPClientRequest) {
 	c.notifier.Notify(req)
 }
 
+// Collect records an HTTP client request that was not performed through the
+// transport returned by Transport, e.g. a response served from a local cache.
+//
+// In contrast to Add, it honours capture sessions and groups the request under
+// the current event group taken from ctx.
+func (c *HTTPClientCollector) Collect(ctx context.Context, req HTTPClientRequest) {
+	if c.eventAggregator != nil && !c.eventAggregator.ShouldCapture(ctx) {
+		return
+	}
+
+	if req.ID == uuid.Nil {
+		req.ID = generateID()
+	}
+	if req.RequestTime.IsZero() {
+		req.RequestTime = time.Now()
+	}
+	if req.ResponseTime.IsZero() {
+		req.ResponseTime = req.RequestTime
+	}
+
+	for _, transformer := range c.options.Transformers {
+		req = transformer(req)
+	}
+
+	c.notifier.Notify(req)
+	if c.eventAggregator != nil {
+		c.eventAggregator.CollectEvent(ctx, req)
+	}
+}
+
+// MaxBodySize returns the configured maximum body size for this collector,
+// for callers building a *Body via NewBodyFromBytes for use with Collect.
+func (c *HTTPClientCollector) MaxBodySize() int {
+	return c.options.MaxBodySize
+}
+
 // Close releases resources used by the collector
 func (c *HTTPClientCollector) Close() {
 	c.notifier.Close()

collector/http_client_test.go

@@ -1,13 +1,16 @@
 package collector_test
 
 import (
+	"context"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/gofrs/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -70,6 +73,243 @@ func TestHTTPClientCollector_UnreadResponseBody(t *testing.T) {
 	assert.True(t, req.ResponseBody.IsFullyCaptured())
 }
 
+func TestHTTPClientCollector_Collect_NoCapture_NoEventNoNotify(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	collect := Collect(t, httpCollector.Subscribe)
+
+	// No storage registered, so ShouldCapture is false for any context
+	httpCollector.Collect(context.Background(), collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com",
+	})
+
+	// Give the notifier a chance to deliver, if it were going to
+	time.Sleep(20 * time.Millisecond)
+
+	assert.Empty(t, collect.Stop())
+}
+
+func TestHTTPClientCollector_Collect_DispatchesToStorage(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	httpCollector.Collect(context.Background(), collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com",
+	})
+
+	events := storage.GetEvents(10)
+	require.Len(t, events, 1)
+
+	req, ok := events[0].Data.(collector.HTTPClientRequest)
+	require.True(t, ok)
+	assert.Equal(t, "https://example.com", req.URL)
+}
+
+func TestHTTPClientCollector_Collect_GroupsUnderParentEvent(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	parentCtx := aggregator.StartEvent(context.Background())
+
+	httpCollector.Collect(parentCtx, collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com/child",
+	})
+
+	aggregator.EndEvent(parentCtx, "parent event")
+
+	events := storage.GetEvents(10)
+	require.Len(t, events, 1, "the child request must not be dispatched as a top-level event")
+
+	parent := events[0]
+	require.Len(t, parent.Children, 1)
+
+	req, ok := parent.Children[0].Data.(collector.HTTPClientRequest)
+	require.True(t, ok)
+	assert.Equal(t, "https://example.com/child", req.URL)
+}
+
+func TestHTTPClientCollector_Collect_NotifiesSubscribers(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	collect := Collect(t, httpCollector.Subscribe)
+
+	httpCollector.Collect(context.Background(), collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com",
+	})
+
+	received := collect.Wait(1)
+	require.Len(t, received, 1)
+	assert.Equal(t, "https://example.com", received[0].URL)
+}
+
+func TestHTTPClientCollector_Collect_AppliesTransformers(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	options.Transformers = []collector.HTTPClientRequestTransformer{
+		func(req collector.HTTPClientRequest) collector.HTTPClientRequest {
+			if req.Tags == nil {
+				req.Tags = map[string]string{}
+			}
+			req.Tags["transformed"] = "true"
+			return req
+		},
+	}
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	httpCollector.Collect(context.Background(), collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com",
+	})
+
+	events := storage.GetEvents(10)
+	require.Len(t, events, 1)
+
+	req, ok := events[0].Data.(collector.HTTPClientRequest)
+	require.True(t, ok)
+	assert.Equal(t, "true", req.Tags["transformed"])
+}
+
+func TestHTTPClientCollector_Collect_FillsZeroValues(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	before := time.Now()
+
+	httpCollector.Collect(context.Background(), collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com",
+	})
+
+	events := storage.GetEvents(10)
+	require.Len(t, events, 1)
+
+	req, ok := events[0].Data.(collector.HTTPClientRequest)
+	require.True(t, ok)
+
+	assert.NotEqual(t, uuid.Nil, req.ID)
+	assert.False(t, req.RequestTime.Before(before))
+	assert.Equal(t, req.RequestTime, req.ResponseTime)
+}
+
+func TestHTTPClientCollector_Collect_PreservesNonZeroValues(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	id := uuid.Must(uuid.NewV7())
+	requestTime := time.Now().Add(-time.Minute)
+	responseTime := time.Now().Add(-30 * time.Second)
+
+	httpCollector.Collect(context.Background(), collector.HTTPClientRequest{
+		ID:           id,
+		Method:       http.MethodGet,
+		URL:          "https://example.com",
+		RequestTime:  requestTime,
+		ResponseTime: responseTime,
+	})
+
+	events := storage.GetEvents(10)
+	require.Len(t, events, 1)
+
+	req, ok := events[0].Data.(collector.HTTPClientRequest)
+	require.True(t, ok)
+
+	assert.Equal(t, id, req.ID)
+	assert.True(t, requestTime.Equal(req.RequestTime))
+	assert.True(t, responseTime.Equal(req.ResponseTime))
+}
+
+func TestHTTPClientCollector_Add_NotifiesButNoEvent(t *testing.T) {
+	aggregator := collector.NewEventAggregator()
+	defer aggregator.Close()
+
+	sessionID := uuid.Must(uuid.NewV4())
+	storage := collector.NewCaptureStorage(sessionID, 100, collector.CaptureModeGlobal)
+	aggregator.RegisterStorage(storage)
+
+	options := collector.DefaultHTTPClientOptions()
+	options.EventAggregator = aggregator
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	collect := Collect(t, httpCollector.Subscribe)
+
+	httpCollector.Add(collector.HTTPClientRequest{
+		Method: http.MethodGet,
+		URL:    "https://example.com",
+	})
+
+	received := collect.Wait(1)
+	require.Len(t, received, 1)
+	assert.Equal(t, "https://example.com", received[0].URL)
+
+	// Add never reaches the event aggregator
+	assert.Empty(t, storage.GetEvents(10))
+}
+
+func TestHTTPClientCollector_MaxBodySize(t *testing.T) {
+	options := collector.DefaultHTTPClientOptions()
+	options.MaxBodySize = 4096
+	httpCollector := collector.NewHTTPClientCollectorWithOptions(options)
+
+	assert.Equal(t, 4096, httpCollector.MaxBodySize())
+}
+
 // BodyReadTracker tracks if a response body was read
 type BodyReadTracker struct {
 	data            string