ref(metrics): Rework metrics aggregator to keep internal partitions

[Dav1dde]

Replaces the current metrics aggregator, which works based off fixed priorities in a priority queue with regular intervals, with a ring buffer based aggregator.

Overview:

• Aggregator is now ring buffer based instead of a priority queue, this minimizes work needed for merging and flushing buckets.
• Aggregator now no longer guarantees a minimum delay for backdated buckets but on average still delays enough (for real time buckets there still is an accurate delay).
• Aggregator is now driven by flushes and this is how it tracks and advances time. This means all operations (merges and flushes) can be done entirely without accessing system time.
• Cost tracking is much more efficient now and tracked for the total and per slot, on flush the slot values are subtracted from the total, which does not require additional iterations and calculations of costs.
• Per projects cost limits are only tracked per slot, instead of overall, reducing the necessary book keeping by a lot.
• On shutdown the aggregator is replaced with an aggregator with much more aggressive flush behaviour. This massively simplifies the code, still has a good time to flush and overall much better flush behaviour by keeping partitions consistent.
• Metric name/tag validation is now a concern of the service instead of the aggregator.
• Uses ahash with a fixed seed instead of fnv (it's faster)
• Lots of unused metrics have been reworked or modified (from a histogram which only was used for sum+count to two counters)

For implementation details see the exhaustive code documentation, especially in the inner aggregator.

Fixes: https://github.com/getsentry/team-ingest/issues/606

[jjbayer]

Looks very nice!

[jjbayer]

Naively I would expect this to be (slot - self.head) % slots_len.

Is the + slots_len here to make the wrapping sub work in case slot < self.head? This operation is complex enough to warrant some documentation, and / or a helper function.

[Dav1dde]

What we need here is rem_euclid in case slot - self.head is negative, but for that we need to go from u64 to i64 which truncates the size of the u64 down to u64 / 2. This is a way to shift the operation entirely into the positive space.

Will add some docs, and/or a function, this is double confusing since the wrapping_sub isn't necessary but still here (it can't wrap).

[jjbayer]

What happens with different time stamps that map to the same index? Is that case prevented by how slots.len() is chosen in the beginning?

[Dav1dde]

This is fine and will happen for backdated und future buckets, there are tests that cover this case. Since the timestamp is part of the key, they won't be aggregated together and are independent.

[jjbayer]

I wonder if this would be simpler with an actual statically sized ring buffer. Then you could reset the slot at head and move the head, without pop / push.

[loewenheim]

I also get the impression that this implementation has features of a queue and a static ring buffer simultaneously.

[Dav1dde]

You can rotate the VecDeque, initially this is what I had, but it turns out it's a bit nicer this way. We need ownership of parts of the slot (e.g. all buckets), with a rotation we then have to std::mem::replace the parts we need ownership of. We also don't get around a fallible access/unwrap like here, since we still need a bounds check.

Removing the item (which just shifts indices internally and returns the value) and adding a new one back turned out to be nicer.

[loewenheim]

I also get the impression that this implementation has features of a queue and a static ring buffer simultaneously.

[jjbayer]

The PR is pretty big so I probably missed some details, but overall the design makes sense & I'm looking forward to see how this performs in production.

The rollout is probably gonna be merge -> test -> revert?

[jjbayer]

What does this do? I'm reading the docs of block_in_place but I don't understand its purpose.

[Dav1dde]

Might not need this anymore, but the previous aggregators' max on the flush was up to a 200ms. Since we run in a tokio task here, we're blocking all queued other tasks on this tokio worker for the duration of the call. With block_in_place we can tell the runtime to clear it's local queue before running the closure. This in theory should bring down p99+ latencies.

[Dav1dde]

The rollout is probably gonna be merge -> test -> revert?

Merge, test, it works, rollout prod ideally. Let's talk tomorrow sync about it?