diff --git a/config/trigger_prometheus_example.yaml b/config/trigger_prometheus_example.yaml
new file mode 100644
index 00000000..66085241
--- /dev/null
+++ b/config/trigger_prometheus_example.yaml
@@ -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"
diff --git a/containers/krknctl-input.json b/containers/krknctl-input.json
index cc807c80..c407dd18 100644
--- a/containers/krknctl-input.json
+++ b/containers/krknctl-input.json
@@ -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"
   }
 ]
\ No newline at end of file
diff --git a/krkn/scenario_plugins/triggers/__init__.py b/krkn/scenario_plugins/triggers/__init__.py
index af3d5d02..63727e73 100644
--- a/krkn/scenario_plugins/triggers/__init__.py
+++ b/krkn/scenario_plugins/triggers/__init__.py
@@ -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"]
diff --git a/krkn/scenario_plugins/triggers/prometheus_trigger.py b/krkn/scenario_plugins/triggers/prometheus_trigger.py
new file mode 100644
index 00000000..4973d362
--- /dev/null
+++ b/krkn/scenario_plugins/triggers/prometheus_trigger.py
@@ -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})"
diff --git a/krkn/scenario_plugins/triggers/trigger_manager.py b/krkn/scenario_plugins/triggers/trigger_manager.py
index adb09825..9f3e84e3 100644
--- a/krkn/scenario_plugins/triggers/trigger_manager.py
+++ b/krkn/scenario_plugins/triggers/trigger_manager.py
@@ -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)
 
diff --git a/requirements.txt b/requirements.txt
index df9c18cf..490d2889 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -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
diff --git a/tests/test_triggers/test_prometheus_trigger.py b/tests/test_triggers/test_prometheus_trigger.py
new file mode 100644
index 00000000..692945ce
--- /dev/null
+++ b/tests/test_triggers/test_prometheus_trigger.py
@@ -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()