mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Javier Molina ReyesandGitHub
a7dde3a25b
* enable return partial traces * fix bug returning an error on max size reached * test combiner with partial trace * allow partial traces to be returned from the ingester * add message to TraceByIdV2 on partial trace returned * fix test and move allowtraces one level down * allow partial traces in segment decoder * fix compilation error * use an enum to describe better the trace response * revert segment decoder changes * do not check the trace size at block level * changelog * move short-circuit to the top of the method * simplify partial trace logic * added missing changelog * perf improvement by not processing partial traces from queriers * do not replace the trace every time * flag when querier returns a partial trace * fix proto
80 lines
1.8 KiB
Go
80 lines
1.8 KiB
Go
package v1
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/gogo/protobuf/proto"
|
|
"github.com/grafana/tempo/pkg/model/decoder"
|
|
"github.com/grafana/tempo/pkg/model/trace"
|
|
"github.com/grafana/tempo/pkg/tempopb"
|
|
)
|
|
|
|
const Encoding = "v1"
|
|
|
|
type ObjectDecoder struct{}
|
|
|
|
var staticDecoder = &ObjectDecoder{}
|
|
|
|
func NewObjectDecoder() *ObjectDecoder {
|
|
return staticDecoder
|
|
}
|
|
|
|
func (d *ObjectDecoder) PrepareForRead(obj []byte) (*tempopb.Trace, error) {
|
|
trace := &tempopb.Trace{}
|
|
traceBytes := &tempopb.TraceBytes{}
|
|
err := proto.Unmarshal(obj, traceBytes)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, bytes := range traceBytes.Traces {
|
|
innerTrace := &tempopb.Trace{}
|
|
err = proto.Unmarshal(bytes, innerTrace)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
trace.ResourceSpans = append(trace.ResourceSpans, innerTrace.ResourceSpans...)
|
|
}
|
|
return trace, err
|
|
}
|
|
|
|
func (d *ObjectDecoder) Combine(objs ...[]byte) ([]byte, error) {
|
|
c := trace.NewCombiner(0, false)
|
|
for i, obj := range objs {
|
|
t, err := staticDecoder.PrepareForRead(obj)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error unmarshaling trace: %w", err)
|
|
}
|
|
|
|
_, err = c.ConsumeWithFinal(t, i == len(obj)-1)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error combining trace: %w", err)
|
|
}
|
|
}
|
|
combinedTrace, _ := c.Result()
|
|
|
|
combinedBytes, err := d.Marshal(combinedTrace)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error marshaling combinedBytes: %w", err)
|
|
}
|
|
|
|
return combinedBytes, nil
|
|
}
|
|
|
|
func (d *ObjectDecoder) FastRange([]byte) (uint32, uint32, error) {
|
|
return 0, 0, decoder.ErrUnsupported
|
|
}
|
|
|
|
func (d *ObjectDecoder) Marshal(t *tempopb.Trace) ([]byte, error) {
|
|
traceBytes := &tempopb.TraceBytes{}
|
|
bytes, err := proto.Marshal(t)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
traceBytes.Traces = append(traceBytes.Traces, bytes)
|
|
|
|
return proto.Marshal(traceBytes)
|
|
}
|