Plugin API

prometheus-virtual-metrics plugins are simple Python classes that can define hooks that get called on certain Grafana requests. A Grafana request can be a query for data or metric metadata.

All hooks are optional. None of them has to be implemented. To get started and to see something in Grafana, on_range_query_request() will be enough as a starting point.

All hooks can be async, and none has to return something. Most hooks get injected a request and a response which contain all API you will need.

prometheus-virtual-metrics does no other error handling than logging it when a plugin hook crashes. If this happens in a startup hook, the server crashes and stops. If this happens in a request hook, an HTTP 500 is returned. If you have flaky plugins and want to shield the other plugins from them, use CatchErrorsPlugin.

class ExamplePlugin:

    # data request hooks
    def on_range_query_request(self, request, response):
        """
        Gets called for every PromQL range query.

        Args:
            request (prometheus_virtual_metrics.PrometheusRequest): Prometheus request
            response (prometheus_virtual_metrics.PrometheusResponse): Prometheus response
        """

        # We can check whether the our metric matches the requested metrics.
        # This is not strictly necessary because `response.add_sample()` checks
        # this too but it can help with performance when adding a lot
        # of samples
        if not request.query.name_matches(self.METRIC_NAME):
            return

        # `PrometheusRequest.timestamps` yields all timestamps between
        # `PrometheusRequest.start` and `PrometheusRequest.end`
        for timestamp in request.timestamps:
            response.add_sample(
                'example_metric',
                metric_value=1,
                metric_labels={
                    'label1': 'value1',
                },
                timestamp=timestamp,
            )

    def on_instant_query_request(self, request, response):
        """
        Gets called for every PromQL instant query (singular value per metric
        at given time).

        Args:
            request (prometheus_virtual_metrics.PrometheusRequest): Prometheus request
            response (prometheus_virtual_metrics.PrometheusResponse): Prometheus response
        """

        if not request.query.name_matches(self.METRIC_NAME):
            return

        response.add_sample(
            'example_metric',
            metric_value=1,
            metric_labels={
                'label1': 'value1',
            },
            timestamp=request.time,
        )

    # metrics and label discovery hooks
    def on_metric_names_request(self, request, response):
        """
        Gets called when Grafana tries to explore all available metrics for
        a certain point in time or in a time range.

        Args:
            request (prometheus_virtual_metrics.PrometheusRequest): Prometheus request
            response (prometheus_virtual_metrics.PrometheusResponse): Prometheus response
        """

        # we can check whether the requested metrics name or name part
        # match our metric name
        if not request.query.name_matches('example_metric'):
            return

        response.add_value('example_metric')

    def on_label_names_request(self, request, response):
        """
        Gets called when Grafana tries to explore all available label names
        for a metric at a certain point in time or in a time range.

        Args:
            request (prometheus_virtual_metrics.PrometheusRequest): Prometheus request
            response (prometheus_virtual_metrics.PrometheusResponse): Prometheus response
        """

        if not request.query.name_matches('example_metric'):
            return

        response.add_value([
            'label1',
            'label2',
        ])

    def on_label_values_request(self, request, response):
        """
        Gets called when Grafana tries to explore all available label values
        for a metric and label at a certain point in time or in a time range.

        Args:
            request (prometheus_virtual_metrics.PrometheusRequest): Prometheus request
            response (prometheus_virtual_metrics.PrometheusResponse): Prometheus response
        """

        if not request.query.name_matches(self.METRIC_NAME):
            return

        if request.label_name == 'label1':
            response.add_value('value1')

        elif request.label_name == 'label2':
            response.add_value([
                'value2.1',
                'value2.2',
            ])

    # context hooks
    def on_startup(self, context):
        """
        Gets called on startup.

        context (prometheus_virtual_metrics.PrometheusVirtualMetricsContext): prometheus-virtual-metrics context object
        """

        pass

    def on_shutdown(self, context):
        """
        Gets called on shutdown.

        context (prometheus_virtual_metrics.PrometheusVirtualMetricsContext): prometheus-virtual-metrics context object
        """

        pass

`prometheus_virtual_metrics.PrometheusVirtualMetricsContext`

Attributes:

Name	Type	Description
`settings`	`module \| namespace`	Central settings

`discover_plugin_hooks()`

Discover plugin hooks in settings.PLUGINS.

`handle_prometheus_request(prometheus_request)`

Handle Prometheus request.

Parameters:

Name	Type	Description	Default
`prometheus_request`	`PrometheusRequest`	prometheus request	required

Returns:

