Merge #75099

75099: tracing: fix corruption of structured recording buffer r=andreimatei a=andreimatei

This patch fixes a bug leading to corruption of one of the Span's
recording buffers.
We had broken code like the following:
```
for _, e := range c.StructuredRecords {
        s.recordInternalLocked(&e, &s.mu.recording.structured)
}
```
`recordInternalLocked(ptr)` takes ownership of ptr, so the `&e` is
broken because `e` keep being reassigned.

The bug was causing buffer elements to essentially be overwritten. In
turn, this was leading to inconsistencies in the accounting of the size
of the buffer's elements - we were maintaining a size sum that was
diverging from the overwritten reality.

Fixes #74994
Fixes #75045

Release note: None

Co-authored-by: Andrei Matei <andrei@cockroachlabs.com>

pkg/util/tracing/crdbspan.go

@@ -349,7 +349,9 @@ func (s *crdbSpan) getStructuredRecording(includeDetachedChildren bool) Recordin
 	return Recording{res}
 }
 
-// recordFinishedChildren adds `children` to the receiver's recording.
+// recordFinishedChildren adds children to s' recording.
+//
+// s takes ownership of children; the caller is not allowed to use them anymore.
 func (s *crdbSpan) recordFinishedChildren(children []tracingpb.RecordedSpan) {
 	if len(children) == 0 {
 		return
@@ -360,6 +362,7 @@ func (s *crdbSpan) recordFinishedChildren(children []tracingpb.RecordedSpan) {
 	s.recordFinishedChildrenLocked(children)
 }
 
+// s takes ownership of children; the caller is not allowed to use them anymore.
 func (s *crdbSpan) recordFinishedChildrenLocked(children []tracingpb.RecordedSpan) {
 	if len(children) == 0 {
 		return
@@ -375,9 +378,10 @@ func (s *crdbSpan) recordFinishedChildrenLocked(children []tracingpb.RecordedSpa
 
 		s.mu.recording.finishedChildren = append(s.mu.recording.finishedChildren, children...)
 	} else {
-		for _, c := range children {
-			for _, e := range c.StructuredRecords {
-				s.recordInternalLocked(&e, &s.mu.recording.structured)
+		for ci := range children {
+			child := &children[ci]
+			for i := range child.StructuredRecords {
+				s.recordInternalLocked(&child.StructuredRecords[i], &s.mu.recording.structured)
 			}
 		}
 	}

pkg/util/tracing/span_test.go

@@ -219,6 +219,27 @@ func TestImportRemoteSpans(t *testing.T) {
 	}
 }
 
+func TestImportRemoteSpansMaintainsRightByteSize(t *testing.T) {
+	tr1 := NewTracer()
+
+	child := tr1.StartSpan("child", WithRecording(RecordingStructured))
+	child.RecordStructured(&types.Int32Value{Value: 42})
+	child.RecordStructured(&types.StringValue{Value: "test"})
+
+	root := tr1.StartSpan("root", WithRecording(RecordingStructured))
+	root.ImportRemoteSpans(child.GetRecording(RecordingStructured))
+	c := root.i.crdb
+	c.mu.Lock()
+	buf := c.mu.recording.structured
+	sz := 0
+	for i := 0; i < buf.Len(); i++ {
+		sz += buf.Get(i).(memorySizable).MemorySize()
+	}
+	c.mu.Unlock()
+	require.NotZero(t, buf.size)
+	require.Equal(t, buf.size, int64(sz))
+}
+
 func TestSpanRecordStructured(t *testing.T) {
 	tr := NewTracer()
 	sp := tr.StartSpan("root", WithRecording(RecordingStructured))