Skip to content

deep.processor.context.metric_action

Provide metric actions.

MetricActionContext

Bases: ActionContext

Action for metrics.

Source code in deep/processor/context/metric_action.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class MetricActionContext(ActionContext):
    """Action for metrics."""

    def can_trigger(self) -> bool:
        """
        Check if the action can trigger.

        If we do not have a metric processor enabled, then skip this action.
        :return: True, if the trigger can be triggered.
        """
        if self.__has_metric_processor():
            return super().can_trigger()
        return False

    def _process_action(self):
        metrics = self._metrics()
        for metric in metrics:
            labels, value = self._process_metric(metric)
            for processor in self.trigger_context.config.metric_processors:
                getattr(processor, self._convert_type(metric.type))(metric.name, labels, metric.namespace or "deep",
                                                                    metric.help, metric.unit, value)

    def __has_metric_processor(self):
        return self.trigger_context.config.has_metric_processor

    def _metrics(self) -> List[MetricDefinition]:
        return self.location_action.config['metrics']

    def _convert_type(self, metric_type):
        return metric_type.lower()

    def _process_metric(self, metric: MetricDefinition) -> Tuple[Dict[str, str], float]:
        metric_value = 1
        if metric.expression:
            try:
                metric_value = float(self.trigger_context.evaluate_expression(metric.expression))
            except Exception:
                deep.logging.exception("Cannot process metric expression %s", metric.expression)

        labels = {}
        if len(metric.labels) > 0:
            for label in metric.labels:
                key = label.key
                if label.expression:
                    try:
                        value = str(self.trigger_context.evaluate_expression(label.expression))
                    except Exception:
                        deep.logging.exception("Cannot process metric label expression %s: %s", key, label.expression)
                        value = 'expression failed'
                else:
                    value = label.static
                labels[key] = value
        return labels, metric_value

can_trigger()

Check if the action can trigger.

If we do not have a metric processor enabled, then skip this action. :return: True, if the trigger can be triggered.

Source code in deep/processor/context/metric_action.py 
28
29
30
31
32
33
34
35
36
37
def can_trigger(self) -> bool:
    """
    Check if the action can trigger.

    If we do not have a metric processor enabled, then skip this action.
    :return: True, if the trigger can be triggered.
    """
    if self.__has_metric_processor():
        return super().can_trigger()
    return False