Function requestsByTraceID

modules/distributor/distributor.go:664–786

requestsByTraceID groups ResourceSpans by trace ID, producing hash-ring tokens and rebatched traces for downstream write-path processing. It truncates oversized attributes and returns the first truncation example (if any) for diagnostic logging.

(batches []*v1.ResourceSpans, userID string, spanCount, maxSpanAttrSize int)

// rebatched traces for downstream write-path processing. It truncates oversized attributes
// and returns the first truncation example (if any) for diagnostic logging.
func requestsByTraceID(batches []*v1.ResourceSpans, userID string, spanCount, maxSpanAttrSize int) ([]uint32, []*rebatchedTrace, truncatedAttributesCount, *truncatedAttrInfo, error) {
	const tracesPerBatch = 20 // p50 of internal env
	tracesByID := make(map[uint64]*rebatchedTrace, tracesPerBatch)
	truncatedCount := truncatedAttributesCount{}

	// truncationExample captures one example of a truncated attribute for rate-limited logging.
	var truncationExample truncatedAttrInfo

	currentTime := uint32(time.Now().Unix())
	for _, b := range batches {
		spansByILS := make(map[uint64]*v1.ScopeSpans)
		// check resource for large attributes
		if maxSpanAttrSize > 0 && b.Resource != nil {
			truncatedCount.Resource += processAttributes(b.Resource.Attributes, maxSpanAttrSize, &truncationExample, "resource")
		}

		for _, ils := range b.ScopeSpans {

			// check instrumentation for large attributes
			if maxSpanAttrSize > 0 && ils.Scope != nil {
				truncatedCount.Scope += processAttributes(ils.Scope.Attributes, maxSpanAttrSize, &truncationExample, "scope")
			}

			for _, span := range ils.Spans {
				// check spans for large attributes
				if maxSpanAttrSize > 0 {
					truncatedCount.Span += processAttributes(span.Attributes, maxSpanAttrSize, &truncationExample, "span")

					// check large attributes for events and links
					for _, event := range span.Events {
						truncatedCount.Event += processAttributes(event.Attributes, maxSpanAttrSize, &truncationExample, "event")
					}

					for _, link := range span.Links {
						truncatedCount.Link += processAttributes(link.Attributes, maxSpanAttrSize, &truncationExample, "link")
					}
				}
				traceID := span.TraceId
				if !validation.ValidTraceID(traceID) {
					overrides.RecordDiscardedSpans(spanCount, overrides.ReasonInvalidTraceID, userID)
					return nil, nil, truncatedAttributesCount{}, nil, status.Errorf(codes.InvalidArgument, "trace ids must be 128 bit, received %d bits", len(traceID)*8)
				}

				if !validation.ValidSpanID(span.SpanId) {
					overrides.RecordDiscardedSpans(spanCount, overrides.ReasonInvalidSpanID, userID)
					return nil, nil, truncatedAttributesCount{}, nil, status.Errorf(codes.InvalidArgument, "span ids must be 64 bit and not all zero, received %d bits", len(span.SpanId)*8)
				}

				traceKey := util.HashForTraceID(traceID)
				ilsKey := traceKey
				if ils.Scope != nil {
					ilsKey = fnv1a.AddString64(ilsKey, ils.Scope.Name)
					ilsKey = fnv1a.AddString64(ilsKey, ils.Scope.Version)
				}

				existingILS, ilsAdded := spansByILS[ilsKey]
				if !ilsAdded {
					existingILS = &v1.ScopeSpans{