feat(otlp/export): offline OTLP trace/metric/log export

CODEOWNERS

@@ -68,8 +68,9 @@
 # llm observability
 /llmobs                                    @DataDog/ml-observability @DataDog/apm-idm-go @DataDog/apm-go
 /llmobs/export                             @DataDog/trajectory @DataDog/ml-observability @DataDog/apm-go
+/otlp/export                               @DataDog/trajectory @DataDog/ml-observability @DataDog/apm-go
 /internal/llmobs                           @DataDog/ml-observability @DataDog/apm-idm-go @DataDog/apm-go
-/internal/llmobs/exportutil                @DataDog/trajectory @DataDog/ml-observability @DataDog/apm-go
+/internal/exportutil                       @DataDog/trajectory @DataDog/ml-observability @DataDog/apm-go
 /contrib/mark3labs/mcp-go                  @DataDog/ml-observability @DataDog/apm-idm-go
 /contrib/modelcontextprotocol/go-sdk       @DataDog/ml-observability @DataDog/apm-idm-go
 

internal/README.md

@@ -30,6 +30,10 @@ It also holds `supported_configurations*`, which maintains which environment var
 
 `go run ./scripts/configinverter/main.go add DD_MY_NEW_KEY`
 
+### Export Utilities
+
+Contains small helpers for the offline export clients ([llmobs/export](../llmobs/export/) and [otlp/export](../otlp/export/)). Both clients use bounded, UTF-8-safe response-body snippets (`Snippet`) and per-request failure aggregation (`Aggregate`); `otlp/export` additionally uses the bounded-retry engine over a POST closure (`Retry`) with a transient-vs-permanent classifier (`Retriable`, or a caller-supplied one), while `llmobs/export` retries via the internal LLM Obs transport. See [exportutil](./exportutil/).
+
 ### Locking
 
 Locking functionality that serves as a replacement for `sync.mutex` and similar locking mechanisms. It enables checking for deadlocks and should be used instead of `sync`. For more information, read the [README](./locking/README.md).

internal/llmobs/exportutil/exportutil.go → internal/exportutil/exportutil.go

@@ -4,8 +4,8 @@
 // Copyright 2026 Datadog, Inc.
 
 // Package exportutil holds small helpers shared by the offline export clients
