feat(Bot): add on_metric value threshold task handler

Author: fubuloubu

bots/example.py

@@ -107,11 +107,17 @@ async def exec_event2(log: ContractLog):
 
 # You can run cron jobs in your apps
 # (useful for functions that execute at a regular time period e.g. metrics)
-@bot.cron("*/1 * * * *")
+@bot.cron("* * * * *")
 def sample_metric(time: datetime):
     return random.random()
 
 
+# You can trigger tasks when metrics returned by other tasks exceed a threshold
+@bot.on_metric("sample_metric", gt=0.5)
+def metric_too_high(value: float):
+    raise ValueError("The metric is too damn high!")
+
+
 @bot.cron("*/5 * * * *")
 # NOTE: You can have multiple handlers for any trigger we support
 def stop_running_now(time):

silverback/main.py

@@ -23,7 +23,7 @@
 from .exceptions import ContainerTypeMismatchError, InvalidContainerTypeError, NoSignerLoaded
 from .settings import Settings
 from .state import StateSnapshot
-from .types import SilverbackID, TaskType
+from .types import ScalarType, SilverbackID, TaskType
 from .utils import encode_topics_to_string, parse_hexbytes_dict
 
 
@@ -377,8 +377,10 @@ def broker_task_decorator(
         self,
         task_type: TaskType,
         container: BlockContainer | ContractEvent | ContractEventWrapper | None = None,
-        cron_schedule: str | None = None,
         filter_args: dict[str, Any] | None = None,
+        cron_schedule: str | None = None,
+        metric_name: str | None = None,
+        value_threshold: dict[str, ScalarType] | None = None,
     ) -> Callable[[Callable], AsyncTaskiqDecoratedTask]:
         """
         Dynamically create a new broker task that handles tasks of ``task_type``.
@@ -483,6 +485,16 @@ def add_taskiq_task(
 
                 labels["cron"] = cron_schedule
 
+            elif task_type is TaskType.METRIC_VALUE:
+                # NOTE: This shouldn't happen to users
+                assert metric_name, "Must supply `metric_name=`."
+                labels["metric"] = metric_name
+
+                if value_threshold:
+                    labels.update(
+                        {f"value:{lbl}": str(val) for lbl, val in value_threshold.items()}
+                    )
+
             self.tasks[task_type].append(TaskData(name=handler.__name__, labels=labels))
 
             if self.use_fork:
@@ -633,3 +645,55 @@ def cron(self, cron_schedule: str) -> Callable:
             cron_schedule (str): A cron-like schedule string.
         """
         return self.broker_task_decorator(TaskType.CRON_JOB, cron_schedule=cron_schedule)
+
+    def on_metric(
+        self,
+        metric_name: str,
+        ge: ScalarType | None = None,
+        gt: ScalarType | None = None,
+        le: ScalarType | None = None,
+        lt: ScalarType | None = None,
+        eq: ScalarType | None = None,
+        ne: ScalarType | None = None,
+        # TODO: Support `rate_[ge|gt|le|...]` too?
+    ) -> Callable:
+        """
+        Create a task that runs when the value of a specified metric has tripped a threshold.
+
+        ```{notice}
+        If no keyword args provided to this decorator, it will trigger on every update of metric.
+        ```
+
+        ```{notice}
+        Multiple keyword args provided to this decorator means logical AND of all of them.
+        ```
+
+        Args:
+            metric_name (str): The name of the metric to monitor for threshold exceedence.
+            ge (ScalarType | None): trigger when metric value is greater than or equal to value.
+            gt (ScalarType | None): trigger when metric value is greater than value.
+            le (ScalarType | None): trigger when metric value is less than or equal to value.
+            lt (ScalarType | None): trigger when metric value is less than value.
+            eq (ScalarType | None): trigger when metric value is equal to value.
+            ne (ScalarType | None): trigger when metric value is not equal to value.
+        """
+        value_threshold: dict[str, ScalarType] = {}
+        if ge is not None:
+            value_threshold["ge"] = ge
+        if gt is not None:
+            value_threshold["gt"] = gt
+        if le is not None:
+            value_threshold["le"] = le
+        if lt is not None:
+            value_threshold["lt"] = lt
+        if eq is not None:
+            value_threshold["eq"] = eq
+        if ne is not None:
+            value_threshold["ne"] = ne
+
+        return self.broker_task_decorator(
+            TaskType.METRIC_VALUE,
+            metric_name=metric_name,
+            # NOTE: When empty, allow all values
+            value_threshold=value_threshold or None,
+        )

silverback/types.py

@@ -25,6 +25,7 @@ class TaskType(str, Enum):
     CRON_JOB = "user:cron-job"
     NEW_BLOCK = "user:new-block"
     EVENT_LOG = "user:event-log"
+    METRIC_VALUE = "user:metric-value"
     SHUTDOWN = "user:shutdown"
 
     def __str__(self) -> str: