Implement span processor's OnEnding

CHANGELOG.md

@@ -8,6 +8,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+### Added
+
+- Support for the `OnEnding` callback in span processors. (#5756)
+
 ### Fixed
 
 - Fix memory leak in the global `MeterProvider` when identical instruments are repeatedly created. (#5754)

sdk/internal/x/onending_processor.go

@@ -0,0 +1,14 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package x // import "go.opentelemetry.io/otel/sdk/internal/x"
+
+import "go.opentelemetry.io/otel/trace"
+
+// OnEndingSpanProcessor represents span processors that allow mutating spans
+// just before they are ended and made immutable.
+type OnEndingSpanProcessor interface {
+	// OnEnding is called while the span is finished, and while spans are still
+	// mutable. It is called synchronously and cannot block.
+	OnEnding(trace.Span)
+}

sdk/trace/span.go

@@ -19,6 +19,7 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/internal/global"
 	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/internal/x"
 	"go.opentelemetry.io/otel/sdk/resource"
 	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 	"go.opentelemetry.io/otel/trace"
@@ -120,6 +121,9 @@ type recordingSpan struct {
 	// value of time.Time until the span is ended.
 	endTime time.Time
 
+	// hasEnded records whether the span is fully ended.
+	hasEnded bool
+
 	// status is the status of this span.
 	status Status
 
@@ -171,10 +175,8 @@ func (s *recordingSpan) IsRecording() bool {
 	if s == nil {
 		return false
 	}
-	s.mu.Lock()
-	defer s.mu.Unlock()
 
-	return s.endTime.IsZero()
+	return !s.hasEnded
 }
 
 // SetStatus sets the status of the Span in the form of a code and a
@@ -417,7 +419,6 @@ func (s *recordingSpan) End(options ...trace.SpanEndOption) {
 	}
 
 	s.mu.Lock()
-	// Setting endTime to non-zero marks the span as ended and not recording.
 	if config.Timestamp().IsZero() {
 		s.endTime = et
 	} else {
@@ -426,6 +427,15 @@ func (s *recordingSpan) End(options ...trace.SpanEndOption) {
 	s.mu.Unlock()
 
 	sps := s.tracer.provider.getSpanProcessors()
+	for _, sp := range sps {
+		if oesp, ok := sp.sp.(x.OnEndingSpanProcessor); ok {
+			oesp.OnEnding(s)
+		}
+	}
+	s.mu.Lock()
+	s.hasEnded = true
+	s.mu.Unlock()
+
 	if len(sps) == 0 {
 		return
 	}

sdk/trace/span_processor_test.go

@@ -46,6 +46,13 @@ func (t *testSpanProcessor) OnStart(parent context.Context, s sdktrace.ReadWrite
 	t.spansStarted = append(t.spansStarted, s)
 }
 
+func (t *testSpanProcessor) OnEnding(s trace.Span) {
+	if t == nil {
+		return
+	}
+	s.SetAttributes(attribute.Bool("OnEnding", true))
+}
+
 func (t *testSpanProcessor) OnEnd(s sdktrace.ReadOnlySpan) {
 	if t == nil {
 		return
@@ -130,6 +137,17 @@ func TestRegisterSpanProcessor(t *testing.T) {
 				}
 			}
 		}
+
+		onEndingOK := false
+		for _, kv := range sp.spansEnded[0].Attributes() {
+			switch kv.Key {
+			case "OnEnding":
+				onEndingOK = true
+			default:
+				continue
+			}
+		}
+
 		if c != len(spNames) {
 			t.Errorf("%s: expected attributes(SpanProcessorName): got %d, want %d\n", name, c, len(spNames))
 		}
@@ -139,6 +157,9 @@ func TestRegisterSpanProcessor(t *testing.T) {
 		if !sidOK {
 			t.Errorf("%s: expected attributes(ParentSpanID)\n", name)
 		}
+		if !onEndingOK {
+			t.Errorf("%s: expected attributes(OnEnding)\n", name)
+		}
 	}
 }