feat(forwarder): add tracing to HTTP client (#243)

This commit introduces tracing support for the legacy HTTP client
embedded in the forwarder. The HTTP client has be refactored out of the
internal package in order to facilitate reuse across components if
necessary.

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 {