From eed39bfb2f3fff2f33e964e46d59a13fa75e8dcc Mon Sep 17 00:00:00 2001 From: Darshan Jain Date: Mon, 24 Aug 2026 13:33:55 +0530 Subject: [PATCH] feat: parallelize SLO PromQL queries with ThreadPoolExecutor (#1560) (#1572) * feat: parallelize SLO PromQL queries with ThreadPoolExecutor (#1560) Use concurrent.futures.ThreadPoolExecutor to run independent SLO range queries in parallel, reducing per-pass evaluation from ~11-12s to ~2-3s. Includes configurable max_workers (default 10, capped to SLO count), per-future error isolation for malformed SLO dicts, and comprehensive unit tests for parallelism, mixed results, and exception isolation. Signed-off-by: Darshan Jain Signed-off-by: ddjain Co-authored-by: Cursor * fix: prevent malformed SLOs from silently inflating resiliency score - Move slo["expr"]/slo["name"] access inside try block so KeyError is caught and re-raised as ValueError with context - Write failure result in the outer future.result() handler so malformed SLOs appear in the results mapping instead of being dropped - Treat SLOs missing from prometheus_results as failed (score.py) rather than excluding them, which previously inflated the score - Clamp max_workers to minimum of 1 to prevent ValueError from ThreadPoolExecutor when max_workers <= 0 Co-authored-by: Cursor Signed-off-by: ddjain Co-authored-by: Cursor --------- Signed-off-by: Darshan Jain Signed-off-by: ddjain Co-authored-by: Cursor --- krkn/prometheus/collector.py | 49 ++++-- krkn/resiliency/score.py | 7 +- tests/test_prometheus_collector.py | 232 +++++++++++++++++++++++++++++ tests/test_resiliency_score.py | 21 +-- 4 files changed, 283 insertions(+), 26 deletions(-) diff --git a/krkn/prometheus/collector.py b/krkn/prometheus/collector.py index c7737449..34e98a90 100644 --- a/krkn/prometheus/collector.py +++ b/krkn/prometheus/collector.py @@ -18,6 +18,7 @@ from __future__ import annotations import datetime import logging +from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Dict, Any, List, Optional from krkn_lib.prometheus.krkn_prometheus import KrknPrometheus @@ -57,39 +58,63 @@ def evaluate_slos( slo_list: List[Dict[str, Any]], start_time: datetime.datetime, end_time: datetime.datetime, + max_workers: int = 10, ) -> Dict[str, bool]: - """Evaluate a list of SLO expressions against Prometheus. + """Evaluate a list of SLO expressions against Prometheus in parallel. Args: prom_cli: Configured Prometheus client. slo_list: List of dicts with keys ``name``, ``expr``. start_time: Start timestamp. end_time: End timestamp. - granularity: Step in seconds for range queries. + max_workers: Maximum number of concurrent PromQL queries. Clamped to + a minimum of 1. Returns: Mapping name -> bool indicating pass status. True means good we passed the SLO test otherwise failed the SLO """ results: Dict[str, bool] = {} + + if not slo_list: + return results + logging.info("Evaluating %d SLOs over window %s – %s", len(slo_list), start_time, end_time) - for slo in slo_list: - expr = slo["expr"] - name = slo["name"] + + def _eval_single(slo: Dict[str, Any]) -> tuple[str, bool]: + try: + expr = slo["expr"] + name = slo["name"] + except KeyError as exc: + raise ValueError( + f"Malformed SLO definition (missing key {exc}): {slo!r}" + ) from exc try: response = prom_cli.process_prom_query_in_range( expr, start_time=start_time, end_time=end_time, ) - passed = slo_passed(response) if passed is None: - # Absence of data indicates the condition did not trigger; treat as pass. logging.debug("SLO '%s' query returned no data; assuming pass.", name) - results[name] = True - else: - results[name] = passed - except Exception as exc: + return name, True + return name, passed + except Exception as exc: logging.error("PromQL query failed for SLO '%s': %s", name, exc) - results[name] = False + return name, False + + worker_count = min(len(slo_list), max(1, max_workers)) + with ThreadPoolExecutor(max_workers=worker_count) as pool: + futures = {pool.submit(_eval_single, slo): slo for slo in slo_list} + for future in as_completed(futures): + try: + name, passed = future.result() + results[name] = passed + except Exception as exc: + slo = futures[future] + slo_name = slo.get("name", "") + logging.error("Unexpected error evaluating SLO '%s': %s", + slo_name, exc) + results[slo_name] = False + return results diff --git a/krkn/resiliency/score.py b/krkn/resiliency/score.py index 1b53c3ff..319ef556 100644 --- a/krkn/resiliency/score.py +++ b/krkn/resiliency/score.py @@ -48,7 +48,7 @@ def calculate_resiliency_score( slo_definitions: Mapping of SLO name -> severity ("critical" | "warning") OR SLO name -> {"severity": str, "weight": int | None}. prometheus_results: Mapping of SLO name -> bool indicating whether the SLO - passed. Any SLO missing in this mapping is treated as failed. + passed. Any SLO defined but missing from this mapping is treated as failed. health_check_results: Mapping of custom health-check name -> bool pass flag. These checks are always treated as *critical*. @@ -59,10 +59,7 @@ def calculate_resiliency_score( slo_objects: List[SLOResult] = [] for slo_name, slo_def in slo_definitions.items(): - # Exclude SLOs that were not evaluated (query returned no data) - if slo_name not in prometheus_results: - continue - passed = bool(prometheus_results[slo_name]) + passed = bool(prometheus_results.get(slo_name, False)) # Support both old format (str) and new format (dict) if isinstance(slo_def, str): diff --git a/tests/test_prometheus_collector.py b/tests/test_prometheus_collector.py index 11bbc3fc..81859f0b 100644 --- a/tests/test_prometheus_collector.py +++ b/tests/test_prometheus_collector.py @@ -26,6 +26,7 @@ How to run these tests: import datetime import unittest from unittest.mock import Mock, patch, MagicMock +from concurrent.futures import ThreadPoolExecutor from krkn.prometheus.collector import slo_passed, evaluate_slos @@ -397,6 +398,237 @@ class TestEvaluateSLOs(unittest.TestCase): self.assertIn("no data", call_args[0]) self.assertIn("test_slo", call_args[1]) + def test_evaluate_slos_respects_max_workers(self): + """Test that max_workers caps to slo count when slo count < max_workers.""" + slo_list = [ + {"name": f"slo_{i}", "expr": f"query_{i}"} + for i in range(8) + ] + + self.mock_prom_cli.process_prom_query_in_range.return_value = [ + {"values": [[1234567890, "0"]]} + ] + + with patch('krkn.prometheus.collector.ThreadPoolExecutor', wraps=ThreadPoolExecutor) as mock_pool: + evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time, + max_workers=25 + ) + # min(8, 25) = 8 — pool should be capped to the SLO count + mock_pool.assert_called_once_with(max_workers=8) + + def test_evaluate_slos_max_workers_capped_to_slo_count(self): + """Test that max_workers is capped to the number of SLOs.""" + slo_list = [ + {"name": "slo_1", "expr": "query_1"}, + {"name": "slo_2", "expr": "query_2"}, + ] + + self.mock_prom_cli.process_prom_query_in_range.return_value = [ + {"values": [[1234567890, "0"]]} + ] + + with patch('krkn.prometheus.collector.ThreadPoolExecutor', wraps=ThreadPoolExecutor) as mock_pool: + evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time, + max_workers=10 + ) + # Should use min(2, 10) = 2 workers + mock_pool.assert_called_once_with(max_workers=2) + + def test_evaluate_slos_parallel_all_queries_executed(self): + """Test that all SLOs are evaluated when running in parallel.""" + slo_list = [ + {"name": f"slo_{i}", "expr": f"query_{i}"} + for i in range(15) + ] + + self.mock_prom_cli.process_prom_query_in_range.return_value = [ + {"values": [[1234567890, "0"]]} + ] + + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time + ) + + self.assertEqual(len(results), 15) + self.assertEqual(self.mock_prom_cli.process_prom_query_in_range.call_count, 15) + for i in range(15): + self.assertTrue(results[f"slo_{i}"]) + + def test_evaluate_slos_parallel_mixed_results(self): + """Test parallel evaluation with mixed pass/fail/error results.""" + slo_list = [ + {"name": "slo_pass", "expr": "pass_query"}, + {"name": "slo_fail", "expr": "fail_query"}, + {"name": "slo_error", "expr": "error_query"}, + {"name": "slo_nodata", "expr": "nodata_query"}, + ] + + def mock_query_side_effect(expr, start_time, end_time): + if expr == "pass_query": + return [{"values": [[1234567890, "0"]]}] + elif expr == "fail_query": + return [{"values": [[1234567890, "1"]]}] + elif expr == "error_query": + raise Exception("Connection timeout") + else: + return [] + + self.mock_prom_cli.process_prom_query_in_range.side_effect = mock_query_side_effect + + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time, + max_workers=4 + ) + + self.assertTrue(results["slo_pass"]) + self.assertFalse(results["slo_fail"]) + self.assertFalse(results["slo_error"]) + self.assertTrue(results["slo_nodata"]) + + def test_evaluate_slos_exception_isolation(self): + """Test that one SLO exception doesn't affect others.""" + slo_list = [ + {"name": "slo_before", "expr": "query_before"}, + {"name": "slo_broken", "expr": "query_broken"}, + {"name": "slo_after", "expr": "query_after"}, + ] + + def mock_query_side_effect(expr, start_time, end_time): + if expr == "query_broken": + raise RuntimeError("Prometheus unavailable") + return [{"values": [[1234567890, "0"]]}] + + self.mock_prom_cli.process_prom_query_in_range.side_effect = mock_query_side_effect + + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time + ) + + self.assertTrue(results["slo_before"]) + self.assertFalse(results["slo_broken"]) + self.assertTrue(results["slo_after"]) + + def test_evaluate_slos_single_worker(self): + """Test that max_workers=1 (sequential fallback) works correctly.""" + slo_list = [ + {"name": "slo_a", "expr": "query_a"}, + {"name": "slo_b", "expr": "query_b"}, + {"name": "slo_c", "expr": "query_c"}, + ] + + def mock_query_side_effect(expr, start_time, end_time): + if expr == "query_b": + return [{"values": [[1234567890, "1"]]}] + return [{"values": [[1234567890, "0"]]}] + + self.mock_prom_cli.process_prom_query_in_range.side_effect = mock_query_side_effect + + with patch('krkn.prometheus.collector.ThreadPoolExecutor', wraps=ThreadPoolExecutor) as mock_pool: + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time, + max_workers=1 + ) + mock_pool.assert_called_once_with(max_workers=1) + + self.assertEqual(len(results), 3) + self.assertTrue(results["slo_a"]) + self.assertFalse(results["slo_b"]) + self.assertTrue(results["slo_c"]) + + def test_evaluate_slos_malformed_slo_dict(self): + """Test that a SLO dict missing required keys is recorded as failed.""" + slo_list = [ + {"name": "good_slo", "expr": "good_query"}, + {"name": "no_expr_slo"}, # missing "expr" + {"expr": "orphan_query"}, # missing "name" + {"name": "another_good", "expr": "another_query"}, + ] + + self.mock_prom_cli.process_prom_query_in_range.return_value = [ + {"values": [[1234567890, "0"]]} + ] + + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time + ) + + self.assertTrue(results["good_slo"]) + self.assertTrue(results["another_good"]) + # Malformed SLOs are recorded as failed + self.assertFalse(results["no_expr_slo"]) + self.assertFalse(results[""]) + + def test_evaluate_slos_zero_max_workers_clamped_to_one(self): + """Test that max_workers=0 is clamped to 1 instead of raising ValueError.""" + slo_list = [ + {"name": "slo_1", "expr": "query_1"}, + ] + + self.mock_prom_cli.process_prom_query_in_range.return_value = [ + {"values": [[1234567890, "0"]]} + ] + + with patch('krkn.prometheus.collector.ThreadPoolExecutor', wraps=ThreadPoolExecutor) as mock_pool: + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time, + max_workers=0 + ) + mock_pool.assert_called_once_with(max_workers=1) + + self.assertEqual(len(results), 1) + self.assertTrue(results["slo_1"]) + + def test_evaluate_slos_negative_max_workers_clamped_to_one(self): + """Test that negative max_workers is clamped to 1.""" + slo_list = [ + {"name": "slo_a", "expr": "query_a"}, + {"name": "slo_b", "expr": "query_b"}, + ] + + self.mock_prom_cli.process_prom_query_in_range.return_value = [ + {"values": [[1234567890, "0"]]} + ] + + with patch('krkn.prometheus.collector.ThreadPoolExecutor', wraps=ThreadPoolExecutor) as mock_pool: + results = evaluate_slos( + self.mock_prom_cli, + slo_list, + self.start_time, + self.end_time, + max_workers=-5 + ) + mock_pool.assert_called_once_with(max_workers=1) + + self.assertEqual(len(results), 2) + self.assertTrue(results["slo_a"]) + self.assertTrue(results["slo_b"]) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_resiliency_score.py b/tests/test_resiliency_score.py index 63730db5..099a03fb 100644 --- a/tests/test_resiliency_score.py +++ b/tests/test_resiliency_score.py @@ -158,17 +158,17 @@ class TestCalculateResiliencyScore(unittest.TestCase): self.assertEqual(breakdown["passed"], 1) self.assertEqual(breakdown["failed"], 1) - def test_slo_not_in_prometheus_results_is_excluded(self): - """Test that SLOs not in prometheus_results are excluded from calculation.""" + def test_slo_not_in_prometheus_results_is_treated_as_failed(self): + """Test that SLOs not in prometheus_results are treated as failed.""" slo_definitions = { - "slo1": "critical", - "slo2": "warning", - "slo3": "critical", # Not in prometheus_results + "slo1": "critical", # weight=3 + "slo2": "warning", # weight=1 + "slo3": "critical", # weight=3, not in prometheus_results -> failed } prometheus_results = { "slo1": True, "slo2": True, - # slo3 is missing (no data) + # slo3 is missing -> treated as failed } health_check_results = {} @@ -176,10 +176,13 @@ class TestCalculateResiliencyScore(unittest.TestCase): slo_definitions, prometheus_results, health_check_results ) - # Only slo1 and slo2 should be counted - self.assertEqual(score, 100) + # All three SLOs counted: total=3+1+3=7, lost=3 (slo3 failed) + # Score: (7-3)/7 * 100 = 57.14... -> 57 + self.assertEqual(score, 57) self.assertEqual(breakdown["passed"], 2) - self.assertEqual(breakdown["failed"], 0) + self.assertEqual(breakdown["failed"], 1) + self.assertEqual(breakdown["total_points"], 7) + self.assertEqual(breakdown["points_lost"], 3) def test_health_checks_are_treated_as_critical(self): """Test that health checks are always weighted as critical."""