+36 -3

api.go

Reviewed

··· 324 324 return 325 325 } 326 326 327 327 + type FlushReason uint8 328 328 + 329 329 + const ( 330 330 + FlushReasonTimerExpired FlushReason = 1 << iota 331 331 + FlushReasonFinalFlush 332 332 + FlushReasonMaxSize 333 333 + ) 334 334 + 335 335 + var FlushReasonStrings = map[FlushReason]string{ 336 336 + FlushReasonTimerExpired: "timer_expired", 337 337 + FlushReasonFinalFlush: "final_flush", 338 338 + FlushReasonMaxSize: "max_size_reached", 339 339 + } 340 340 + 341 341 + func (fr FlushReason) String() string { 342 342 + if v, ok := FlushReasonStrings[fr]; ok { 343 343 + return v 344 344 + } 345 345 + 346 346 + return "none" 347 347 + } 348 348 + 327 349 // Stats contains interesting numbers on the search 328 350 type Stats struct { 329 351 // Amount of I/O for reading contents. ··· 376 398 377 399 // Number of times regexp was called on files that we evaluated. 378 400 RegexpsConsidered int 401 401 + 402 402 + // FlushReason explains why results were flushed. 403 403 + FlushReason FlushReason 379 404 } 380 405 381 381 - func (s *Stats) sizeBytes() uint64 { 382 382 - // This assumes we are running on a 64-bit architecture. 383 383 - return 16 * 8 406 406 + func (s *Stats) sizeBytes() (sz uint64) { 407 407 + sz = 16 * 8 // This assumes we are running on a 64-bit architecture 408 408 + sz += 1 // FlushReason 409 409 + 410 410 + return 384 411 } 385 412 386 413 func (s *Stats) Add(o Stats) { ··· 399 426 s.ShardsSkippedFilter += o.ShardsSkippedFilter 400 427 s.Wait += o.Wait 401 428 s.RegexpsConsidered += o.RegexpsConsidered 429 429 + 430 430 + // We want the first non-zero FlushReason to be sticky. This is a useful 431 431 + // property when aggregating stats from several Zoekts. 432 432 + if s.FlushReason == 0 { 433 433 + s.FlushReason = o.FlushReason 434 434 + } 402 435 } 403 436 404 437 // Zero returns true if stats is empty.

+2 -2

api_test.go

Reviewed

··· 78 78 79 79 func TestSizeBytesSearchResult(t *testing.T) { 80 80 var sr = SearchResult{ 81 81 - Stats: Stats{}, // 128 bytes 81 81 + Stats: Stats{}, // 129 bytes 82 82 Progress: Progress{}, // 16 bytes 83 83 Files: []FileMatch{{ // 24 bytes + 460 bytes 84 84 Score: 0, // 8 bytes ··· 109 109 LineFragments: nil, // 48 bytes 110 110 } 111 111 112 112 - var wantBytes uint64 = 724 112 112 + var wantBytes uint64 = 725 113 113 if sr.SizeBytes() != wantBytes { 114 114 t.Fatalf("want %d, got %d", wantBytes, sr.SizeBytes()) 115 115 }

+5 -4

shards/aggregate.go

Reviewed

··· 108 108 109 109 // stopCollectingAndFlush will send what we have collected and all future 110 110 // sends will go via sender directly. 111 111 - stopCollectingAndFlush := func(reason string) { 111 111 + stopCollectingAndFlush := func(reason zoekt.FlushReason) { 112 112 mu.Lock() 113 113 defer mu.Unlock() 114 114 ··· 117 117 } 118 118 119 119 if agg, ok := collectSender.Done(); ok { 120 120 - metricFinalAggregateSize.WithLabelValues(reason).Observe(float64(len(agg.Files))) 120 120 + metricFinalAggregateSize.WithLabelValues(reason.String()).Observe(float64(len(agg.Files))) 121 121 + agg.FlushReason = reason 121 122 sender.Send(agg) 122 123 } 123 124 ··· 135 136 case <-timerCancel: 136 137 timer.Stop() 137 138 case <-timer.C: 138 138 - stopCollectingAndFlush("timer_expired") 139 139 + stopCollectingAndFlush(zoekt.FlushReasonTimerExpired) 139 140 } 140 141 }() 141 142 142 143 finalFlush := func() { 143 143 - stopCollectingAndFlush("final_flush") 144 144 + stopCollectingAndFlush(zoekt.FlushReasonFinalFlush) 144 145 } 145 146 146 147 return stream.SenderFunc(func(event *zoekt.SearchResult) {