mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Oleg V. KozlyukandGitHub
b1dc89c38b
* Add support for protobuf format in all endpoints * go fmt * Added CHANGELOG entry * Added explicit MarshallingFormat type * Finalize refactoring - use MarshallingFormat type everywhere * Take review notes into account * Reduce code duplication * More review notes
58 lines
1.9 KiB
Go
58 lines
1.9 KiB
Go
package combiner
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/grafana/tempo/pkg/api"
|
|
"github.com/grafana/tempo/pkg/model/trace"
|
|
"github.com/grafana/tempo/pkg/tempopb"
|
|
)
|
|
|
|
func NewTypedTraceByIDV2(maxBytes int, marshalingFormat api.MarshallingFormat, traceRedactor TraceRedactor) GRPCCombiner[*tempopb.TraceByIDResponse] {
|
|
return NewTraceByIDV2(maxBytes, marshalingFormat, traceRedactor).(GRPCCombiner[*tempopb.TraceByIDResponse])
|
|
}
|
|
|
|
func NewTraceByIDV2(maxBytes int, marshalingFormat api.MarshallingFormat, traceRedactor TraceRedactor) Combiner {
|
|
combiner := trace.NewCombiner(maxBytes, true)
|
|
var partialTrace bool
|
|
metricsCombiner := NewTraceByIDMetricsCombiner()
|
|
gc := &genericCombiner[*tempopb.TraceByIDResponse]{
|
|
combine: func(partial *tempopb.TraceByIDResponse, _ *tempopb.TraceByIDResponse, pipelineResp PipelineResponse) error {
|
|
if partial.Status == tempopb.PartialStatus_PARTIAL {
|
|
partialTrace = true
|
|
}
|
|
|
|
metricsCombiner.Combine(partial.Metrics, pipelineResp)
|
|
|
|
_, err := combiner.Consume(partial.Trace)
|
|
return err
|
|
},
|
|
finalize: func(resp *tempopb.TraceByIDResponse) (*tempopb.TraceByIDResponse, error) {
|
|
traceResult, _ := combiner.Result()
|
|
if traceResult == nil {
|
|
traceResult = &tempopb.Trace{}
|
|
}
|
|
|
|
// dedupe duplicate span ids
|
|
deduper := newDeduper()
|
|
traceResult = deduper.dedupe(traceResult)
|
|
if traceRedactor != nil {
|
|
traceRedactor.RedactTraceAttributes(traceResult)
|
|
}
|
|
resp.Trace = traceResult
|
|
resp.Metrics = metricsCombiner.Metrics
|
|
|
|
if partialTrace || combiner.IsPartialTrace() {
|
|
resp.Status = tempopb.PartialStatus_PARTIAL
|
|
resp.Message = fmt.Sprintf("Trace exceeds maximum size of %d bytes, a partial trace is returned", maxBytes)
|
|
}
|
|
|
|
return resp, nil
|
|
},
|
|
new: func() *tempopb.TraceByIDResponse { return &tempopb.TraceByIDResponse{} },
|
|
current: &tempopb.TraceByIDResponse{},
|
|
}
|
|
initHTTPCombiner(gc, marshalingFormat)
|
|
return gc
|
|
}
|