-// (llmobs/export and otlp/export): bounded response-body snippets and
-// per-request failure aggregation.
+// (llmobs/export and otlp/export): bounded response-body snippets,
+// per-request failure aggregation, and a bounded HTTP retry engine.
 package exportutil
 
 import (

internal/exportutil/retry.go

@@ -0,0 +1,118 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026 Datadog, Inc.
+
+package exportutil
+
+import (
+	"context"
+	"math/rand/v2"
+	"net/http"
+	"time"
+)
+
+const (
+	initialBackoff = 100 * time.Millisecond
+	maxBackoff     = time.Second
+)
+
+// jitter applies equal-jitter to d (half fixed, half random in [0, d/2]) so that
+// many exporters hitting the same transient failure do not retry in lockstep.
+func jitter(d time.Duration) time.Duration {
+	if d <= 0 {
+		return d
+	}
+	return d/2 + time.Duration(rand.Int64N(int64(d/2)+1))
+}
+
+// Result reports the outcome of a bounded Retry over a single request's POST.
+type Result struct {
+	StatusCode int
+	Attempts   int
+	Body       []byte
+	Retriable  bool
+}
+
+// Retriable classifies a failed attempt. A caller-cancelled/expired context is
+// not transient; status 0 (network error), 408, 429 and 5xx are transient;
+// other 4xx are permanent.
+func Retriable(ctx context.Context, status int) bool {
+	if ctx.Err() != nil {
+		return false
+	}
+	switch {
+	case status == 0:
+		return true
+	case status == http.StatusRequestTimeout, status == http.StatusTooManyRequests:
+		return true
+	case status >= 500 && status <= 599:
+		return true
+	default:
+		return false
+	}
+}
+
+// Attempt is the outcome of a single do() call inside Retry.
+type Attempt struct {
+	// Status is the HTTP status (0 for a network-level error).
+	Status int
+	// Body is the (bounded) response body.
+	Body []byte
+	// RetryAfter is a server-requested minimum delay before the next attempt
+	// (e.g. parsed from a Retry-After header); 0 falls back to exponential
+	// backoff. It is only honored when the attempt is retriable.
+	RetryAfter time.Duration
+	// Err is nil on success.
+	Err error
+}
+
+// RetryOptions configures Retry.
+type RetryOptions struct {
+	// MaxAttempts bounds the total number of do() calls (>=1).
+	MaxAttempts uint
+	// Retriable classifies a failed status as transient. Defaults to Retriable
+	// when nil, letting callers with stricter rules (e.g. the OTLP/HTTP spec)
+	// override the classification.
+	Retriable func(ctx context.Context, status int) bool
+}
+
+// Retry runs do up to opts.MaxAttempts times with exponential backoff, retrying
+// only transient failures (per opts.Retriable). A retriable attempt that reports
+// a RetryAfter waits at least that long instead of the backoff. It returns a
+// structured Result plus the error from the final attempt (nil on success).
+func Retry(ctx context.Context, opts RetryOptions, do func(context.Context) Attempt) (Result, error) {
+	retriable := opts.Retriable
+	if retriable == nil {
+		retriable = Retriable
+	}
+	var res Result
+	backoff := initialBackoff
+	for attempt := 1; ; attempt++ {
+		res.Attempts = attempt
+		a := do(ctx)
+		res.StatusCode = a.Status
+		res.Body = a.Body
+		if a.Err == nil {
+			res.Retriable = false
+			return res, nil
+		}
+		res.Retriable = retriable(ctx, a.Status)
+		if !res.Retriable || uint(attempt) >= opts.MaxAttempts {
+			return res, a.Err
+		}
+		wait := jitter(backoff)
+		if a.RetryAfter > 0 {
+			wait = a.RetryAfter
+		}
+		select {
+		case <-ctx.Done():
+			res.Retriable = false
+			return res, ctx.Err()
+		case <-time.After(wait):
+		}
+		if backoff *= 2; backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}

internal/exportutil/retry_test.go

@@ -0,0 +1,103 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026 Datadog, Inc.
+
+package exportutil
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRetry_SucceedsFirstTry(t *testing.T) {
+	calls := 0
+	res, err := Retry(context.Background(), RetryOptions{MaxAttempts: 3}, func(context.Context) Attempt {
+		calls++
+		return Attempt{Status: 200, Body: []byte("ok")}
+	})
+	require.NoError(t, err)
+	assert.Equal(t, 1, calls)
+	assert.Equal(t, 1, res.Attempts)
+	assert.False(t, res.Retriable)
+	assert.Equal(t, 200, res.StatusCode)
+}
+
+func TestRetry_StopsAtMaxAttemptsWithDefaultClassifier(t *testing.T) {
+	calls := 0
+	res, err := Retry(context.Background(), RetryOptions{MaxAttempts: 3}, func(context.Context) Attempt {
+		calls++
+		return Attempt{Status: 503, Err: errors.New("unavailable")} // default Retriable retries 5xx
+	})
+	require.Error(t, err)
+	assert.Equal(t, 3, calls) // exhausted every attempt
+	assert.Equal(t, 3, res.Attempts)
+	assert.True(t, res.Retriable)
+	assert.Equal(t, 503, res.StatusCode)
+}
+
+func TestRetry_NonRetriableStopsImmediately(t *testing.T) {
+	calls := 0
+	res, err := Retry(context.Background(), RetryOptions{MaxAttempts: 3}, func(context.Context) Attempt {
+		calls++
+		return Attempt{Status: 400, Err: errors.New("bad request")}
+	})
+	require.Error(t, err)
+	assert.Equal(t, 1, calls) // 400 is permanent under the default classifier
+	assert.False(t, res.Retriable)
+}
+
+func TestRetry_HonorsRetryAfterOverBackoff(t *testing.T) {
+	calls := 0
+	start := time.Now()
+	_, err := Retry(context.Background(), RetryOptions{MaxAttempts: 2}, func(context.Context) Attempt {
+		calls++
+		if calls == 1 {
+			return Attempt{Status: 503, RetryAfter: 400 * time.Millisecond, Err: errors.New("throttled")}
+		}
+		return Attempt{Status: 200}
+	})
+	require.NoError(t, err)
+	assert.Equal(t, 2, calls)
+	// The wait must honor RetryAfter (~400ms), not the ~100ms exponential backoff.
+	assert.GreaterOrEqual(t, time.Since(start), 350*time.Millisecond)
+}
+
+func TestRetry_ContextCancelInterruptsWait(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	start := time.Now()
+	go func() {
+		time.Sleep(30 * time.Millisecond)
+		cancel()
+	}()
+	_, err := Retry(ctx, RetryOptions{MaxAttempts: 5}, func(context.Context) Attempt {
+		return Attempt{Status: 503, RetryAfter: 10 * time.Second, Err: errors.New("throttled")}
+	})
+	require.ErrorIs(t, err, context.Canceled)
+	assert.Less(t, time.Since(start), 2*time.Second) // did not sleep the full 10s
+}
+
+func TestRetry_CustomRetriablePredicate(t *testing.T) {
+	only418 := func(_ context.Context, status int) bool { return status == 418 }
+
+	calls := 0
+	_, err := Retry(context.Background(), RetryOptions{MaxAttempts: 3, Retriable: only418}, func(context.Context) Attempt {
+		calls++
+		return Attempt{Status: 500, Err: errors.New("err")} // not retriable under this predicate
+	})
+	require.Error(t, err)
+	assert.Equal(t, 1, calls)
+
+	calls = 0
+	_, err = Retry(context.Background(), RetryOptions{MaxAttempts: 3, Retriable: only418}, func(context.Context) Attempt {
+		calls++
+		return Attempt{Status: 418, Err: errors.New("teapot")}
+	})
+	require.Error(t, err)
+	assert.Equal(t, 3, calls) // 418 retries under this predicate
+}

llmobs/export/export.go

@@ -14,7 +14,7 @@ import (
 	"slices"
 	"strings"
 
-	"github.com/DataDog/dd-trace-go/v2/internal/llmobs/exportutil"
+	"github.com/DataDog/dd-trace-go/v2/internal/exportutil"
 	"github.com/DataDog/dd-trace-go/v2/internal/llmobs/transport"
 )
 

otlp/export/bench_test.go

@@ -0,0 +1,51 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026 Datadog, Inc.
+
+package export_test
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	tracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+
+	"github.com/DataDog/dd-trace-go/v2/otlp/export"
+)
+
+// discardTransport accepts every request and retains nothing, so the benchmark
+// measures marshal/POST-assembly cost without network or bookkeeping noise.
+type discardTransport struct{}
+
+func (discardTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+		_ = req.Body.Close()
+	}
+	return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader("")), Header: http.Header{}}, nil
+}
+
+func BenchmarkExportTraces(b *testing.B) {
+	c, err := export.NewTraceClient(export.Config{
+		Site: "datadoghq.com", APIKey: "k",
+		HTTPClient: &http.Client{Transport: discardTransport{}},
+	})
+	if err != nil {
+		b.Fatal(err)
+	}
+	reqs := make([]*tracepb.ExportTraceServiceRequest, 50)
+	for i := range reqs {
+		reqs[i] = sampleTrace()
+	}
+	ctx := context.Background()
+	b.ReportAllocs()
+	for b.Loop() {
+		if _, err := c.ExportTraces(ctx, reqs); err != nil {
+			b.Fatal(err)
+		}
+	}
+}

