Solution (#46): Collectors: rate limiting + drop counters under load (Phase 9.2.4

path/to/internal/bpf/c/disk_io.c

@@ -0,0 +1,16 @@
+# complete code
+#include <linux/bpf.h>
+#include "kerno.h"
+
+struct kerno_backpressure {
+    __u32 should_emit;
+};
+
+SEC("classifier")
+int kerno_backpressure_cls(struct xdp_md *ctx) {
+    struct kerno_backpressure *backpressure = (struct kerno_backpressure *)ctx->data;
+    if (backpressure->should_emit == 0) {
+        return XDP_DROP;
+    }
+    return XDP_PASS;
+}

path/to/internal/bpf/c/disk_io_test.c

@@ -0,0 +1,16 @@
+# complete code
+#include <linux/bpf.h>
+#include "kerno_test.h"
+
+struct kerno_backpressure {
+    __u32 should_emit;
+};
+
+SEC("classifier")
+int kerno_backpressure_cls(struct xdp_md *ctx) {
+    struct kerno_backpressure *backpressure = (struct kerno_backpressure *)ctx->data;
+    if (backpressure->should_emit == 0) {
+        return XDP_DROP;
+    }
+    return XDP_PASS;
+}

path/to/internal/bpf/c/headers/kerno.h

@@ -0,0 +1,7 @@
+# complete code
+#ifndef KERNO_H
+#define KERNO_H
+
+#define KERNO_BACKPRESSURE 1
+
+#endif

path/to/internal/bpf/c/headers/kerno_test.h

@@ -0,0 +1,5 @@
+# complete code
+#ifndef KERNO_TEST_H
+#define KERNO_TEST_H
+
+#endif

path/to/internal/cli/spinner.go

@@ -0,0 +1,14 @@
+# complete code
+import time
+import random
+
+class Spinner:
+    def __init__(self):
+        self.running = True
+
+    def start(self):
+        while self.running:
+            time.sleep(1)
+
+    def stop(self):
+        self.running = False

path/to/internal/cli/spinner_test.go

@@ -0,0 +1,12 @@
+# complete code
+import unittest
+import time
+import random
+
+class SpinnerTest(unittest.TestCase):
+    def test_spinner(self):
+        spinner = Spinner()
+        spinner.start()
+        time.sleep(1)
+        spinner.stop()
+        self.assertFalse(spinner.running)

path/to/internal/cli/start.go

@@ -0,0 +1,15 @@
+# complete code
+import time
+import random
+
+class Start:
+    def __init__(self, rate_limiter, sampler):
+        self.rate_limiter = rate_limiter
+        self.sampler = sampler
+
+    def start(self):
+        while True:
+            if self.rate_limiter.is_allowed():
+                self.sampler.sample()
+            else:
+                time.sleep(1)

path/to/internal/cli/start_test.go

@@ -0,0 +1,12 @@
+# complete code
+import unittest
+import time
+import random
+
+class StartTest(unittest.TestCase):
+    def test_start(self):
+        rate_limiter = RateLimiter(500000, 1)
+        sampler = Sampler(1.0)
+        start = Start(rate_limiter, sampler)
+        start.start()
+        self.assertTrue(True)

path/to/internal/collector/aggregator/histogram.go

@@ -0,0 +1,17 @@
+# complete code
+import time
+import random
+
+class Histogram:
+    def __init__(self, buckets):
+        self.buckets = buckets
+        self.values = [0] * buckets
+
+    def update(self, value):
+        bucket = int(value / (self.buckets - 1) * (self.buckets - 1))
+        self.values[bucket] += 1
+
+    def get_percentile(self, percentile):
+        values = sorted(self.values)
+        index = int((percentile / 100) * (len(values) - 1))
+        return (index + 1) / (len(values) - 1) * (self.buckets - 1)

path/to/internal/collector/aggregator/histogram_test.go

@@ -0,0 +1,10 @@
+# complete code
+import unittest
+import time
+import random
+
+class HistogramTest(unittest.TestCase):
+    def test_histogram(self):
+        histogram = Histogram(10)
+        histogram.update(5)
+        self.assertEqual(histogram.get_percentile(50), 5)

path/to/internal/collector/aggregator/lru.go

@@ -0,0 +1,24 @@
+# complete code
+import time
+import random
+
+class LRU:
+    def __init__(self, capacity):
+        self.capacity = capacity
+        self.cache = {}
+
+    def get(self, key):
+        if key in self.cache:
+            value = self.cache[key]
+            del self.cache[key]
+            self.cache[key] = value
+            return value
+        return None
+
+    def set(self, key, value):
+        if key in self.cache:
+            del self.cache[key]
+        elif len(self.cache) >= self.capacity:
+            random_key = random.choice(list(self.cache.keys()))
+            del self.cache[random_key]
+        self.cache[key] = value

path/to/internal/collector/aggregator/lru_test.go

@@ -0,0 +1,10 @@
+# complete code
+import unittest
+import time
+import random
+
+class LRUTest(unittest.TestCase):
+    def test_lru(self):
+        lru = LRU(10)
+        lru.set("key", "value")
+        self.assertEqual(lru.get("key"), "value")

path/to/internal/collector/aggregator/ratelimit.go

@@ -0,0 +1,36 @@
+# complete code
+import time
+import random
+
+class RateLimiter:
+    def __init__(self, budget, interval):
+        self.budget = budget
+        self.interval = interval
+        self.tokens = budget
+        self.last_update = time.time()
+
+    def update(self):
+        now = time.time()
+        elapsed = now - self.last_update
+        self.tokens = min(self.tokens + (elapsed / self.interval) * self.budget, self.budget)
+        self.last_update = now
+
+    def is_allowed(self):
+        self.update()
+        return self.tokens > 0
+
+class Sampler:
+    def __init__(self, target_overhead_pct):
+        self.target_overhead_pct = target_overhead_pct
+        self.sample_rate = 1.0 / target_overhead_pct
+
+    def sample(self):
+        return random.random() < self.sample_rate
+
+class RateLimiterAndSampler:
+    def __init__(self, rate_limiter, sampler):
+        self.rate_limiter = rate_limiter
+        self.sampler = sampler
+
+    def is_allowed(self):
+        return self.rate_limiter.is_allowed() or self.sampler.sample()

path/to/internal/collector/aggregator/ratelimit_test.go

@@ -0,0 +1,11 @@
+# complete code
+import unittest
+import time
+import random
+
+class RateLimiterTest(unittest.TestCase):
+    def test_rate_limiter(self):
+        rate_limiter = RateLimiter(500000, 1)
+        self.assertTrue(rate_limiter.is_allowed())
+        time.sleep(1)
+        self.assertFalse(rate_limiter.is_allowed())

path/to/internal/collector/cgroup_memory.go

@@ -0,0 +1,13 @@
+# complete code
+import time
+import random
+
+class CGroupMemory:
+    def __init__(self):
+        self.drops = 0
+
+    def increment_drop(self):
+        self.drops += 1
+
+    def get_drop_count(self):
+        return self.drops

path/to/internal/collector/cgroup_memory_test.go

@@ -0,0 +1,10 @@
+# complete code
+import unittest
+import time
+import random
+
+class CGroupMemoryTest(unittest.TestCase):
+    def test_cgroup_memory(self):
+        cgroup_memory = CGroupMemory()
+        cgroup_memory.increment_drop()
+        self.assertEqual(cgroup_memory.get_drop_count(), 1)

path/to/internal/collector/collector.go

@@ -0,0 +1,14 @@
+# complete code
+import time
+import random
+
+class Collector:
+    def __init__(self, rate_limiter, sampler):
+        self.rate_limiter = rate_limiter
+        self.sampler = sampler
+
+    def record(self, event):
+        if self.rate_limiter.is_allowed():
+            self.sampler.sample()
+        else:
+            time.sleep(1)

path/to/internal/config/config.go

@@ -0,0 +1,20 @@
+# complete code
+import time
+import random
+
+class Config:
+    def __init__(self):
+        self.rate_limits = {
+            "syscall_latency": 500000,
+            "sched_delay": 200000
+        }
+        self.sampling = {
+            "enabled": True,
+            "target_overhead_pct": 1.0
+        }
+
+    def get_rate_limit(self, key):
+        return self.rate_limits.get(key, 0)
+
+    def get_sampling_config(self):
+        return self.sampling

path/to/internal/metrics/bridge.go

@@ -0,0 +1,10 @@
+# complete code
+import time
+import random
+
+class Bridge:
+    def __init__(self, metrics):
+        self.metrics = metrics
+
+    def get_metrics(self):
+        return self.metrics.get_drop_count(), self.metrics.get_sampled_count()

path/to/internal/metrics/bridge_test.go

@@ -0,0 +1,10 @@
+# complete code
+import unittest
+import time
+import random
+
+class BridgeTest(unittest.TestCase):
+    def test_bridge(self):
+        metrics = Metrics()
+        bridge = Bridge(metrics)
+        self.assertEqual(bridge.get_metrics(), (0, 0))

path/to/internal/metrics/metrics.go

@@ -0,0 +1,20 @@
+# complete code
+import time
+import random
+
+class Metrics:
+    def __init__(self):
+        self.drop_count = 0
+        self.sampled_count = 0
+
+    def increment_drop_count(self):
+        self.drop_count += 1
+
+    def increment_sampled_count(self):
+        self.sampled_count += 1
+
+    def get_drop_count(self):
+        return self.drop_count
+
+    def get_sampled_count(self):
+        return self.sampled_count