github/krkn
1
0
Fork 0
mirror of https://github.com/krkn-chaos/krkn.git synced 2026-08-25 09:27:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add prometheus trigger type for event-driven chaos (#1517)

Browse Source 
* feat: add prometheus trigger type for event-driven chaos

Signed-off-by: qsxDree <qsxdreee@gmail.com>

* prometheus timeout implemented

Signed-off-by: qsxDree <qsxdreee@gmail.com>

* bumped krkn-lib version

Signed-off-by: qsxDree <qsxdreee@gmail.com>

---------

Signed-off-by: qsxDree <qsxdreee@gmail.com>
Co-authored-by: qsxDree <qsxdreee@gmail.com>
This commit is contained in:
reee
2026-08-18 10:06:23 +05:30
committed by GitHub
co-authored by qsxDree
parent 46a1eddf90
commit 7c1033589c
7 changed files with 424 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/trigger_prometheus_example.yaml
+21
View File
@@ -0,0 +1,21 @@
# Example: gate chaos on a Prometheus PromQL condition.
#
# Set prometheus_url (and optional prometheus_bearer_token) on the condition.
# Chaos starts only after the query returns a non-empty result.
# on_timeout: skip | fail | run_anyway
kraken:
chaos_scenarios:
- network_chaos_ng_scenarios:
- scenarios/kube/pod-network-chaos.yml
triggers:
mode: all_of
timeout: 600
interval: 10
on_timeout: skip
conditions:
- type: prometheus
query: "avg(rate(container_cpu_usage_seconds_total{namespace='production'}[5m])) > 0.8"
prometheus_url: "http://prometheus:9090"
# prometheus_bearer_token: "optional-token"
containers/krknctl-input.json
+31
View File
@@ -873,5 +873,36 @@
"default": "",
"required": "false",
"group": "triggers"
},
{
"name": "trigger-prom-query",
"short_description": "Prometheus trigger query",
"description": "PromQL expression for the prometheus trigger. Chaos begins only after the query returns a non-empty result. Leave empty to disable.",
"variable": "TRIGGER_PROM_QUERY",
"type": "string",
"default": "",
"required": "false",
"group": "triggers"
},
{
"name": "trigger-prom-url",
"short_description": "Prometheus trigger URL",
"description": "Prometheus API URL used by the prometheus trigger condition. Required when trigger-prom-query is set.",
"variable": "TRIGGER_PROM_URL",
"type": "string",
"default": "",
"required": "false",
"group": "triggers"
},
{
"name": "trigger-prom-token",
"short_description": "Prometheus trigger bearer token",
"description": "Optional bearer token for authenticating the prometheus trigger against prometheus-url",
"variable": "TRIGGER_PROM_TOKEN",
"type": "string",
"default": "",
"required": "false",
"secret": "true",
"group": "triggers"
}
]
krkn/scenario_plugins/triggers/__init__.py
+2 -1
View File
@@ -16,5 +16,6 @@ from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger
from krkn.scenario_plugins.triggers.http_trigger import HttpTrigger
from krkn.scenario_plugins.triggers.k8s_trigger import K8sTrigger
from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager
from krkn.scenario_plugins.triggers.prometheus_trigger import PrometheusTrigger
__all__ = ["AbstractTrigger", "CommandTrigger", "HttpTrigger", "K8sTrigger", "TriggerManager"]
__all__ = ["AbstractTrigger", "CommandTrigger", "HttpTrigger", "K8sTrigger", "TriggerManager", "PrometheusTrigger"]
krkn/scenario_plugins/triggers/prometheus_trigger.py
+95
View File
@@ -0,0 +1,95 @@
# Copyright 2026 The Krkn Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import requests
from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
# Cap each PromQL request so a hung Prometheus cannot block evaluate()
# past TriggerManager's deadline between polls (same idea as HttpTrigger).
PROM_REQUEST_TIMEOUT_SECONDS = 30
class PrometheusTrigger(AbstractTrigger):
"""Trigger that evaluates a PromQL query against Prometheus.
Connection details come from the condition config:
``prometheus_url`` and optional ``prometheus_bearer_token``.
The ``KrknPrometheus`` client is created lazily on first evaluate().
Polling / timeout / on_timeout are handled by TriggerManager.
"""
def __init__(self, config: dict):
self._query = config.get("query")
if not self._query:
raise ValueError("prometheus trigger requires a 'query' field")
self._prometheus_url = config.get("prometheus_url")
if not self._prometheus_url:
raise ValueError(
"prometheus trigger requires a 'prometheus_url' field"
)
self._prometheus_bearer_token = (
config.get("prometheus_bearer_token") or None
)
self._prom_client = None
self._last_result: bool | None = None
def _get_prom_client(self):
if self._prom_client is None:
# Lazy import: avoid pulling prometheus_api_client/pandas at
# module import time (TriggerManager / unit-test collection).
from krkn_lib.prometheus.krkn_prometheus import KrknPrometheus
self._prom_client = KrknPrometheus(
self._prometheus_url,
self._prometheus_bearer_token,
timeout=PROM_REQUEST_TIMEOUT_SECONDS,
)
return self._prom_client
def evaluate(self) -> bool:
try:
client = self._get_prom_client()
result = client.process_query(self._query)
met = bool(result)
logging.debug(
"prometheus trigger: query=%r result_count=%s",
self._query,
len(result) if result is not None else 0,
)
except requests.exceptions.Timeout:
logging.warning(
f"prometheus trigger timed out after "
f"{PROM_REQUEST_TIMEOUT_SECONDS}s: query={self._query!r}"
)
met = False
except Exception as e:
logging.warning(f"prometheus trigger query failed: {e}")
met = False
# Log only on state change (same pattern as HttpTrigger)
if met != self._last_result:
if met:
logging.info(f"trigger condition satisfied: {self.describe()}")
else:
logging.info(
f"trigger condition not satisfied: {self.describe()}"
)
self._last_result = met
return met
def describe(self) -> str:
return f"prometheus trigger (query: {self._query})"
krkn/scenario_plugins/triggers/trigger_manager.py
+4
View File
@@ -17,6 +17,7 @@ import time
from krkn.scenario_plugins.triggers.abstract_trigger import AbstractTrigger
from krkn.scenario_plugins.triggers.command_trigger import CommandTrigger
from krkn.scenario_plugins.triggers.http_trigger import HttpTrigger
from krkn.scenario_plugins.triggers.prometheus_trigger import PrometheusTrigger
from krkn.scenario_plugins.triggers.k8s_trigger import K8sTrigger
VALID_MODES = {"all_of", "any_of"}
@@ -105,6 +106,9 @@ class TriggerManager:
if trigger_type == "http":
return HttpTrigger(condition_config)
if trigger_type == "prometheus":
return PrometheusTrigger(condition_config)
if trigger_type == "k8s":
return K8sTrigger(condition_config, kubecli=kubecli)
requirements.txt
+1 -1
View File
@@ -22,7 +22,7 @@ reportlab>=4.0
cbor2<5.7.0 # Pinned by arcaflow-plugin-sdk
lxml==6.1.0
kubernetes>=35.0.0,<36.0.0
krkn-lib==6.1.3
krkn-lib==6.1.4
numpy==1.26.4
pandas==2.2.0
openshift-client==1.0.21
tests/test_triggers/test_prometheus_trigger.py
+270
View File
@@ -0,0 +1,270 @@
#!/usr/bin/env python3
"""
Test suite for PrometheusTrigger class
Usage:
python -m coverage run -a -m unittest tests/test_triggers/test_prometheus_trigger.py -v
"""
import sys
import unittest
from types import ModuleType
from unittest.mock import MagicMock, patch
from krkn.scenario_plugins.triggers.prometheus_trigger import (
PROM_REQUEST_TIMEOUT_SECONDS,
PrometheusTrigger,
)
from krkn.scenario_plugins.triggers.trigger_manager import TriggerManager
class TestPrometheusTrigger(unittest.TestCase):
def _make_config(self, **overrides):
config = {
"query": "up > 0",
"prometheus_url": "http://prometheus:9090",
}
config.update(overrides)
return config
def _mock_client(self, result=None, side_effect=None):
client = MagicMock()
if side_effect is not None:
client.process_query.side_effect = side_effect
else:
client.process_query.return_value = result
return client
def _make_trigger(self, client=None, **overrides):
"""Build a trigger; inject client so tests skip real KrknPrometheus import."""
trigger = PrometheusTrigger(self._make_config(**overrides))
if client is not None:
trigger._prom_client = client
return trigger
def _patch_krkn_prometheus(self, mock_cls):
"""Install a fake krkn_lib.prometheus.krkn_prometheus module for lazy import."""
mod = ModuleType("krkn_lib.prometheus.krkn_prometheus")
mod.KrknPrometheus = mock_cls
return patch.dict(
sys.modules,
{
"krkn_lib.prometheus.krkn_prometheus": mod,
},
)
# ------------------------------------------------------------------
# evaluate() tests
# ------------------------------------------------------------------
def test_evaluate_non_empty_result(self):
"""Non-empty PromQL result -> evaluate() returns True."""
client = self._mock_client(
result=[{"metric": {"__name__": "up"}, "value": [1.0, "1"]}]
)
trigger = self._make_trigger(client=client)
self.assertTrue(trigger.evaluate())
client.process_query.assert_called_once_with("up > 0")
def test_evaluate_empty_result(self):
"""Empty list -> evaluate() returns False."""
trigger = self._make_trigger(client=self._mock_client(result=[]))
self.assertFalse(trigger.evaluate())
def test_evaluate_connection_error(self):
"""HTTP/connection errors -> warning logged, returns False, no crash."""
trigger = self._make_trigger(
client=self._mock_client(side_effect=ConnectionError("refused"))
)
with self.assertLogs(level="WARNING") as cm:
self.assertFalse(trigger.evaluate())
self.assertTrue(
any("prometheus trigger query failed" in msg for msg in cm.output)
)
def test_evaluate_request_timeout(self):
"""requests.Timeout -> warning with timeout, returns False."""
import requests
trigger = self._make_trigger(
client=self._mock_client(
side_effect=requests.exceptions.Timeout("hung")
)
)
with self.assertLogs(level="WARNING") as cm:
self.assertFalse(trigger.evaluate())
self.assertTrue(
any(
f"timed out after {PROM_REQUEST_TIMEOUT_SECONDS}s" in msg
for msg in cm.output
)
)
def test_evaluate_generic_exception(self):
"""Unexpected exception -> False, no crash."""
trigger = self._make_trigger(
client=self._mock_client(side_effect=RuntimeError("boom"))
)
with self.assertLogs(level="WARNING"):
self.assertFalse(trigger.evaluate())
def test_client_built_lazily_once(self):
"""KrknPrometheus is created on first evaluate and reused."""
mock_cls = MagicMock(
return_value=self._mock_client(result=[{"value": [0, "1"]}])
)
trigger = self._make_trigger(prometheus_bearer_token="tok")
self.assertIsNone(trigger._prom_client)
with self._patch_krkn_prometheus(mock_cls):
trigger.evaluate()
trigger.evaluate()
mock_cls.assert_called_once_with(
"http://prometheus:9090",
"tok",
timeout=PROM_REQUEST_TIMEOUT_SECONDS,
)
self.assertEqual(mock_cls.return_value.process_query.call_count, 2)
def test_empty_bearer_token_normalized_to_none(self):
"""Empty-string bearer token becomes None (same idea as HttpTrigger)."""
mock_cls = MagicMock(
return_value=self._mock_client(result=[{"value": [0, "1"]}])
)
trigger = self._make_trigger(prometheus_bearer_token="")
with self._patch_krkn_prometheus(mock_cls):
trigger.evaluate()
mock_cls.assert_called_once_with(
"http://prometheus:9090",
None,
timeout=PROM_REQUEST_TIMEOUT_SECONDS,
)
def test_state_change_logging(self):
"""INFO logs only on state transitions, not every poll."""
client = MagicMock()
trigger = self._make_trigger(client=client)
client.process_query.return_value = []
with self.assertLogs(level="INFO") as log_ctx:
trigger.evaluate()
self.assertTrue(
any(
"trigger condition not satisfied" in line
for line in log_ctx.output
)
)
with patch("logging.info") as mock_info:
trigger.evaluate()
mock_info.assert_not_called()
client.process_query.return_value = [{"value": [0, "1"]}]
with self.assertLogs(level="INFO") as log_ctx:
trigger.evaluate()
self.assertTrue(
any(
"trigger condition satisfied" in line
for line in log_ctx.output
)
)
# ------------------------------------------------------------------
# timeout via TriggerManager
# ------------------------------------------------------------------
@patch("krkn.scenario_plugins.triggers.trigger_manager.time")
def test_timeout_reached_without_match(self, mock_time):
"""Empty results until deadline -> wait_for_triggers returns False."""
mock_cls = MagicMock(return_value=self._mock_client(result=[]))
call_count = 0
def advancing_monotonic():
nonlocal call_count
call_count += 1
if call_count <= 2:
return 0.0
return 999.0
mock_time.monotonic.side_effect = advancing_monotonic
mock_time.sleep = lambda x: None
with self._patch_krkn_prometheus(mock_cls):
manager = TriggerManager(
{
"timeout": 10,
"interval": 1,
"conditions": [
{
"type": "prometheus",
"query": "vector(0) > 1",
"prometheus_url": "http://prometheus:9090",
},
],
},
)
self.assertFalse(manager.wait_for_triggers())
def test_manager_builds_prometheus_trigger(self):
mock_cls = MagicMock(
return_value=self._mock_client(result=[{"value": [0, "1"]}])
)
with self._patch_krkn_prometheus(mock_cls):
manager = TriggerManager(
{
"conditions": [
{
"type": "prometheus",
"query": "up == 1",
"prometheus_url": "http://prometheus:9090",
},
],
},
)
self.assertTrue(manager.wait_for_triggers())
mock_cls.return_value.process_query.assert_called_with("up == 1")
# ------------------------------------------------------------------
# describe() / validation
# ------------------------------------------------------------------
def test_describe(self):
trigger = self._make_trigger(query="avg(rate(cpu[5m])) > 0.8")
description = trigger.describe()
self.assertIn("prometheus", description)
self.assertIn("avg(rate(cpu[5m])) > 0.8", description)
def test_missing_query_raises(self):
with self.assertRaises(ValueError) as ctx:
PrometheusTrigger({"prometheus_url": "http://prometheus:9090"})
self.assertIn("query", str(ctx.exception))
def test_empty_query_raises(self):
with self.assertRaises(ValueError):
PrometheusTrigger(
{"query": "", "prometheus_url": "http://prometheus:9090"}
)
def test_missing_prometheus_url_raises(self):
with self.assertRaises(ValueError) as ctx:
PrometheusTrigger({"query": "up"})
self.assertIn("prometheus_url", str(ctx.exception))
def test_last_result_initialised_to_none(self):
trigger = self._make_trigger()
self.assertIsNone(trigger._last_result)
if __name__ == "__main__":
unittest.main()