otlp/export/client.go

@@ -0,0 +1,58 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026 Datadog, Inc.
+
+package export
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/DataDog/dd-trace-go/v2/internal/exportutil"
+)
+
+// partialSuccessFunc decodes a signal's OTLP 200 response body, reporting how
+// many records the intake rejected (0 if none) and any accompanying message. It
+// returns an error when the body is not a decodable Export*ServiceResponse, so a
+// non-OTLP 200 (e.g. a proxy/login page) is surfaced as a failed export rather
+// than silently counted as zero rejections.
+type partialSuccessFunc func(body []byte) (rejected int64, message string, err error)
+
+// exportEach posts each request atomically (one request -> one POST -> one
+// result row), preserving input order by index. A 200 response that reports OTLP
+// partial success (rejected_* > 0), or whose body does not decode as the
+// expected OTLP response, is surfaced as a per-request error via partial. It
+// returns a non-nil error if any request failed; per-request detail is in the
+// result.
+func exportEach[T proto.Message](ctx context.Context, t *rawTransport, reqs []T, partial partialSuccessFunc) (*ExportResult, error) {
+	res := &ExportResult{}
+	for i, r := range reqs {
+		if isNilMessage(r) {
+			res.Requests = append(res.Requests, RequestResult{Index: i, Err: errNilRequest})
+			continue
+		}
+		rr, body := t.export(ctx, r)
+		rr.Index = i
+		if rr.Err == nil {
+			switch rejected, msg, derr := partial(body); {
+			case derr != nil:
+				rr.Err = fmt.Errorf("otlp/export: response body is not a valid OTLP response: %w", derr)
+			case rejected > 0:
+				rr.Err = fmt.Errorf("otlp/export: intake reported partial success, %d record(s) rejected: %s", rejected, msg)
+			}
+		}
+		res.Requests = append(res.Requests, rr)
+	}
+	return res, exportutil.Aggregate(res.Failed(), len(res.Requests), "otlp/export")
+}
+
+// isNilMessage reports whether v is a typed-nil proto message pointer, which
+// would otherwise marshal to an empty body and be silently sent.
+func isNilMessage[T proto.Message](v T) bool {
+	rv := reflect.ValueOf(v)
+	return rv.Kind() == reflect.Ptr && rv.IsNil()
+}