Prepare getting rid of CompleteSpan

fhanau · fhanau · commit c5c817feea7a · 2026-02-10T18:14:23.000-05:00
diff --git a/src/workerd/io/trace.c++ b/src/workerd/io/trace.c++
@@ -1516,14 +1516,11 @@ CompleteSpan::CompleteSpan(rpc::UserSpanData::Reader reader)
   }
 }
 
-CompleteSpan CompleteSpan::clone() const {
-  CompleteSpan copy(spanId, parentSpanId, operationName.clone(), startTime, endTime);
-  copy.tags.reserve(tags.size());
-  for (auto& tag: tags) {
-    copy.tags.insert(tag.key.clone(), spanTagClone(tag.value));
-  }
-  return copy;
-}
+SpanOpenData::SpanOpenData(rpc::SpanOpenData::Reader reader)
+    : spanId(reader.getSpanId()),
+      parentSpanId(reader.getParentSpanId()),
+      operationName(kj::str(reader.getOperationName())),
+      startTime(kj::UNIX_EPOCH + reader.getStartTimeNs() * kj::NANOSECONDS) {}
 
 void SpanOpenData::copyTo(rpc::SpanOpenData::Builder builder) const {
   builder.setOperationName(operationName.asPtr());
@@ -1532,12 +1529,37 @@ void SpanOpenData::copyTo(rpc::SpanOpenData::Builder builder) const {
   builder.setParentSpanId(parentSpanId);
 }
 
-SpanOpenData::SpanOpenData(rpc::SpanOpenData::Reader reader)
+SpanEndData::SpanEndData(CompleteSpan&& span)
+    : spanId(span.spanId),
+      startTime(span.startTime),
+      endTime(span.endTime),
+      tags(kj::mv(span.tags)) {}
+
+SpanEndData::SpanEndData(rpc::SpanEndData::Reader reader)
     : spanId(reader.getSpanId()),
-      parentSpanId(reader.getParentSpanId()),
-      operationName(kj::str(reader.getOperationName())),
-      startTime(kj::UNIX_EPOCH + reader.getStartTimeNs() * kj::NANOSECONDS) {}
+      startTime(kj::UNIX_EPOCH + reader.getStartTimeNs() * kj::NANOSECONDS),
+      endTime(kj::UNIX_EPOCH + reader.getEndTimeNs() * kj::NANOSECONDS) {
+  auto tagsParam = reader.getTags();
+  tags.reserve(tagsParam.size());
+  for (auto tagParam: tagsParam) {
+    tags.insert(kj::ConstString(kj::heapString(tagParam.getKey())),
+        deserializeTagValue(tagParam.getValue()));
+  }
+}
+
+void SpanEndData::copyTo(rpc::SpanEndData::Builder builder) const {
+  builder.setStartTimeNs((startTime - kj::UNIX_EPOCH) / kj::NANOSECONDS);
+  builder.setEndTimeNs((endTime - kj::UNIX_EPOCH) / kj::NANOSECONDS);
+  builder.setSpanId(spanId);
 
+  auto tagsParam = builder.initTags(tags.size());
+  auto i = 0;
+  for (auto& tag: tags) {
+    auto tagParam = tagsParam[i++];
+    tagParam.setKey(tag.key.asPtr());
+    serializeTagValue(tagParam.initValue(), tag.value);
+  }
+}
 }  // namespace tracing
 
 // ======================================================================================
