diff --git a/fracmanager/search_stats.go b/fracmanager/search_stats.go new file mode 100644 index 00000000..223e855c --- /dev/null +++ b/fracmanager/search_stats.go @@ -0,0 +1,122 @@ +package fracmanager + +import ( + "fmt" + "time" + + "go.uber.org/zap" + + "github.com/ozontech/seq-db/frac" + "github.com/ozontech/seq-db/querytracer" + "github.com/ozontech/seq-db/seq" +) + +const maxFracsSlowSearchLog = 10 + +type fracTimings struct { + semaphoreWait time.Duration + elapsed []time.Duration +} + +type searchStats struct { + iterations int + + total time.Duration + semaphoreWait time.Duration + + min time.Duration + fastest string + + max time.Duration + slowest string + + matched int + unmatched int + matchedNames []string + unmatchedNames []string +} + +func (s *searchStats) merge(fracs []frac.Fraction, qprs []*seq.QPR, timings fracTimings) { + s.iterations++ + s.semaphoreWait += timings.semaphoreWait + + for i, elapsed := range timings.elapsed { + if s.fracs() == 0 || elapsed < s.min { + s.min = elapsed + s.fastest = fracs[i].Info().Name() + } + + if elapsed > s.max { + s.max = elapsed + s.slowest = fracs[i].Info().Name() + } + + s.total += elapsed + + if qprs[i].Empty() { + s.unmatched++ + if len(s.unmatchedNames) < maxFracsSlowSearchLog { + s.unmatchedNames = append(s.unmatchedNames, fracs[i].Info().Name()) + } + continue + } + + s.matched++ + if len(s.matchedNames) < maxFracsSlowSearchLog { + s.matchedNames = append(s.matchedNames, fracs[i].Info().Name()) + } + } +} + +func (s *searchStats) fracs() int { + return s.matched + s.unmatched +} + +func (s *searchStats) mean() time.Duration { + if s.fracs() == 0 { + return 0 + } + return s.total / time.Duration(s.fracs()) +} + +func (s *searchStats) addToTracer(tr *querytracer.Tracer) { + if !tr.Enabled() { + return + } + + tr.AddChildWithSpan(&querytracer.Span{ + Message: "waiting on the search worker semaphore", + Duration: s.semaphoreWait, + }) + + tr.AddChildWithSpan(&querytracer.Span{ + Message: fmt.Sprintf( + "fraction search time: %d fractions (%d matched) in %d iterations", + s.fracs(), s.matched, s.iterations, + ), + Duration: s.total, + Children: []*querytracer.Span{ + {Message: "mean", Duration: s.mean()}, + {Message: fmt.Sprintf("min (%s)", s.fastest), Duration: s.min}, + {Message: fmt.Sprintf("max (%s)", s.slowest), Duration: s.max}, + }, + }) +} + +func (s *searchStats) zapFields() []zap.Field { + return []zap.Field{ + zap.Int("search_iterations", s.iterations), + zap.Int("fracs_searched", s.fracs()), + zap.Int("fracs_matched", s.matched), + zap.Strings("fracs_matched_sample", s.matchedNames), + zap.Int("fracs_unmatched", s.unmatched), + zap.Strings("fracs_unmatched_sample", s.unmatchedNames), + zap.Int64("semaphore_wait_ms", s.semaphoreWait.Milliseconds()), + zap.Int64("frac_search_total_ms", s.total.Milliseconds()), + zap.Int64("frac_search_min_ms", s.min.Milliseconds()), + zap.String("frac_search_min_frac", s.fastest), + zap.Int64("frac_search_mean_ms", s.mean().Milliseconds()), + zap.Int64("frac_search_max_ms", s.max.Milliseconds()), + zap.String("frac_search_max_frac", s.slowest), + } +} diff --git a/fracmanager/searcher.go b/fracmanager/searcher.go index 0f7e3444..7141e01b 100644 --- a/fracmanager/searcher.go +++ b/fracmanager/searcher.go @@ -19,10 +19,6 @@ import ( "github.com/ozontech/seq-db/util" ) -const ( - maxFracsSlowSearchLog = 10 -) - type SearcherCfg struct { MaxFractionHits int // the maximum number of fractions used in the search FractionsPerIteration int @@ -52,7 +48,6 @@ func (s *Searcher) SearchDocs(ctx context.Context, fracs []frac.Fraction, params return nil, err } - subSearchesCnt := 0 origLimit := params.Limit scanAll := params.IsScanAllRequest() @@ -66,85 +61,53 @@ func (s *Searcher) SearchDocs(ctx context.Context, fracs []frac.Fraction, params fracsChunkSize = len(remainingFracs) } - var totalSearchTimeNanos int64 - var totalWaitTimeNanos int64 - var totalMemUsage int - totalFracsFound := 0 - totalFracsSkipped := 0 - var fracsFound []string - var fracsSkipped []string + var ( + stats searchStats + qprMemUsage int + ) for len(remainingFracs) > 0 && (scanAll || params.Limit > 0) { chunk := remainingFracs.Shift(fracsChunkSize) - subQPRs, searchTimeNanos, waitTimeNanos, err := s.searchDocsAsync(ctx, chunk, params) + subQPRs, timings, err := s.searchDocsAsync(ctx, chunk, params) if err != nil { return nil, err } - - for i, qpr := range subQPRs { - if !qpr.Empty() { - totalFracsFound++ - if len(fracsFound) < maxFracsSlowSearchLog { - fracsFound = append(fracsFound, chunk[i].Info().Name()) - } - } else { - totalFracsSkipped++ - if len(fracsSkipped) < maxFracsSlowSearchLog { - fracsSkipped = append(fracsSkipped, chunk[i].Info().Name()) - } - } - } - - totalSearchTimeNanos += searchTimeNanos - totalWaitTimeNanos += waitTimeNanos + stats.merge(chunk, subQPRs, timings) seq.MergeQPRs(total, subQPRs, origLimit, seq.MillisToMID(params.HistInterval), params.Order) - totalMemUsage = total.MemUsage() - - if s.cfg.MaxQprMemory > 0 && totalMemUsage > s.cfg.MaxQprMemory { - return nil, fmt.Errorf("%w: used %d bytes, limit %d", consts.ErrMemoryLimitExceeded, total.MemUsage(), s.cfg.MaxQprMemory) + qprMemUsage = total.MemUsage() + if s.cfg.MaxQprMemory > 0 && qprMemUsage > s.cfg.MaxQprMemory { + return nil, fmt.Errorf( + "%w: used %d bytes, limit %d", + consts.ErrMemoryLimitExceeded, qprMemUsage, s.cfg.MaxQprMemory, + ) } // reduce the limit on the number of ensured docs in response params.Limit = origLimit - calcEnsuredIDsCount(total.IDs, remainingFracs, params.Order) - - subSearchesCnt++ - } - - if tr.Enabled() { - searchSpan := &querytracer.Span{ - Message: "search iteratively (cpu time)", - Duration: time.Duration(totalSearchTimeNanos), - } - tr.AddChildWithSpan(searchSpan) - waitSpan := &querytracer.Span{ - Message: "waiting goroutines (all cores)", - Duration: time.Duration(totalWaitTimeNanos), - } - tr.AddChildWithSpan(waitSpan) } - searchSubSearches.Observe(float64(subSearchesCnt)) + stats.addToTracer(tr) + searchSubSearches.Observe(float64(stats.iterations)) took := time.Since(start) if s.cfg.SlowLogThreshold != 0 && took >= s.cfg.SlowLogThreshold { fields := []zap.Field{ - zap.Int64("took_ms", took.Milliseconds()), zap.Object("params", params), - zap.Int("total_fracs_found", totalFracsFound), - zap.Strings("fracs_found", fracsFound), - zap.Int("total_fracs_skipped", totalFracsSkipped), - zap.Strings("fracs_skipped", fracsSkipped), - util.ZapUint64AsSizeStr("qpr_size", uint64(totalMemUsage)), zap.Uint64("total", total.Total), + zap.Int64("took_ms", took.Milliseconds()), + util.ZapUint64AsSizeStr("qpr_size", uint64(qprMemUsage)), } - logger.Warn("slow search", fields...) + logger.Warn( + "slow search", + append(fields, stats.zapFields()...)..., + ) } - return total, nil + return total, nil } func (s *Searcher) prepareFracs(fracs List, params processor.SearchParams) (List, error) { @@ -183,64 +146,56 @@ func calcEnsuredIDsCount(ids seq.IDSources, remainingFracs List, order seq.DocsO return sort.Search(len(ids), func(i int) bool { return ids[i].ID.MID <= nextFracInfo.To }) } -func (s *Searcher) searchDocsAsync(ctx context.Context, fracs []frac.Fraction, params processor.SearchParams) ([]*seq.QPR, int64, int64, error) { +func (s *Searcher) searchDocsAsync( + ctx context.Context, + fracs []frac.Fraction, + params processor.SearchParams, +) ([]*seq.QPR, fracTimings, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() - var err error + var ( + err error + once sync.Once + wg sync.WaitGroup + ) - once := sync.Once{} - wg := sync.WaitGroup{} qprs := make([]*seq.QPR, len(fracs)) - - wgDoneNanos := make([]int64, len(fracs)) - searchElapsedNanos := make([]int64, len(fracs)) + timings := fracTimings{elapsed: make([]time.Duration, len(fracs))} loop: - for i, frac := range fracs { + for i, f := range fracs { + acquireStart := time.Now() + select { case <-ctx.Done(): once.Do(func() { err = ctx.Err() }) break loop case s.sem <- struct{}{}: // acquire semaphore - wg.Add(1) - go func() { - searchStart := time.Now() - var fracErr error - if qprs[i], fracErr = s.fracSearch(ctx, params, frac); fracErr != nil { - once.Do(func() { - err = fracErr - cancel() - }) - } - searchElapsedNanos[i] = time.Since(searchStart).Nanoseconds() - - <-s.sem // release semaphore - - wgDoneNanos[i] = time.Now().UnixNano() - wg.Done() - }() } - } - wg.Wait() - waitEndTime := time.Now().UnixNano() + timings.semaphoreWait += time.Since(acquireStart) - totalSearchTimeNanos := int64(0) - totalWaitTimeNanos := int64(0) + wg.Go(func() { + defer func() { <-s.sem }() // release semaphore - for i := range fracs { - if wgDoneNanos[i] != 0 { - totalWaitTimeNanos += waitEndTime - wgDoneNanos[i] - } - totalSearchTimeNanos += searchElapsedNanos[i] - } + searchStart := time.Now() + qpr, fracErr := s.fracSearch(ctx, params, f) + timings.elapsed[i] = time.Since(searchStart) - if err != nil { - return nil, 0, 0, err + if fracErr != nil { + once.Do(func() { + err = fracErr + cancel() + }) + return + } + qprs[i] = qpr + }) } - return qprs, totalSearchTimeNanos, totalWaitTimeNanos, nil + wg.Wait() + return qprs, timings, err } func (s *Searcher) fracSearch(ctx context.Context, params processor.SearchParams, f frac.Fraction) (_ *seq.QPR, err error) {