Log original OTLP log-export errors to worker logs

internal/tasktelemetry/otlp/exporter.go

@@ -7,6 +7,7 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/rs/zerolog/log"
 	"github.com/warpdotdev/oz-agent-worker/internal/types"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
@@ -108,6 +109,8 @@ type logExporter struct {
 func (e logExporter) Export(ctx context.Context, records []sdklog.Record) error {
 	// OTel reports exporter errors outside the task-local log redactor.
 	if err := e.Exporter.Export(ctx, records); err != nil {
+		// The context logger would send the diagnostic back to the task collector.
+		log.Error().Err(err).Msg("task log export failed")
 		return errors.New("task log export failed")
 	}
 	return nil

internal/tasktelemetry/otlp/exporter_test.go

@@ -1,16 +1,21 @@
 package otlp
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"encoding/pem"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 	"github.com/warpdotdev/oz-agent-worker/internal/types"
 	"go.opentelemetry.io/otel/attribute"
 	sdklog "go.opentelemetry.io/otel/sdk/log"
@@ -20,6 +25,61 @@ import (
 	"golang.org/x/oauth2"
 )
 
+func TestLogExportErrorStaysInWorkerLog(t *testing.T) {
+	const diagnostic = "collector rejected the export: diagnostic detail"
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		requests.Add(1)
+		http.Error(w, diagnostic, http.StatusForbidden)
+	}))
+	defer server.Close()
+
+	var workerOutput, taskOutput bytes.Buffer
+	previous := log.Logger
+	log.Logger = zerolog.New(&workerOutput)
+	t.Cleanup(func() { log.Logger = previous })
+	ctx := zerolog.New(&taskOutput).WithContext(t.Context())
+	config, err := NewExportConfig(ctx, "", &types.TaskAssignmentMessage{
+		TelemetryCollection: &types.TelemetryCollectionConfig{
+			Endpoint: server.URL,
+			BootstrapToken: &types.TelemetryBootstrapToken{
+				Token: "task-token", ExpiresAt: time.Now().Add(time.Hour).Format(time.RFC3339Nano),
+			},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	exporter, err := config.LogExporter()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = exporter.Shutdown(context.Background()) }()
+	var record sdklog.Record
+	record.SetBody(attribute.StringValue("task output"))
+	err = exporter.Export(ctx, []sdklog.Record{record})
+	if err == nil || err.Error() != "task log export failed" {
+		t.Fatalf("exporter returned an unsafe error: %v", err)
+	}
+	if taskOutput.Len() != 0 {
+		t.Fatalf("exporter diagnostic reached the task logger: %s", &taskOutput)
+	}
+	var event struct {
+		Level   string `json:"level"`
+		Error   string `json:"error"`
+		Message string `json:"message"`
+	}
+	if err := json.Unmarshal(workerOutput.Bytes(), &event); err != nil {
+		t.Fatalf("missing worker diagnostic: %v", err)
+	}
+	if event.Level != "error" || !strings.Contains(event.Error, diagnostic) || event.Message != "task log export failed" {
+		t.Fatalf("worker diagnostic lost the original error: %+v", event)
+	}
+	if requests.Load() != 1 {
+		t.Fatalf("exporter made %d requests, want one failed export", requests.Load())
+	}
+}
+
 func TestHTTPSExport(t *testing.T) {
 	for _, signal := range []string{"traces", "logs"} {
 		t.Run(signal, func(t *testing.T) {