///|
pub(all) struct LatencyBucket {
upper_bound_ms : Int
count : Int
} derive(Eq, Debug)
///|
pub(all) struct LatencyHistogram {
buckets : Array[LatencyBucket]
overflow_count : Int
total_count : Int
total_ms : Int
min_ms : Int
max_ms : Int
} derive(Eq, Debug)
///|
pub(all) struct LatencySnapshot {
count : Int
average_ms : Int
min_ms : Int
max_ms : Int
p50_ms : Int
p90_ms : Int
p99_ms : Int
buckets : Array[LatencyBucket]
overflow_count : Int
} derive(Eq, Debug)
///|
pub fn new_latency_histogram(upper_bounds_ms : Array[Int]) -> LatencyHistogram {
let normalized = normalize_bounds(upper_bounds_ms)
let buckets : Array[LatencyBucket] = []
for bound in normalized {
buckets.push({ upper_bound_ms: bound, count: 0 })
}
{
buckets,
overflow_count: 0,
total_count: 0,
total_ms: 0,
min_ms: 0,
max_ms: 0,
}
}
///|
pub fn default_latency_histogram() -> LatencyHistogram {
new_latency_histogram([5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000])
}
///|
pub fn latency_observe(
histogram : LatencyHistogram,
latency_ms : Int,
) -> LatencyHistogram {
let value = clamp_non_negative(latency_ms)
let buckets : Array[LatencyBucket] = []
let mut placed = false
for bucket in histogram.buckets {
if !placed && value <= bucket.upper_bound_ms {
buckets.push({ ..bucket, count: bucket.count + 1 })
placed = true
} else {
buckets.push(bucket)
}
}
let overflow_increment = if placed { 0 } else { 1 }
{
buckets,
overflow_count: histogram.overflow_count + overflow_increment,
total_count: histogram.total_count + 1,
total_ms: histogram.total_ms + value,
min_ms: if histogram.total_count == 0 {
value
} else {
min_int(histogram.min_ms, value)
},
max_ms: max_int(histogram.max_ms, value),
}
}
///|
pub fn latency_merge(
left : LatencyHistogram,
right : LatencyHistogram,
) -> Result[LatencyHistogram, String] {
if !same_bounds(left.buckets, right.buckets) {
return Err("latency histograms use different bucket boundaries")
}
let buckets : Array[LatencyBucket] = []
for index = 0; index < left.buckets.length(); index = index + 1 {
buckets.push({
upper_bound_ms: left.buckets[index].upper_bound_ms,
count: left.buckets[index].count + right.buckets[index].count,
})
}
let total_count = left.total_count + right.total_count
Ok({
buckets,
overflow_count: left.overflow_count + right.overflow_count,
total_count,
total_ms: left.total_ms + right.total_ms,
min_ms: if left.total_count == 0 {
right.min_ms
} else if right.total_count == 0 {
left.min_ms
} else {
min_int(left.min_ms, right.min_ms)
},
max_ms: max_int(left.max_ms, right.max_ms),
})
}
///|
pub fn latency_percentile(
histogram : LatencyHistogram,
percentile : Int,
) -> Int {
if histogram.total_count == 0 {
return 0
}
let safe = min_int(100, max_int(1, percentile))
let target = (histogram.total_count * safe + 99) / 100
let mut cumulative = 0
for bucket in histogram.buckets {
cumulative = cumulative + bucket.count
if cumulative >= target {
return bucket.upper_bound_ms
}
}
histogram.max_ms
}
///|
pub fn latency_snapshot(histogram : LatencyHistogram) -> LatencySnapshot {
{
count: histogram.total_count,
average_ms: if histogram.total_count == 0 {
0
} else {
histogram.total_ms / histogram.total_count
},
min_ms: histogram.min_ms,
max_ms: histogram.max_ms,
p50_ms: latency_percentile(histogram, 50),
p90_ms: latency_percentile(histogram, 90),
p99_ms: latency_percentile(histogram, 99),
buckets: histogram.buckets.copy(),
overflow_count: histogram.overflow_count,
}
}
///|
pub fn format_latency_snapshot(snapshot : LatencySnapshot) -> String {
"count=" +
snapshot.count.to_string() +
" average_ms=" +
snapshot.average_ms.to_string() +
" min_ms=" +
snapshot.min_ms.to_string() +
" max_ms=" +
snapshot.max_ms.to_string() +
" p50_ms=" +
snapshot.p50_ms.to_string() +
" p90_ms=" +
snapshot.p90_ms.to_string() +
" p99_ms=" +
snapshot.p99_ms.to_string() +
" overflow=" +
snapshot.overflow_count.to_string()
}
///|
fn normalize_bounds(bounds : Array[Int]) -> Array[Int] {
let result : Array[Int] = []
for bound in bounds {
let safe = clamp_non_negative(bound)
if !contains_int(result, safe) {
insert_sorted(result, safe)
}
}
result
}
///|
fn insert_sorted(values : Array[Int], value : Int) -> Unit {
let mut index = 0
while index < values.length() && values[index] < value {
index = index + 1
}
values.insert(index, value)
}
///|
fn contains_int(values : Array[Int], target : Int) -> Bool {
for value in values {
if value == target {
return true
}
}
false
}
///|
fn same_bounds(
left : Array[LatencyBucket],
right : Array[LatencyBucket],
) -> Bool {
if left.length() != right.length() {
return false
}
for index = 0; index < left.length(); index = index + 1 {
if left[index].upper_bound_ms != right[index].upper_bound_ms {
return false
}
}
true
}