rework exposures, add tests and better flattening

Signed-off-by: Eliott Bouhana <[email protected]>

Author: eliottness

ddtrace/tracer/option.go

@@ -6,6 +6,7 @@
 package tracer
 
 import (
+	"cmp"
 	"context"
 	"encoding/json"
 	"errors"
@@ -551,14 +552,10 @@ func newConfig(opts ...StartOption) (*config, error) {
 		if c.agentURL.Scheme == "unix" {
 			// If we're connecting over UDS we can just rely on the agent to provide the hostname
 			log.Debug("connecting to agent over unix, do not set hostname on any traces")
-			c.httpClient = udsClient(c.agentURL.Path, c.httpClientTimeout)
-			// TODO(darccio): use internal.UnixDataSocketURL instead
-			c.agentURL = &url.URL{
-				Scheme: "http",
-				Host:   fmt.Sprintf("UDS_%s", strings.NewReplacer(":", "_", "/", "_", `\`, "_").Replace(c.agentURL.Path)),
-			}
+			c.httpClient = internal.UDSClient(c.agentURL.Path, cmp.Or(c.httpClientTimeout, defaultHTTPTimeout))
+			c.agentURL = internal.UnixDataSocketURL(c.agentURL.Path)
 		} else {
-			c.httpClient = defaultHTTPClient(c.httpClientTimeout, false)
+			c.httpClient = internal.DefaultHTTPClient(c.httpClientTimeout, false)
 		}
 	}
 	WithGlobalTag(ext.RuntimeID, globalconfig.RuntimeID())(c)
@@ -742,29 +739,6 @@ func newStatsdClient(c *config) (internal.StatsdClient, error) {
 	return internal.NewStatsdClient(c.dogstatsdAddr, statsTags(c))
 }
 
-// udsClient returns a new http.Client which connects using the given UDS socket path.
-func udsClient(socketPath string, timeout time.Duration) *http.Client {
-	if timeout == 0 {
-		timeout = defaultHTTPTimeout
-	}
-	return &http.Client{
-		Transport: &http.Transport{
-			Proxy: http.ProxyFromEnvironment,
-			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
-				return defaultDialer(timeout).DialContext(ctx, "unix", (&net.UnixAddr{
-					Name: socketPath,
-					Net:  "unix",
-				}).String())
-			},
-			MaxIdleConns:          100,
-			IdleConnTimeout:       90 * time.Second,
-			TLSHandshakeTimeout:   10 * time.Second,
-			ExpectContinueTimeout: 1 * time.Second,
-		},
-		Timeout: timeout,
-	}
-}
-
 // defaultDogstatsdAddr returns the default connection address for Dogstatsd.
 func defaultDogstatsdAddr() string {
 	envHost, envPort := env.Get("DD_DOGSTATSD_HOST"), env.Get("DD_DOGSTATSD_PORT")

ddtrace/tracer/transport.go

@@ -9,7 +9,6 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"runtime"
 	"strconv"
@@ -31,32 +30,6 @@ const (
 	headerComputedTopLevel = "Datadog-Client-Computed-Top-Level"
 )
 
-func defaultDialer(timeout time.Duration) *net.Dialer {
-	return &net.Dialer{
-		Timeout:   timeout,
-		KeepAlive: 30 * time.Second,
-		DualStack: true,
-	}
-}
-
-func defaultHTTPClient(timeout time.Duration, disableKeepAlives bool) *http.Client {
-	if timeout == 0 {
-		timeout = defaultHTTPTimeout
-	}
-	return &http.Client{
-		Transport: &http.Transport{
-			Proxy:                 http.ProxyFromEnvironment,
-			DialContext:           defaultDialer(timeout).DialContext,
-			MaxIdleConns:          100,
-			IdleConnTimeout:       90 * time.Second,
-			TLSHandshakeTimeout:   10 * time.Second,
-			ExpectContinueTimeout: 1 * time.Second,
-			DisableKeepAlives:     disableKeepAlives,
-		},
-		Timeout: timeout,
-	}
-}
-
 const (
 	defaultHostname          = "localhost"
 	defaultPort              = "8126"

internal/agent.go

@@ -7,8 +7,10 @@ package internal
 
 import (
 	"net"
+	"net/http"
 	"net/url"
 	"os"
+	"time"
 
 	"github.com/DataDog/dd-trace-go/v2/internal/env"
 	"github.com/DataDog/dd-trace-go/v2/internal/log"
@@ -74,3 +76,26 @@ func AgentURLFromEnv() *url.URL {
 	}
 	return httpURL
 }
+
+func DefaultDialer(timeout time.Duration) *net.Dialer {
+	return &net.Dialer{
+		Timeout:   timeout,
+		KeepAlive: 30 * time.Second,
+		DualStack: true,
+	}
+}
+
+func DefaultHTTPClient(timeout time.Duration, disableKeepAlives bool) *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			Proxy:                 http.ProxyFromEnvironment,
+			DialContext:           DefaultDialer(timeout).DialContext,
+			MaxIdleConns:          100,
+			IdleConnTimeout:       90 * time.Second,
+			TLSHandshakeTimeout:   10 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+			DisableKeepAlives:     disableKeepAlives,
+		},
+		Timeout: timeout,
+	}
+}

internal/telemetry/internal/writer.go

@@ -11,7 +11,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"os"
 	"runtime"
@@ -29,25 +28,6 @@ import (
 	"github.com/DataDog/dd-trace-go/v2/internal/version"
 )
 
-// We copy the transport to avoid using the default one, as it might be
-// augmented with tracing and we don't want these calls to be recorded.
-// See https://golang.org/pkg/net/http/#DefaultTransport .
-var defaultHTTPClient = &http.Client{
-	Transport: &http.Transport{
-		Proxy: http.ProxyFromEnvironment,
-		DialContext: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-		}).DialContext,
-		ForceAttemptHTTP2:     true,
-		MaxIdleConns:          100,
-		IdleConnTimeout:       90 * time.Second,
-		TLSHandshakeTimeout:   10 * time.Second,
-		ExpectContinueTimeout: 1 * time.Second,
-	},
-	Timeout: 5 * time.Second,
-}
-
 func newBody(config TracerConfig, debugMode bool) *transport.Body {
 	osHostname, err := os.Hostname()
 	if err != nil {
@@ -130,7 +110,7 @@ func NewWriter(config WriterConfig) (Writer, error) {
 	}
 
 	if config.HTTPClient == nil {
-		config.HTTPClient = defaultHTTPClient
+		config.HTTPClient = internal.DefaultHTTPClient(5*time.Second, true)
 	}
 
 	// Don't allow the client to have a timeout higher than 5 seconds

internal/uds.go

@@ -6,9 +6,13 @@
 package internal
 
 import (
+	"context"
 	"fmt"
+	"net"
+	"net/http"
 	"net/url"
 	"strings"
+	"time"
 )
 
 func UnixDataSocketURL(path string) *url.URL {
@@ -17,3 +21,23 @@ func UnixDataSocketURL(path string) *url.URL {
 		Host:   fmt.Sprintf("UDS_%s", strings.NewReplacer(":", "_", "/", "_", `\`, "_").Replace(path)),
 	}
 }
+
+// UDSClient returns a new http.Client which connects using the given UDS socket path.
+func UDSClient(socketPath string, timeout time.Duration) *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			Proxy: http.ProxyFromEnvironment,
+			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+				return DefaultDialer(timeout).DialContext(ctx, "unix", (&net.UnixAddr{
+					Name: socketPath,
+					Net:  "unix",
+				}).String())
+			},
+			MaxIdleConns:          100,
+			IdleConnTimeout:       90 * time.Second,
+			TLSHandshakeTimeout:   10 * time.Second,
+			ExpectContinueTimeout: 1 * time.Second,
+		},
+		Timeout: timeout,
+	}
+}

openfeature/evaluator.go

@@ -77,7 +77,7 @@ func evaluateFlag(flag *flag, defaultValue any, context map[string]any) evaluati
 			if allocation.DoLog != nil {
 				doLog = *allocation.DoLog
 			}
-			metadata[metadataDoLog] = doLog
+			metadata[metadataDoLogKey] = doLog
 
 			return evaluationResult{
 				Value:      variant.Value,

openfeature/exposure.go

@@ -9,18 +9,21 @@ import (
 	"bytes"
 	"cmp"
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/url"
+	"os"
 	"sync"
 	"time"
 
 	"github.com/DataDog/dd-trace-go/v2/internal"
 	"github.com/DataDog/dd-trace-go/v2/internal/env"
 	"github.com/DataDog/dd-trace-go/v2/internal/globalconfig"
 	"github.com/DataDog/dd-trace-go/v2/internal/log"
+	telemetrylog "github.com/DataDog/dd-trace-go/v2/internal/telemetry/log"
+	jsoniter "github.com/json-iterator/go"
 )
 
 const (
@@ -75,9 +78,9 @@ type exposureSubject struct {
 
 // exposureContext represents service context metadata for the exposure payload
 type exposureContext struct {
-	ServiceName string `json:"service"`
-	Version     string `json:"version,omitempty"`
-	Env         string `json:"env,omitempty"`
+	Service string `json:"service"`
+	Version string `json:"version,omitempty"`
+	Env     string `json:"env,omitempty"`
 }
 
 // exposurePayload represents the complete payload sent to the exposure endpoint
@@ -97,31 +100,34 @@ type exposureWriter struct {
 	ticker        *time.Ticker
 	stopChan      chan struct{}
 	stopped       bool
+	jsonConfig    jsoniter.API
 }
 
 // newExposureWriter creates a new exposure writer with the given configuration
 func newExposureWriter(config ProviderConfig) *exposureWriter {
-	// Build service context from environment variables
-	serviceName := cmp.Or(env.Get("DD_SERVICE"), globalconfig.ServiceName())
-	if serviceName == "" {
-		serviceName = "unknown"
+	agentURL := internal.AgentURLFromEnv()
+	var httpClient *http.Client
+	if agentURL.Scheme == "unix" {
+		httpClient = internal.UDSClient(agentURL.Path, defaultHTTPTimeout)
+		agentURL = internal.UnixDataSocketURL(agentURL.Path)
+	} else {
+		httpClient = internal.DefaultHTTPClient(defaultHTTPTimeout, false)
 	}
 
-	context := exposureContext{
-		ServiceName: serviceName,
-		Version:     env.Get("DD_VERSION"),
-		Env:         env.Get("DD_ENV"),
-	}
+	executable, _ := os.Executable()
 
 	return &exposureWriter{
-		buffer:        make([]exposureEvent, 0),
+		buffer:        make([]exposureEvent, 1<<8), // Initial capacity of 256
 		flushInterval: cmp.Or(config.ExposureFlushInterval, defaultExposureFlushInterval),
-		httpClient: &http.Client{
-			Timeout: defaultHTTPTimeout,
+		httpClient:    httpClient,
+		agentURL:      agentURL,
+		stopChan:      make(chan struct{}),
+		jsonConfig:    jsoniter.Config{}.Froze(),
+		context: exposureContext{
+			Service: cmp.Or(env.Get("DD_SERVICE"), globalconfig.ServiceName(), executable),
+			Version: env.Get("DD_VERSION"),
+			Env:     env.Get("DD_ENV"),
 		},
-		agentURL: internal.AgentURLFromEnv(),
-		context:  context,
-		stopChan: make(chan struct{}),
 	}
 }
 
@@ -132,6 +138,13 @@ func (w *exposureWriter) start() {
 		defer func() {
 			if r := recover(); r != nil {
 				log.Error("openfeature: exposure writer recovered panic: %v", r)
+				var errAttr slog.Attr
+				if err, ok := r.(error); ok {
+					errAttr = slog.Any("panic", telemetrylog.NewSafeError(err))
+				} else {
+					errAttr = slog.Any("panic", r)
+				}
+				telemetrylog.Error("openfeature: exposure writer recovered panic", errAttr)
 			}
 			w.stop()
 		}()
@@ -171,17 +184,14 @@ func (w *exposureWriter) flush() {
 
 	// Move buffer to local variable and create new buffer
 	events := w.buffer
-	w.buffer = make([]exposureEvent, len(events)/2)
+	w.buffer = make([]exposureEvent, 0, len(events)/2)
 	w.mu.Unlock()
 
-	// Build payload
-	payload := exposurePayload{
+	// Send to agent
+	if err := w.sendToAgent(exposurePayload{
 		Context:   w.context,
 		Exposures: events,
-	}
-
-	// Send to agent
-	if err := w.sendToAgent(payload); err != nil {
+	}); err != nil {
 		log.Error("openfeature: failed to send exposure events: %v", err.Error())
 	} else {
 		log.Debug("openfeature: successfully sent %d exposure events", len(events))
@@ -191,16 +201,19 @@ func (w *exposureWriter) flush() {
 // sendToAgent sends the exposure payload to the Datadog Agent via EVP proxy
 func (w *exposureWriter) sendToAgent(payload exposurePayload) error {
 	// Serialize payload
-	jsonData, err := json.Marshal(payload)
-	if err != nil {
-		return fmt.Errorf("failed to marshal exposure payload: %w", err)
+	var bytesBuffer bytes.Buffer
+	encoder := w.jsonConfig.NewEncoder(&bytesBuffer)
+	if err := encoder.Encode(payload); err != nil {
+		return fmt.Errorf("failed to encode exposure payload: %w", err)
 	}
 
 	// Build request URL
-	requestURL := w.buildRequestURL()
+	u := *w.agentURL
+	u.Path = exposureEndpoint
+	requestURL := u.String()
 
 	// Create HTTP request
-	req, err := http.NewRequestWithContext(context.Background(), "POST", requestURL, bytes.NewReader(jsonData))
+	req, err := http.NewRequestWithContext(context.Background(), "POST", requestURL, &bytesBuffer)
 	if err != nil {
 		return fmt.Errorf("failed to create request: %w", err)
 	}
@@ -227,21 +240,6 @@ func (w *exposureWriter) sendToAgent(payload exposurePayload) error {
 	return nil
 }
 
-// buildRequestURL constructs the full URL for the exposure endpoint
-func (w *exposureWriter) buildRequestURL() string {
-	if w.agentURL.Scheme == "unix" {
-		// For Unix domain sockets, use the HTTP adapter
-		u := internal.UnixDataSocketURL(w.agentURL.Path)
-		u.Path = exposureEndpoint
-		return u.String()
-	}
-
-	// For HTTP/HTTPS URLs, append the endpoint path
-	u := *w.agentURL
-	u.Path = exposureEndpoint
-	return u.String()
-}
-
 // stop stops the exposure writer and flushes any remaining events
 func (w *exposureWriter) stop() {
 	w.mu.Lock()