Approach: keep incremental, outlier-resistant summaries per small time bucket (e.g., 1-minute). Use t-digest per bucket (good for streaming percentiles, mergeable, robust to extreme outliers). Maintain a fixed-size circular buffer for last 7 days of buckets, merge t-digests on demand (or maintain an incremental merged digest updated as oldest buckets expire) to compute rolling p99 quickly. Persist raw digests (not raw samples) to durable store for recovery.Code (sketch using tdigest package):python
from tdigest import TDigest
from collections import deque, defaultdict
import time
import threading
BUCKET_SECONDS = 60
WINDOW_SECONDS = 7 * 24 * 3600
NUM_BUCKETS = WINDOW_SECONDS // BUCKET_SECONDS
class RollingSLI:
def __init__(self, bucket_seconds=BUCKET_SECONDS):
self.bucket_seconds = bucket_seconds
self.num_buckets = NUM_BUCKETS
self.buckets = deque(maxlen=self.num_buckets) # each item: (bucket_ts, TDigest)
self.lock = threading.Lock()
self.current_bucket_ts = None
def _bucket_ts(self, ts):
return int(ts // self.bucket_seconds) * self.bucket_seconds
def ingest(self, latency_ms, ts=None):
ts = ts or time.time()
bts = self._bucket_ts(ts)
with self.lock:
if not self.buckets or self.buckets[-1][0] != bts:
# append new bucket
self.buckets.append((bts, TDigest()))
# ensure last bucket matches
if self.buckets[-1][0] != bts:
# gap handling: create empty buckets until bts
while not self.buckets or self.buckets[-1][0] < bts:
next_ts = (self.buckets[-1][0] + self.bucket_seconds) if self.buckets else bts
self.buckets.append((next_ts, TDigest()))
self.buckets[-1][1].update(latency_ms)
def compute_p99(self):
with self.lock:
merged = TDigest()
for _, td in self.buckets:
merged = TDigest.merge(merged, td) # uses tdigest's efficient merge
return merged.percentile(99)
def availability_sli(self, threshold_ms):
p99 = self.compute_p99()
return p99 < threshold_ms, p99
