mirror of
https://github.com/krkn-chaos/krkn.git
synced 2026-08-25 09:27:36 +00:00
* 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 <ddjain@redhat.com> Signed-off-by: ddjain <darjain@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com> * 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 <cursoragent@cursor.com> Signed-off-by: ddjain <darjain@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com> --------- Signed-off-by: Darshan Jain <ddjain@redhat.com> Signed-off-by: ddjain <darjain@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Darshan Jain
Cursor
@@ -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", "<unknown>")
|
||||
logging.error("Unexpected error evaluating SLO '%s': %s",
|
||||
slo_name, exc)
|
||||
results[slo_name] = False
|
||||
|
||||
return results
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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["<unknown>"])
|
||||
|
||||
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()
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user