feat(forwarder): add tracing to HTTP client

cmd/forwarder/main.go

@@ -63,6 +63,8 @@ func realInit() (err error) {
 
 	logger.V(4).Info("initialized", "config", env)
 
+	tracing.SetLogger(logger)
+
 	tracerProvider, err := tracing.NewTracerProvider(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to initialize tracing: %w", err)
@@ -116,6 +118,10 @@ func realInit() (err error) {
 		s3Client, err = s3http.New(&s3http.Config{
 			DestinationURI:     env.DestinationURI,
 			GetObjectAPIClient: awsS3Client,
+			HTTPClient: tracing.NewHTTPClient(&tracing.HTTPClientConfig{
+				TracerProvider: tracerProvider,
+				Logger:         &logger,
+			}),
 		})
 		if err != nil {
 			return fmt.Errorf("failed to load http client: %w", err)

go.mod

@@ -48,6 +48,7 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
@@ -57,6 +58,7 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.52.3 // indirect
 	github.com/prometheus/procfs v0.13.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.51.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.26.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.26.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.26.0 // indirect

go.sum

@@ -55,6 +55,8 @@ github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
+github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
 github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
@@ -103,6 +105,8 @@ go.opentelemetry.io/contrib/exporters/autoexport v0.51.0 h1:imlL5MBzKu+NWhnJM62b
 go.opentelemetry.io/contrib/exporters/autoexport v0.51.0/go.mod h1:gn1wFA1uVEKIXrM3DC7SN9ee83oJ0yALY/HbUfqMszo=
 go.opentelemetry.io/contrib/instrumentation/github.com/aws/aws-sdk-go-v2/otelaws v0.51.0 h1:FGMfzzxfkNkw+gvKJOeT8dSmBjgrSFh+ClLl+OMKPno=
 go.opentelemetry.io/contrib/instrumentation/github.com/aws/aws-sdk-go-v2/otelaws v0.51.0/go.mod h1:hmHUXiKhyxbIhuNfG5ZTySq9HqqxJFNxaFOfXXvoMmQ=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.51.0 h1:Xs2Ncz0gNihqu9iosIZ5SkBbWo5T8JhhLJFMQL1qmLI=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.51.0/go.mod h1:vy+2G/6NvVMpwGX/NyLqcC41fxepnuKHk16E6IZUcJc=
 go.opentelemetry.io/contrib/propagators/b3 v1.26.0 h1:wgFbVA+bK2k+fGVfDOCOG4cfDAoppyr5sI2dVlh8MWM=
 go.opentelemetry.io/contrib/propagators/b3 v1.26.0/go.mod h1:DDktFXxA+fyItAAM0Sbl5OBH7KOsCTjvbBdPKtoIf/k=
 go.opentelemetry.io/otel v1.26.0 h1:LQwgL5s/1W7YiiRwxf03QGnWLb2HW4pLiAhaA5cZXBs=

handler/forwarder/s3http/client.go

@@ -145,15 +145,11 @@ func New(cfg *Config) (*Client, error) {
 		return nil, fmt.Errorf("failed to validate config: %w", err)
 	}
 
-	r, err := request.NewBuilder(&request.BuilderConfig{
-		URL: cfg.DestinationURI,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("failed to configure request builder: %w", err)
-	}
-
 	return &Client{
 		GetObjectAPIClient: cfg.GetObjectAPIClient,
-		RequestBuilder:     r,
+		RequestBuilder: &request.Builder{
+			URL:    cfg.DestinationURI,
+			Client: cfg.HTTPClient,
+		},
 	}, nil
 }

handler/forwarder/s3http/config.go

@@ -3,6 +3,7 @@ package s3http
 import (
 	"errors"
 	"fmt"
+	"net/http"
 	"net/url"
 )
 
@@ -14,6 +15,7 @@ var (
 type Config struct {
 	DestinationURI string // HTTP URI to upload data to
 	GetObjectAPIClient
+	HTTPClient *http.Client
 }
 
 func (c *Config) Validate() error {

handler/forwarder/s3http/internal/request/builder.go

@@ -2,13 +2,8 @@ package request
 
 import (
 	"errors"
-	"fmt"
 	"net/http"
 	"net/url"
-	"time"
-
-	"github.com/go-logr/logr"
-	"github.com/hashicorp/go-retryablehttp"
 )
 
 var (
@@ -21,17 +16,6 @@ type Doer interface {
 	Do(*http.Request) (*http.Response, error)
 }
 
-type BuilderConfig struct {
-	URL string
-
-	RetryWaitMin *time.Duration // Minimum time to wait on retry
-	RetryWaitMax *time.Duration // Maximumum time to wait on retry
-	RetryMax     *int           // Maximum number of retries
-
-	HTTPClient *http.Client
-	Logger     *logr.Logger
-}
-
 // Builder shares an HTTP client across request handlers.
 type Builder struct {
 	Client Doer
@@ -48,48 +32,13 @@ func (b *Builder) With(tags map[string]string) *Handler {
 	u, _ := url.Parse(b.URL)
 	u.RawQuery = values.Encode()
 
-	return &Handler{
-		URL:    u.String(),
-		Client: b.Client,
+	client := b.Client
+	if client == nil {
+		client = http.DefaultClient
 	}
-}
-
-func NewBuilder(cfg *BuilderConfig) (*Builder, error) {
-	if cfg == nil {
-		return nil, ErrNoConfig
-	}
-
-	if _, err := url.Parse(cfg.URL); err != nil {
-		return nil, fmt.Errorf("failed to parse base uri: %w", err)
-	}
-
-	client := retryablehttp.NewClient()
 
-	if cfg.HTTPClient != nil {
-		client.HTTPClient = cfg.HTTPClient
-	}
-
-	if cfg.RetryWaitMin != nil {
-		client.RetryWaitMin = *cfg.RetryWaitMin
-	}
-
-	if cfg.RetryWaitMax != nil {
-		client.RetryWaitMax = *cfg.RetryWaitMax
-	}
-
-	if cfg.RetryMax != nil {
-		client.RetryMax = *cfg.RetryMax
-	}
-
-	logger := logr.Discard()
-	if cfg.Logger != nil {
-		logger = *cfg.Logger
+	return &Handler{
+		URL:    u.String(),
+		Client: client,
 	}
-
-	client.Logger = &leveledLogger{logger}
-
-	return &Builder{
-		URL:    cfg.URL,
-		Client: client.StandardClient(),
-	}, nil
 }

tracing/http.go

@@ -0,0 +1,77 @@
+package tracing
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/hashicorp/go-retryablehttp"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.opentelemetry.io/otel/sdk/trace"
+)
+
+// leveledLogger provides an adapter between logr.Logger and retryablehttp.LeveledLogger.
+type leveledLogger struct {
+	logr.Logger
+}
+
+func (l *leveledLogger) Error(msg string, keysAndValues ...interface{}) {
+	l.V(1).Info(msg, keysAndValues...)
+}
+
+func (l *leveledLogger) Warn(msg string, keysAndValues ...interface{}) {
+	l.V(2).Info(msg, keysAndValues...)
+}
+
+func (l *leveledLogger) Info(msg string, keysAndValues ...interface{}) {
+	l.V(3).Info(msg, keysAndValues...)
+}
+
+func (l *leveledLogger) Debug(msg string, keysAndValues ...interface{}) {
+	l.V(4).Info(msg, keysAndValues...)
+}
+
+type HTTPClientConfig struct {
+	RetryWaitMin   *time.Duration // Minimum time to wait on retry
+	RetryWaitMax   *time.Duration // Maximumum time to wait on retry
+	RetryMax       *int           // Maximum number of retries
+	HTTPClient     *http.Client
+	Logger         *logr.Logger
+	TracerProvider *trace.TracerProvider
+}
+
+func NewHTTPClient(cfg *HTTPClientConfig) *http.Client {
+	if cfg == nil {
+		cfg = &HTTPClientConfig{}
+	}
+
+	client := retryablehttp.NewClient()
+
+	if cfg.HTTPClient != nil {
+		client.HTTPClient = cfg.HTTPClient
+	}
+
+	if cfg.RetryWaitMin != nil {
+		client.RetryWaitMin = *cfg.RetryWaitMin
+	}
+
+	if cfg.RetryWaitMax != nil {
+		client.RetryWaitMax = *cfg.RetryWaitMax
+	}
+
+	if cfg.RetryMax != nil {
+		client.RetryMax = *cfg.RetryMax
+	}
+
+	logger := logr.Discard()
+	if cfg.Logger != nil {
+		logger = *cfg.Logger
+	}
+
+	client.Logger = &leveledLogger{logger}
+
+	transport := &retryablehttp.RoundTripper{Client: client}
+	return &http.Client{
+		Transport: otelhttp.NewTransport(transport, otelhttp.WithTracerProvider(cfg.TracerProvider)),
+	}
+}

tracing/tracing.go

@@ -6,12 +6,18 @@ import (
 	"net/url"
 	"os"
 
+	"github.com/go-logr/logr"
 	"go.opentelemetry.io/contrib/detectors/aws/lambda"
 	"go.opentelemetry.io/contrib/exporters/autoexport"
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/sdk/resource"
 	"go.opentelemetry.io/otel/sdk/trace"
 )
 
+func SetLogger(logger logr.Logger) {
+	otel.SetLogger(logger)
+}
+
 // The OTEL SDK does not handle basic auth in OTEL_EXPORTER_OTLP_ENDPOINT
 // Extract username and password and set as OTLP Headers.
 func handleOTLPEndpointAuth() error {