diff --git a/src/workerd/io/trace.capnp b/src/workerd/io/trace.capnp
@@ -43,7 +43,7 @@ struct UserSpanData {
 }
 
 struct SpanOpenData {
-  # Representation of a SpanOpen event
+  # Representation of a SpanOpen event, created when a user span is opened.
   operationName @0 :Text;
 
   startTimeNs @1 :Int64;
@@ -55,15 +55,15 @@ struct SpanOpenData {
 
 struct SpanEndData {
   # Representation of an event that indicates completion of a user span. This information is
-  # provided to the tail worker in the Attributes and SpanClose events.
+  # provided to the streaming tail worker in the Attributes and SpanClose events.
 
-  # TODO: These can probably go?
-  operationName @0 :Text;
-  startTimeNs @1 :Int64;
+  # TODO(cleanup): startTimeNs is merely used as a fallback timestamp, consider obsoleting it.
+  startTimeNs @0 :Int64;
   # Nanoseconds since Unix epoch
-  endTimeNs @2 :Int64;
+  endTimeNs @1 :Int64;
   # Nanoseconds since Unix epoch
 
-  tags @3 :List(Tag);
-  spanId @4 :UInt64;
-}
+  # List of span attributes
+  tags @2 :List(Tag);
+  spanId @3 :UInt64;
+}
diff --git a/src/workerd/io/trace.h b/src/workerd/io/trace.h
@@ -632,8 +632,6 @@ struct CompleteSpan {
 
   CompleteSpan(rpc::UserSpanData::Reader reader);
   void copyTo(rpc::UserSpanData::Builder builder) const;
-  // TODO: Is clone() still needed?
-  CompleteSpan clone() const;
   explicit CompleteSpan(tracing::SpanId spanId,
       tracing::SpanId parentSpanId,
       kj::ConstString operationName,
@@ -669,6 +667,30 @@ struct SpanOpenData {
         startTime(startTime) {}
 };
 
+struct SpanEndData {
+  // Represents the data needed when closing a span, including the Attributes and SpanClose events.
+  tracing::SpanId spanId;
+
+  kj::Date startTime;
+  kj::Date endTime;
+  // Should be Span::TagMap, but we can't forward-declare that.
+  kj::HashMap<kj::ConstString, tracing::Attribute::Value> tags;
+
+  // Convert CompleteSpan to SpanEndData
+  explicit SpanEndData(CompleteSpan&& span);
+  SpanEndData(rpc::SpanEndData::Reader reader);
+  void copyTo(rpc::SpanEndData::Builder builder) const;
+  explicit SpanEndData(tracing::SpanId spanId,
+      kj::Date startTime,
+      kj::Date endTime,
+      kj::HashMap<kj::ConstString, tracing::Attribute::Value> tags =
+          kj::HashMap<kj::ConstString, tracing::Attribute::Value>())
+      : spanId(spanId),
+        startTime(startTime),
+        endTime(endTime),
+        tags(kj::mv(tags)) {}
+};
+
 // A Return mark is used to mark the point at which a span operation returned
 // a value. For instance, when a fetch subrequest response is received, or when
 // the fetch handler returns a Response. Importantly, it does not signal that the
@@ -1132,10 +1154,10 @@ class SpanObserver: public kj::Refcounted {
 
   // Report the span data. Called at the end of the span.
   //
-  // This should always be called exactly once per observer.
+  // This should always be called exactly once per observer at span completion time.
   virtual void report(const Span& span) = 0;
+  // Report information about the span onset.
   virtual void reportStart(kj::ConstString& operationName, kj::Date startTime) = 0;
-  virtual void reportEnd(const Span& span) = 0;
 
   // The current time to be provided for the span. For user tracing, we will override this to
   // provide I/O time. This *requires* that spans are only created when an IOContext is available
diff --git a/src/workerd/io/tracer.c++ b/src/workerd/io/tracer.c++
@@ -124,71 +124,17 @@ void WorkerTracer::addLog(const tracing::InvocationSpanContext& context,
 }
 
 void WorkerTracer::addSpan(tracing::CompleteSpan&& span) {
+  // The span information is not transmitted via RPC at this point, we can decompose the span into
+  // spanOpen/spanEnd.
   addSpanOpen(span.spanId, span.parentSpanId, span.operationName, span.startTime);
-  addSpanEnd(kj::mv(span));
-#if 0
-  // This is where we'll actually encode the span.
-  if (pipelineLogLevel == PipelineLogLevel::NONE) {
-    return;
-  }
-
-  // Note: spans are not available in the buffered tail worker, so we don't need an exceededSpanLimit
-  // variable for it and it can't cause truncation.
-  auto& tailStreamWriter = KJ_UNWRAP_OR_RETURN(maybeTailStreamWriter);
-
-  adjustSpanTime(span);
-
-  size_t spanTagsSize = 0;
-  size_t spanNameSize = span.operationName.size();
-  for (const Span::TagMap::Entry& tag: span.tags) {
-    spanTagsSize += tag.key.size();
-    KJ_SWITCH_ONEOF(tag.value) {
-      KJ_CASE_ONEOF(str, kj::ConstString) {
-        spanTagsSize += str.size();
-      }
-      KJ_CASE_ONEOF(val, bool) {
-        spanTagsSize++;
-      }
-      // int64_t and double
-      KJ_CASE_ONEOF_DEFAULT {
-        spanTagsSize += sizeof(int64_t);
-      }
-    }
-  }
-
-  // Compose span events – attributes and spanClose are transmitted together for now.
-  auto& topLevelContext = KJ_ASSERT_NONNULL(topLevelInvocationSpanContext);
-  // Compose span events. For SpanOpen, an all-zero spanId is interpreted as having no spans above
-  // this one, thus we use the Onset spanId instead (taken from topLevelContext). We go to great
-  // lengths to rule out getting an all-zero spanId by chance (see SpanId::fromEntropy()), so this
-  // should be safe.
-  tracing::SpanId parentSpanId = span.parentSpanId;
-  if (parentSpanId == tracing::SpanId::nullId) {
-    parentSpanId = topLevelContext.getSpanId();
-  }
-
-  auto spanOpenContext = tracing::InvocationSpanContext(
-    topLevelContext.getTraceId(), topLevelContext.getInvocationId(), parentSpanId);
-  auto spanComponentContext = tracing::InvocationSpanContext(
-    topLevelContext.getTraceId(), topLevelContext.getInvocationId(), span.spanId);
-  tailStreamWriter->report(spanOpenContext,
-      tracing::SpanOpen(span.spanId, span.operationName.clone()), span.startTime, spanNameSize);
-  // If a span manages to exceed the size limit, truncate it by not providing span attributes.
-  if (span.tags.size() && spanTagsSize <= MAX_TRACE_BYTES) {
-    tracing::CustomInfo attr = KJ_MAP(tag, span.tags) {
-      return tracing::Attribute(kj::mv(tag.key), kj::mv(tag.value));
-    };
-    tailStreamWriter->report(spanComponentContext, kj::mv(attr), span.startTime, spanTagsSize);
-  }
-  tailStreamWriter->report(spanComponentContext, tracing::SpanClose(), span.endTime, 0);
-#endif
+  tracing::SpanEndData spanEnd(kj::mv(span));
+  addSpanEnd(kj::mv(spanEnd));
 }
 
 void WorkerTracer::addSpanOpen(tracing::SpanId spanId,
     tracing::SpanId parentSpanId,
     kj::ConstString& operationName,
     kj::Date startTime) {
-  // This is where we'll actually encode the span.
   if (pipelineLogLevel == PipelineLogLevel::NONE) {
     return;
   }
@@ -208,8 +154,7 @@ void WorkerTracer::addSpanOpen(tracing::SpanId spanId,
       spanOpenContext, tracing::SpanOpen(spanId, operationName.clone()), startTime, spanNameSize);
 }
 
-void WorkerTracer::addSpanEnd(tracing::CompleteSpan&& span) {
-  // This is where we'll actually encode the span.
+void WorkerTracer::addSpanEnd(tracing::SpanEndData&& span) {
   if (pipelineLogLevel == PipelineLogLevel::NONE) {
     return;
   }
@@ -237,7 +182,8 @@ void WorkerTracer::addSpanEnd(tracing::CompleteSpan&& span) {
     }
   }
 
-  // Compose span events – attributes and spanClose are transmitted together for now.
+  // Compose Attributes and SpanClose, which are available at span completion time and transmitted
+  // together.
   auto& topLevelContext = KJ_ASSERT_NONNULL(topLevelInvocationSpanContext);
   auto spanComponentContext = tracing::InvocationSpanContext(
       topLevelContext.getTraceId(), topLevelContext.getInvocationId(), span.spanId);
@@ -513,6 +459,56 @@ void BaseTracer::adjustSpanTime(tracing::CompleteSpan& span) {
   }
 }
 
+void BaseTracer::adjustSpanTime(tracing::SpanEndData& span) {
+  // To report I/O time, we need the IOContext to still be alive.
+  // weakIoContext is only none if we are tracing via RPC (in this case span times have already been
+  // adjusted) or if we failed to transmit an Onset event (in that case we'll get an error based on
+  // missing topLevelInvocationSpanContext right after).
+  if (weakIoContext != kj::none) {
+    auto& weakIoCtx = KJ_ASSERT_NONNULL(weakIoContext);
+    weakIoCtx->runIfAlive([this, &span](IoContext& context) {
+      if (context.hasCurrentIncomingRequest()) {
+        span.endTime = context.now();
+      } else {
+        // We have an IOContext, but there's no current IncomingRequest. Always log a warning here,
+        // this should not be happening. Still report completeTime as a useful timestamp if
+        // available.
+        bool hasCompleteTime = false;
+        if (completeTime != kj::UNIX_EPOCH) {
+          span.endTime = completeTime;
+          hasCompleteTime = true;
+        } else {
+          span.endTime = span.startTime;
+        }
+        if (isPredictableModeForTest()) {
+          KJ_FAIL_ASSERT("reported span without current request", hasCompleteTime);
+        } else {
+          LOG_WARNING_PERIODICALLY("reported span without current request");
+        }
+      }
+    });
+    if (!weakIoCtx->isValid()) {
+      // This can happen if we start a customEvent from this event and cancel it after this IoContext
+      // gets destroyed. In that case we no longer have an IoContext available and can't get the
+      // current time, but the outcome timestamp will have already been set. Since the outcome
+      // timestamp is "late enough", simply use that.
+      // TODO(o11y): fix this – spans should not be outliving the IoContext.
+      if (completeTime != kj::UNIX_EPOCH) {
+        span.endTime = completeTime;
+      } else {
+        // Otherwise, we can't actually get an end timestamp that makes sense. Report a zero-duration
+        // span and log a warning (or fail assert in test mode).
+        span.endTime = span.startTime;
+        if (isPredictableModeForTest()) {
+          KJ_FAIL_ASSERT("reported span after IoContext was deallocated");
+        } else {
+          KJ_LOG(WARNING, "reported span after IoContext was deallocated");
+        }
+      }
+    }
+  }
+}
+
 void WorkerTracer::setReturn(
     kj::Maybe<kj::Date> timestamp, kj::Maybe<tracing::FetchResponseInfo> fetchResponseInfo) {
   // Match the behavior of setEventInfo(). Any resolution of the TODO comments in setEventInfo()
@@ -588,11 +584,7 @@ void UserSpanObserver::report(const Span& span) {
 }
 
 void UserSpanObserver::reportStart(kj::ConstString& operationName, kj::Date startTime) {
-  submitter->submitSpanStart(spanId, parentSpanId, operationName, startTime);
-}
-
-void UserSpanObserver::reportEnd(const Span& span) {
-  submitter->submitSpanEnd(spanId, span);
+  submitter->submitSpanOpen(spanId, parentSpanId, operationName, startTime);
 }
 
 // Provide I/O time to the tracing system for user spans.
diff --git a/src/workerd/io/tracer.h b/src/workerd/io/tracer.h
@@ -31,13 +31,15 @@ class BaseTracer: public kj::Refcounted {
       kj::Date timestamp,
       LogLevel logLevel,
       kj::String message) = 0;
-  // Add a span.
+  // Add a complete span.
   virtual void addSpan(tracing::CompleteSpan&& span) = 0;
+  // Add information about a span when it is opened, corresponds to SpanOpen event.
   virtual void addSpanOpen(tracing::SpanId spanId,
       tracing::SpanId parentSpanId,
       kj::ConstString& operationName,
       kj::Date startTime) = 0;
-  virtual void addSpanEnd(tracing::CompleteSpan&& span) = 0;
+  // Add span events when the span is complete (Attributes and SpanClose).
+  virtual void addSpanEnd(tracing::SpanEndData&& span) = 0;
 
   virtual void addException(const tracing::InvocationSpanContext& context,
       kj::Date timestamp,
@@ -95,6 +97,7 @@ class BaseTracer: public kj::Refcounted {
 
   // helper method for addSpan() implementations
   void adjustSpanTime(tracing::CompleteSpan& span);
+  void adjustSpanTime(tracing::SpanEndData& span);
 
   // Function to create the root span for the new tracing format.
   kj::Maybe<MakeUserRequestSpanFunc> makeUserRequestSpanFunc;
@@ -135,7 +138,7 @@ class WorkerTracer final: public BaseTracer {
       tracing::SpanId parentSpanId,
       kj::ConstString& operationName,
       kj::Date startTime) override;
-  void addSpanEnd(tracing::CompleteSpan&& span) override;
+  void addSpanEnd(tracing::SpanEndData&& span) override;
   void addException(const tracing::InvocationSpanContext& context,
       kj::Date timestamp,
       kj::String name,
@@ -192,7 +195,7 @@ class WorkerTracer final: public BaseTracer {
 class SpanSubmitter: public kj::Refcounted {
  public:
   virtual void submitSpan(tracing::SpanId context, tracing::SpanId spanId, const Span& span) = 0;
-  virtual void submitSpanStart(tracing::SpanId spanId,
+  virtual void submitSpanOpen(tracing::SpanId spanId,
       tracing::SpanId parentSpanId,
       kj::ConstString& operationName,
       kj::Date startTime) = 0;
@@ -219,7 +222,6 @@ class UserSpanObserver final: public SpanObserver {
   kj::Own<SpanObserver> newChild() override;
   void report(const Span& span) override;
   void reportStart(kj::ConstString& operationName, kj::Date startTime) override;
-  void reportEnd(const Span& span) override;
   kj::Date getTime() override;
 
  private:
diff --git a/src/workerd/server/server.c++ b/src/workerd/server/server.c++
@@ -1701,14 +1701,14 @@ class SequentialSpanSubmitter final: public SpanSubmitter {
  public:
   SequentialSpanSubmitter(kj::Own<WorkerTracer> workerTracer): workerTracer(kj::mv(workerTracer)) {}
   void submitSpan(tracing::SpanId spanId, tracing::SpanId parentSpanId, const Span& span) override {
-    // We will soon report SpanOpen and SpanClose separately. In workerd, we can already use this to
-    // implement submitSpan using submitSpanStart and submitSpanEnd.
+    // This code path is workerd-only, we can safely decompose this span into its components and
+    // call submitSpanOpen/submitSpanEnd instead of reimplementing them here.
     kj::ConstString blah = span.operationName.clone();
-    submitSpanStart(spanId, parentSpanId, blah, span.startTime);
+    submitSpanOpen(spanId, parentSpanId, blah, span.startTime);
     submitSpanEnd(spanId, span);
   }
 
-  void submitSpanStart(tracing::SpanId spanId,
+  void submitSpanOpen(tracing::SpanId spanId,
       tracing::SpanId parentSpanId,
       kj::ConstString& operationName,
       kj::Date startTime) override {
@@ -1719,10 +1719,7 @@ class SequentialSpanSubmitter final: public SpanSubmitter {
   }
 
   void submitSpanEnd(tracing::SpanId spanId, const Span& span) override {
-    // We largely recreate the span here which feels inefficient, but is hard to avoid given the
-    // mismatch between the Span type and the full span information required for OTel.
-    tracing::CompleteSpan span2(
-        spanId, tracing::SpanId::nullId, span.operationName.clone(), span.startTime, span.endTime);
+    tracing::SpanEndData span2(spanId, span.startTime, span.endTime);
     span2.tags.reserve(span.tags.size());
     for (auto& tag: span.tags) {
       span2.tags.insert(tag.key.clone(), spanTagClone(tag.value));
