feat(tracing): add trace-aware logging to writeback executor (#544)

* Tracing putTag

* Nginx and logs in functions

* Tracing executor tasks

* Revert mocks

* Fix storage mock

* Partially revert

* Revert mock

* Update

* Remove changes to executor

* Separate DB operations

* DecodeString

* Fix typo

* Database persistence for writeback trace context

* feat(origin): add tracing to blob upload handlers

* Remove context import

* otel middleware

* feat(tracing): propagate trace context to writeback tasks in build-index and origin

* Update tagstore tests

* feat(tracing): add trace-aware logging to writeback executor

Author: hweawer

build-index/tagserver/server.go

@@ -252,7 +252,6 @@ func (s *Server) putTagHandler(w http.ResponseWriter, r *http.Request) error {
 	}
 
 	span.SetStatus(codes.Ok, "tag put successfully")
-
 	span.SetAttributes(attribute.Bool("success", true))
 	w.WriteHeader(http.StatusOK)
 	return nil

lib/persistedretry/writeback/executor.go

@@ -14,6 +14,7 @@
 package writeback
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"time"
@@ -25,6 +26,8 @@ import (
 	"github.com/uber/kraken/lib/store/metadata"
 	"github.com/uber/kraken/utils/closers"
 	"github.com/uber/kraken/utils/log"
+
+	"go.opentelemetry.io/otel/trace"
 )
 
 // FileStore defines store operations required for write-back.
@@ -44,8 +47,8 @@ type Executor struct {
 func NewExecutor(
 	stats tally.Scope,
 	fs FileStore,
-	backends *backend.Manager) *Executor {
-
+	backends *backend.Manager,
+) *Executor {
 	stats = stats.Tagged(map[string]string{
 		"module": "writebackexecutor",
 	})
@@ -65,33 +68,91 @@ func (e *Executor) Exec(r persistedretry.Task) error {
 	if !ok {
 		return fmt.Errorf("expected *Task, got %T", r)
 	}
-	if err := e.upload(t); err != nil {
+
+	// Extract context from task for trace propagation
+	// Tasks from public endpoints will have trace context, internal tasks may not
+	ctx := e.getContextFromTask(t)
+
+	log.WithTraceContext(ctx).With(
+		"namespace", t.Namespace,
+		"name", t.Name,
+		"has_trace_context", t.HasTraceContext(),
+	).Debug("Executing writeback task")
+
+	if err := e.upload(ctx, t); err != nil {
+		log.WithTraceContext(ctx).With(
+			"namespace", t.Namespace,
+			"name", t.Name,
+			"error", err,
+		).Error("Failed to upload during writeback")
 		return err
 	}
+
 	err := e.fs.DeleteCacheFileMetadata(t.Name, &metadata.Persist{})
 	if err != nil && !os.IsNotExist(err) {
+		log.WithTraceContext(ctx).With(
+			"namespace", t.Namespace,
+			"name", t.Name,
+			"error", err,
+		).Error("Failed to delete persist metadata")
 		return fmt.Errorf("delete persist metadata: %s", err)
 	}
+
+	log.WithTraceContext(ctx).With(
+		"namespace", t.Namespace,
+		"name", t.Name,
+	).Debug("Successfully completed writeback task")
+
 	return nil
 }
 
-func (e *Executor) upload(t *Task) error {
+// getContextFromTask extracts trace context from a task if available.
+// Returns context.Background() if no trace context is present.
+func (e *Executor) getContextFromTask(t *Task) context.Context {
+	if !t.HasTraceContext() {
+		return context.Background()
+	}
+
+	spanCtx := t.SpanContext()
+	if !spanCtx.IsValid() {
+		return context.Background()
+	}
+
+	// Create a context with the span context for logging correlation
+	return trace.ContextWithSpanContext(context.Background(), spanCtx)
+}
+
+func (e *Executor) upload(ctx context.Context, t *Task) error {
 	start := time.Now()
 
-	log.With("namespace", t.Namespace, "name", t.Name).Info("Uploading cache file to the remote backend")
+	log.WithTraceContext(ctx).With(
+		"namespace", t.Namespace,
+		"name", t.Name,
+	).Info("Uploading cache file to the remote backend")
+
 	client, err := e.backends.GetClient(t.Namespace)
 	if err != nil {
 		if err == backend.ErrNamespaceNotFound {
-			log.With(
+			log.WithTraceContext(ctx).With(
 				"namespace", t.Namespace,
-				"name", t.Name).Info("Dropping writeback for unconfigured namespace")
+				"name", t.Name,
+			).Info("Dropping writeback for unconfigured namespace")
 			return nil
 		}
+		log.WithTraceContext(ctx).With(
+			"namespace", t.Namespace,
+			"name", t.Name,
+			"error", err,
+		).Error("Failed to get backend client")
 		return fmt.Errorf("get client: %s", err)
 	}
 
 	if _, err := client.Stat(t.Namespace, t.Name); err == nil {
 		// File already uploaded, no-op.
+		log.WithTraceContext(ctx).With(
+			"namespace", t.Namespace,
+			"name", t.Name,
+		).Debug("File already exists in backend, skipping upload")
 		return nil
 	}
 
@@ -100,17 +161,39 @@ func (e *Executor) upload(t *Task) error {
 		if os.IsNotExist(err) {
 			// Nothing we can do about this but make noise and drop the task.
 			e.stats.Counter("missing_files").Inc(1)
-			log.With("name", t.Name).Error("Invariant violation: writeback cache file missing")
+			log.WithTraceContext(ctx).With(
+				"namespace", t.Namespace,
+				"name", t.Name,
+			).Error("Invariant violation: writeback cache file missing")
 			return nil
 		}
+		log.WithTraceContext(ctx).With(
+			"namespace", t.Namespace,
+			"name", t.Name,
+			"error", err,
+		).Error("Failed to get cache file reader")
 		return fmt.Errorf("get file: %s", err)
 	}
 	defer closers.Close(f)
 
+	log.WithTraceContext(ctx).With(
+		"namespace", t.Namespace,
+		"name", t.Name,
+	).Debug("Starting backend upload")
+
 	if err := client.Upload(t.Namespace, t.Name, f); err != nil {
+		log.WithTraceContext(ctx).With(
+			"namespace", t.Namespace,
+			"name", t.Name,
+			"error", err,
+		).Error("Backend upload failed")
 		return fmt.Errorf("upload: %s", err)
 	}
-	log.With("namespace", t.Namespace, "name", t.Name).Info("Uploaded cache file to remote backend")
+
+	log.WithTraceContext(ctx).With(
+		"namespace", t.Namespace,
+		"name", t.Name,
+	).Info("Uploaded cache file to remote backend")
 
 	// We don't want to time noops nor errors.
 	e.stats.Timer("upload").Record(time.Since(start))