Repository navigation
Conversation
When reserveChunkForLine fails to reserve, if it is the first time that happens in a flush window, emit the name of the metric that was dropped as a hint for identifying problematic metrics. While it's not guaranteed that the logged metric name is itself the problematic one, if the same name is observing multiple times, that would be more suggestive.
| defer buffer.mu.Unlock() | ||
|
|
||
| // If there is no data to flush from the shard, then skip socket writes | ||
| if buffer.chunkIndex == 0 && len(buffer.data[0]) == 0 { |
There was a problem hiding this comment.
This early return skips reporting and resetting overflows/overflowBytes/droppedSampleId. If the only thing that landed in a shard this window was an oversized line (the lineLen > chunkSize drop path), nothing is logged and no overflow metric is emitted, and the state carries over. In a later window with real drops, the log reports the merged count as "this window" and names the stale meter id from the earlier window. Since recordOverflow only samples when overflows == 0, the stale id is never replaced. Oversized lines are probably the main pathological case this is meant to diagnose. The overflow reporting and reset should happen before the early return.
| if id, ok := meterIdKey(prefix); ok { | ||
| b.droppedSampleId = id | ||
| } else { | ||
| b.droppedSampleId = prefix |
There was a problem hiding this comment.
If meterIdKey fails, the raw prefix is stored and later logged verbatim with no length limit and no escaping. A malformed line containing \n can split or forge log records, and a malformed line over 60KB (the oversized path) would be logged in full every window. Suggest truncating to a bounded length and formatting with %q.
| func (b *bufferShard) recordOverflow(prefix string, totalWriteLength int) { | ||
| if b.overflows == 0 { | ||
| if id, ok := meterIdKey(prefix); ok { | ||
| b.droppedSampleId = id |
There was a problem hiding this comment.
This is a substring of the caller's string, so it keeps the whole original allocation alive until the shard next reports (possibly indefinitely given the early-return issue in flushBufferShard). For a multi-MB oversized line that could be significant. strings.Clone of a truncated id would avoid it; it only happens once per window so the cost is negligible.
|
|
||
| // Log a single overflow for each window. | ||
| if totalOverflows > 0 { | ||
| llb.logger.Infof("LowLatencyBuffer overflow in bufferSet=%s: dropped %d lines (%d bytes) this window; sample dropped meter id: %s", bufferSet, totalOverflows, totalOverflowBytes, droppeSampleId) |
There was a problem hiding this comment.
This logs at Info on every flush window with no rate limit. Overflow is usually sustained (buffer too small for the app's volume), so with the default 5s interval that is ~17k lines/day per process. The overflows metric already signals the condition. Consider limiting to at most once per minute or similar.
| } | ||
| if !b.reserveChunkForLine(len(prefix) + len(value)) { | ||
| // overflows (drops) are counted in reserveChunkForLine, for metric reporting | ||
| b.recordOverflow(prefix, len(prefix)+len(value)) |
There was a problem hiding this comment.
Behavior change for the overflowBytes metric: previously the drop path in reserveChunkForLine that advanced chunkIndex counted totalWriteLength (including the +1 separator); now it counts len(prefix)+len(value). Small, but it silently shifts the metric, and the new test encodes the new semantics. Intentional?
| llb.writer.WriteString(fmt.Sprintf("d:spectator-go.lowLatencyBuffer.overflowBytes,bufferSet=%s:%d", bufferSet, buffer.overflowBytes)) | ||
| overflows, overflowBytes, sampleId := buffer.overflows, buffer.overflowBytes, buffer.droppedSampleId | ||
| if overflows > 0 { | ||
| llb.writer.WriteString(fmt.Sprintf("c:spectator-go.lowLatencyBuffer.overflows,bufferSet=%s:%d", bufferSet, overflows)) |
There was a problem hiding this comment.
Minor: swapAndFlush now has window-wide totals, but the overflow metrics are still written once per shard, up to 128 separate writes per window on a 64-CPU host during sustained overflow. Emitting the two aggregate lines once from swapAndFlush would reduce that to 2.
| // "<type>[,<extra>]:<id>:<value>" line; the id is the substring between the | ||
| // first and last ':'. Returns ok=false for malformed lines (fewer than two ':' | ||
| // characters). | ||
| func meterIdKey(line string) (string, bool) { |
There was a problem hiding this comment.
Nit: this repeats the first/last-colon parsing in orderSensitiveMeterIdKey and meter.idFromLinePrefix, with slightly different guards (first < 0 vs first <= 0). orderSensitiveMeterIdKey could check the type symbol and then delegate to this.
| var bytesWritten int | ||
| var bytesWritten, totalOverflows int | ||
| var totalOverflowBytes int64 | ||
| var droppeSampleId string |
There was a problem hiding this comment.
Nit: typo droppeSampleId, should be droppedSampleId.
| } | ||
|
|
||
| type flushBufferResult struct { | ||
| bytesWritten, |
There was a problem hiding this comment.
Nit: bytesWritten, on its own line reads as if it has a separate type; bytesWritten, overflows int or separate declarations would be clearer.
| } | ||
| } | ||
|
|
||
| func TestLowLatencyBuffer_OverflowLogsOncePerFlushWindow(t *testing.T) { |
There was a problem hiding this comment.
This only covers a single window. Worth adding a second swapAndFlush cycle with a different dropped id to verify the sample and counters reset, plus a case where the only line in a shard is an oversized one. Those would catch the early-return issue in flushBufferShard.
2 participants
When reserveChunkForLine fails to reserve, if it is the first time that happens in a flush window, emit the name of the metric that was dropped as a hint for identifying problematic metrics. While it's not guaranteed that the logged metric name is itself the problematic one, if the same name is observing multiple times, that would be more suggestive.