Handling Large Scale Data and Time Series Data Questions

# Python-like streaming implementation
import heapq
import mmh3   # murmurhash3 library, assume installed
from collections import defaultdict

class CountMinSketch:
    def __init__(self, width, depth, seed=0):
        self.width = width
        self.depth = depth
        self.tables = [[0]*width for _ in range(depth)]
        self.seeds = [seed + i*0x9e3779b1 for i in range(depth)]
    def _hashes(self, key):
        for s in self.seeds:
            yield mmh3.hash(key, s) % self.width
    def add(self, key, delta=1):
        for i, idx in enumerate(self._hashes(key)):
            self.tables[i][idx] += delta
    def estimate(self, key):
        return min(self.tables[i][idx] for i, idx in enumerate(self._hashes(key)))

class TopKTracker:
    def __init__(self, k, cms):
        self.k = k
        self.cms = cms
        self.entry_map = {}  # key -> (est_count)
        self.heap = []  # min-heap of (est_count, key)

    def process_sample(self, series_key):
        self.cms.add(series_key, 1)
        est = self.cms.estimate(series_key)
        if series_key in self.entry_map:
            # update: push new tuple; lazy removal when popped
            self.entry_map[series_key] = est
            heapq.heappush(self.heap, (est, series_key))
            return
        if len(self.heap) < self.k:
            self.entry_map[series_key] = est
            heapq.heappush(self.heap, (est, series_key))
        else:
            min_est, min_key = self.heap[0]
            if est > min_est:
                # evict until heap top is valid
                while self.heap and self.entry_map.get(self.heap[0][1], None) != self.heap[0][0]:
                    heapq.heappop(self.heap)
                if self.heap and est > self.heap[0][0]:
                    # remove evicted
                    _, evicted = heapq.heappop(self.heap)
                    self.entry_map.pop(evicted, None)
                    self.entry_map[series_key] = est
                    heapq.heappush(self.heap, (est, series_key))

    def topk(self):
        # return stable snapshot
        items = [(c,k) for c,k in self.heap if self.entry_map.get(k)==c]
        items.sort(reverse=True)
        return items

# Example usage: streaming loop consuming prometheus remote_write-like samples
cms = CountMinSketch(width=2000, depth=5)
tracker = TopKTracker(k=100, cms=cms)
# for sample in stream:
#     series_key = serialize_labels(sample.labels)
#     tracker.process_sample(series_key)
# At any time: tracker.topk()

relabel_configs:
- source_labels: [__meta_kubernetes_pod_label_request_id]
  action: drop
- regex: '.*-instance-\d+'
  source_labels: [instance]
  replacement: 'instance'
  action: replace

alert: HighServiceErrorRate
expr: job:error_ratio:5m{env="prod"} > 0.05
for: 10m
labels: { severity: critical }

WITH per_min AS (
  SELECT
    host,
    time_bucket_gapfill('1 minute', ts) AS minute,
    AVG(value) FILTER (WHERE metric_name = 'cpu') AS avg_cpu
  FROM metrics
  WHERE ts >= now() - interval '24 hours'
  GROUP BY host, minute
)
SELECT host, max(avg_cpu) AS max_1min_avg_cpu
FROM per_min
WHERE avg_cpu IS NOT NULL
GROUP BY host
ORDER BY max_1min_avg_cpu DESC
LIMIT 10;