Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Drop tag cache 2 #1

Merged
merged 2 commits into from
Oct 28, 2021
Merged

Drop tag cache 2 #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
16570c7
Reuse SearchEntry buffer where possible
mdisibio Oct 27, 2021
ab3b5b4
fix key/value typo
mdisibio Oct 27, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions modules/ingester/instance_search.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,8 @@ func (i *instance) searchLiveTraces(ctx context.Context, p search.Pipeline, sr *

span.LogFields(ot_log.Event("live traces mtx acquired"))

entry := &tempofb.SearchEntry{} // buffer

for _, t := range i.traces {
if sr.Quit() {
return
Expand All @@ -105,7 +107,7 @@ func (i *instance) searchLiveTraces(ctx context.Context, p search.Pipeline, sr *
for _, s := range t.searchData {
sr.AddBytesInspected(uint64(len(s)))

entry := tempofb.SearchEntryFromBytes(s)
entry.Reset(s)
if p.Matches(entry) {
newResult := search.GetSearchResultFromData(entry)
if result != nil {
Expand Down Expand Up @@ -187,9 +189,10 @@ func (i *instance) SearchTags(ctx context.Context) (*tempopb.SearchTagsResponse,
err := i.visitSearchEntriesLiveTraces(ctx, func(entry *tempofb.SearchEntry) {
for i, ii := 0, entry.TagsLength(); i < ii; i++ {
entry.Tags(kv, i)
key := string(kv.Key())
// check the tag is already set, this is more performant with repetitive values
if _, ok := tags[string(kv.Key())]; !ok {
tags[string(kv.Key())] = struct{}{}
if _, ok := tags[key]; !ok {
tags[key] = struct{}{}
}
}
})
Expand Down Expand Up @@ -249,9 +252,11 @@ func (i *instance) visitSearchEntriesLiveTraces(ctx context.Context, visitFn fun
i.tracesMtx.Lock()
defer i.tracesMtx.Unlock()

se := &tempofb.SearchEntry{}
for _, t := range i.traces {
for _, s := range t.searchData {
visitFn(tempofb.SearchEntryFromBytes(s))
se.Reset(s)
visitFn(se)

if err := ctx.Err(); err != nil {
return err
Expand Down
2 changes: 1 addition & 1 deletion pkg/tempofb/searchdata_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ func TestContainsTag(t *testing.T) {
m.AddTag("key5", "value")
m.AddTag("key6", "value")

e := SearchEntryFromBytes(m.ToBytes())
e := NewSearchEntryFromBytes(m.ToBytes())

kv := &KeyValues{}

Expand Down
7 changes: 6 additions & 1 deletion pkg/tempofb/searchdata_util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,12 @@ func (s *SearchEntry) Contains(k []byte, v []byte, buffer *KeyValues) bool {
return ContainsTag(s, buffer, k, v)
}

func SearchEntryFromBytes(b []byte) *SearchEntry {
func (s *SearchEntry) Reset(b []byte) {
n := flatbuffers.GetUOffsetT(b)
s.Init(b, n)
}

func NewSearchEntryFromBytes(b []byte) *SearchEntry {
return GetRootAsSearchEntry(b, 0)
}

Expand Down
5 changes: 3 additions & 2 deletions tempodb/search/backend_search_block.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@ func NewBackendSearchBlock(input *StreamingSearchBlock, rw backend.Writer, block
var err error
ctx := context.TODO()
indexPageSize := 100 * 1024
kv := &tempofb.KeyValues{} // buffer
kv := &tempofb.KeyValues{} // buffer
s := &tempofb.SearchEntry{} // buffer

// Pinning specific version instead of latest for safety
version, err := encoding.FromVersion("v2")
Expand Down Expand Up @@ -69,7 +70,7 @@ func NewBackendSearchBlock(input *StreamingSearchBlock, rw backend.Writer, block
continue
}

s := tempofb.SearchEntryFromBytes(data)
s.Reset(data)

header.AddEntry(s)

Expand Down
5 changes: 3 additions & 2 deletions tempodb/search/data_combiner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,13 +23,14 @@ func (*DataCombiner) Combine(_ string, searchData ...[]byte) ([]byte, bool) {

// Squash all datas into 1
data := tempofb.SearchEntryMutable{}
kv := &tempofb.KeyValues{} // buffer
kv := &tempofb.KeyValues{} // buffer
sd := &tempofb.SearchEntry{} // buffer
for _, sb := range searchData {
// we append zero-length entries to the WAL even when search is disabled. skipping to prevent unmarshalling and panik :)
if len(sb) == 0 {
continue
}
sd := tempofb.SearchEntryFromBytes(sb)
sd.Reset(sb)
for i, ii := 0, sd.TagsLength(); i < ii; i++ {
sd.Tags(kv, i)
for j, jj := 0, kv.ValueLength(); j < jj; j++ {
Expand Down
6 changes: 3 additions & 3 deletions tempodb/search/pipeline_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ func TestPipelineMatchesTags(t *testing.T) {
data := tempofb.SearchEntryMutable{
Tags: tempofb.NewSearchDataMapWithData(tc.searchData),
}
sd := tempofb.SearchEntryFromBytes(data.ToBytes())
sd := tempofb.NewSearchEntryFromBytes(data.ToBytes())
matches := p.Matches(sd)

require.Equal(t, tc.shouldMatch, matches)
Expand Down Expand Up @@ -125,7 +125,7 @@ func TestPipelineMatchesTraceDuration(t *testing.T) {
StartTimeUnixNano: uint64(tc.spanStart),
EndTimeUnixNano: uint64(tc.spanEnd),
}
sd := tempofb.SearchEntryFromBytes(data.ToBytes())
sd := tempofb.NewSearchEntryFromBytes(data.ToBytes())
matches := p.Matches(sd)

require.Equal(t, tc.shouldMatch, matches)
Expand Down Expand Up @@ -185,7 +185,7 @@ func TestPipelineMatchesBlock(t *testing.T) {

func BenchmarkPipelineMatches(b *testing.B) {

entry := tempofb.SearchEntryFromBytes((&tempofb.SearchEntryMutable{
entry := tempofb.NewSearchEntryFromBytes((&tempofb.SearchEntryMutable{
StartTimeUnixNano: 0,
EndTimeUnixNano: uint64(500 * time.Millisecond / time.Nanosecond), //500ms in nanoseconds
Tags: tempofb.NewSearchDataMapWithData(map[string][]string{
Expand Down
2 changes: 1 addition & 1 deletion tempodb/search/rescan_blocks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ func newStreamingSearchBlockFromWALReplay(searchFilepath, filename string) (*Str

blockHeader := tempofb.NewSearchBlockHeaderMutable()
records, warning, err := wal.ReplayWALAndGetRecords(f, v, enc, func(bytes []byte) error {
entry := tempofb.SearchEntryFromBytes(bytes)
entry := tempofb.NewSearchEntryFromBytes(bytes)
blockHeader.AddEntry(entry)
return nil
})
Expand Down
6 changes: 4 additions & 2 deletions tempodb/search/streaming_search_block.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ func (s *StreamingSearchBlock) Append(ctx context.Context, id common.ID, searchD
}

s.headerMtx.Lock()
s.header.AddEntry(tempofb.SearchEntryFromBytes(combined))
s.header.AddEntry(tempofb.NewSearchEntryFromBytes(combined))
s.headerMtx.Unlock()

return s.appender.Append(id, combined)
Expand Down Expand Up @@ -111,6 +111,8 @@ func (s *StreamingSearchBlock) TagValues(ctx context.Context, tagName string, ta

// Search the streaming block.
func (s *StreamingSearchBlock) Search(ctx context.Context, p Pipeline, sr *Results) error {
entry := &tempofb.SearchEntry{}

if s.closed.Load() {
// Generally this means block has already been deleted
return nil
Expand Down Expand Up @@ -149,7 +151,7 @@ func (s *StreamingSearchBlock) Search(ctx context.Context, p Pipeline, sr *Resul
sr.AddBytesInspected(uint64(len(obj)))
sr.AddTraceInspected(1)

entry := tempofb.SearchEntryFromBytes(obj)
entry.Reset(obj)

if !p.Matches(entry) {
continue
Expand Down