Name	Type	Description
`prometheus_response`	`PrometheusResponse`	prometheus_response

`run_plugin_hook(hook_name, hook_args=None, hook_kwargs=None)`

Run plugin hook.

Parameters:

Name	Type	Description	Default
`hook_name`	`str`	Name of the hook to run	required
`hook_args`	`tuple \| None`	Hook args	`None`
`hook_kwargs`	`tuple \| dict`	Hook keyword args	`None`

`valid_prometheus_request_path(path)`

Returns True if the given path is a valid Prometheus request path.

Parameters:

Name	Type	Description	Default
`path`	`str`	HTTP request path	required

Returns:

Name	Type	Description
`path_is_valid`	`bool`	path_is_valid

`prometheus_virtual_metrics.PrometheusRequest`

Attributes:

Name	Type	Description
`query_string`	`str`	PromQl query of incoming request as string
`query`	`PromqlQuery`	PromQl query of incoming request
`time`	`datetime \| None`	Requested time. Is only set on instant queries
`start`	`datetime \| None`	Start of the requested time range. Is only set on range queries
`end`	`datetime \| None`	End of the requested time range. Is only set on range queries
`step`	`int \| None`	Requested interval between samples. Is only set on range queries
`label_name`	`str`	Requested label. Is only set label value requests
`timestamps`	`Generator[datetime]`	List of timestamps between `start` and `end` with an interval of `step`
`duration_string`	`str`	`step` as duration string. Example: `30s` (30 Seconds)
`context`	`PrometheusVirtualMetricsContext`	prometheus-virtual-metrics context
`http_remote`	`str`	HTTP client IP
`http_headers`	`CIMultiDict`	HTTP header of incoming request
`http_path`	`str`	HTTP path of incoming request
`http_query`	`CIMultiDict`	HTTP query of incoming request
`http_post_data`	`CIMultiDict`	HTTP post data of incoming request

`prometheus_virtual_metrics.PromqlQuery`

Attributes:

Name	Type	Description
`string`	`str`	PromQL query as string
`ast`	`Ast \| None`	PromQL query object
`name`	`str`	Requested metric name in `ast`

`matches(name='', labels=None)`

Returns True when the given string and labels match query

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name	`''`
`labels`	`dict[str, str] \| None`	Metric labels	`None`

Returns:

Type	Description
`bool`	Given string and labels match `query`

`name_matches(name)`

Returns True when the given string matches query

Parameters:

Name	Type	Description	Default
`name`	`str`	Metric name	required

Returns:

Type	Description
`bool`	Given string matches `query`

`prometheus_virtual_metrics.PrometheusResponse`

Attributes:

Name	Type	Description
`response_type`	`PROMETHEUS_RESPONSE_TYPE`	Response type
`request`	`Request`	Prometheus request
`result_count`	`int`	Count of added results/samples

`add_info(message, skip_type_checks=False)`

Add info message

Parameters:

Name	Type	Description	Default
`message`	`list[str] \| str`	Info message	required
`skip_type_checks`	`bool`	Skip type checks	`False`

`add_sample(metric_name, metric_value, timestamp, metric_labels=None, skip_type_checks=False, skip_query_checks=False)`

Add sample. Only available in vector and matrix responses.

When skip_query_checks is not disabled, add_sample will check whether the added sample (metric name + metric labels) matches the PromQL query. If not, the sample is skipped. If you have computational metric values, you can provide a callback for metric_value, which then is only called if the added sample is not skipped.

Parameters:

Name	Type	Description	Default
`metric_name`	`str`	Metric name	required
`metric_value`	`Number \| Callable[None, Number]`	Metric value	required
`timestamp`	`datetime \| float`	Timestamp	required
`metric_labels`	`dict[str, str] \| None`	Metric labels	`None`
`skip_type_checks`	`bool`	Skip type checks	`False`
`skip_query_checks`	`bool`	Skip query checks	`False`

Returns:

Name	Type	Description
`sample_added`	`bool`	Returns `True` if the sample was not skipped

`add_value(value, skip_type_checks=False)`

Add value. Only available in data and series responses.

Parameters:

Name	Type	Description	Default
`value`	`str`	Value	required
`skip_type_checks`	`bool`	Skip type checks	`False`

Raises:

Type	Description
`ValueError`	If response is not a data or series response

`add_warning(message, skip_type_checks=False)`

Add warning message

Parameters:

Name	Type	Description	Default
`message`	`list[str] \| str`	Info message	required
`skip_type_checks`	`bool`	Skip type checks	`False`