"""A Class for metric object."""
from copy import deepcopy
import datetime
try:
import pandas
except ImportError as e:
raise ImportError(
"Pandas is required for Metric class. "
"Please install it with: pip install prometheus-api-client[dataframe] "
"or pip install prometheus-api-client[all]"
) from e
from prometheus_api_client.exceptions import MetricValueConversionError
[docs]
class Metric:
r"""
A Class for `Metric` object.
:param metric: (dict) A metric item from the list of metrics received from prometheus
:param oldest_data_datetime: (datetime|timedelta) Any metric values in the dataframe that are
older than this value will be deleted when new data is added to the dataframe
using the __add__("+") operator.
* `oldest_data_datetime=datetime.timedelta(days=2)`, will delete the
metric data that is 2 days older than the latest metric.
The dataframe is pruned only when new data is added to it.
* `oldest_data_datetime=datetime.datetime(2019,5,23,12,0)`, will delete
any data that is older than "23 May 2019 12:00:00"
* `oldest_data_datetime=datetime.datetime.fromtimestamp(1561475156)`
can also be set using the unix timestamp
Example Usage:
.. code-block:: python
prom = PrometheusConnect()
my_label_config = {'cluster': 'my_cluster_id', 'label_2': 'label_2_value'}
metric_data = prom.get_metric_range_data(metric_name='up', label_config=my_label_config)
# Here metric_data is a list of metrics received from prometheus
# only for the first item in the list
my_metric_object = Metric(metric_data[0], datetime.timedelta(days=10))
"""
def __init__(self, metric, oldest_data_datetime=None):
"""Functions as a Constructor for the Metric object."""
if not isinstance(
oldest_data_datetime, (datetime.datetime, datetime.timedelta, type(None))
):
# if it is neither a datetime object nor a timedelta object raise exception
raise TypeError(
"oldest_data_datetime can only be datetime.datetime/ datetime.timedelta or None"
)
if isinstance(metric, Metric):
# if metric is a Metric object, just copy the object and update its parameters
self.metric_name = metric.metric_name
self.label_config = metric.label_config
self.metric_values = metric.metric_values
self.oldest_data_datetime = oldest_data_datetime
else:
self.metric_name = metric["metric"].get("__name__", None)
self.label_config = deepcopy(metric["metric"])
if "__name__" in self.label_config:
del self.label_config["__name__"]
self.oldest_data_datetime = oldest_data_datetime
# if it is a single value metric change key name
if "value" in metric:
datestamp = metric["value"][0]
metric_value = metric["value"][1]
if isinstance(metric_value, str):
try:
metric_value = float(metric_value)
except (TypeError, ValueError):
raise MetricValueConversionError(
"Converting string metric value to float failed."
)
metric["values"] = [[datestamp, metric_value]]
self.metric_values = pandas.DataFrame(metric["values"], columns=["ds", "y"]).apply(
pandas.to_numeric, errors="raise"
)
self.metric_values["ds"] = pandas.to_datetime(self.metric_values["ds"], unit="s")
# Set the metric start time and the metric end time
self.start_time = self.metric_values.iloc[0, 0]
self.end_time = self.metric_values.iloc[-1, 0]
# We store the plot information as Class variable
Metric._plot = None
[docs]
def __eq__(self, other):
"""
Overloading operator ``=``.
Check whether two metrics are the same (are the same time-series regardless of their data)
Example Usage:
.. code-block:: python
metric_1 = Metric(metric_data_1)
metric_2 = Metric(metric_data_2)
print(metric_1 == metric_2) # will print True if they belong to the same time-series
:return: (bool) If two Metric objects belong to the same time-series,
i.e. same name and label config, it will return True, else False
"""
return bool(
(self.metric_name == other.metric_name) and (self.label_config == other.label_config)
)
[docs]
def __str__(self):
"""
Make it print in a cleaner way when print function is used on a Metric object.
Example Usage:
.. code-block:: python
metric_1 = Metric(metric_data_1)
print(metric_1) # will print the name, labels and the head of the dataframe
"""
name = "metric_name: " + repr(self.metric_name) + "\n"
labels = "label_config: " + repr(self.label_config) + "\n"
values = "metric_values: " + repr(self.metric_values)
return "{" + "\n" + name + labels + values + "\n" + "}"
[docs]
def __add__(self, other):
r"""
Overloading operator ``+``.
Add two metric objects for the same time-series
Example Usage:
.. code-block:: python
metric_1 = Metric(metric_data_1)
metric_2 = Metric(metric_data_2)
metric_12 = metric_1 + metric_2 # will add the data in ``metric_2`` to ``metric_1``
# so if any other parameters are set in ``metric_1``
# will also be set in ``metric_12``
# (like ``oldest_data_datetime``)
:return: (`Metric`) Returns a `Metric` object with the combined metric data
of the two added metrics
:raises: (TypeError) Raises an exception when two metrics being added are
from different metric time-series
"""
if self == other:
new_metric = deepcopy(self)
new_metric.metric_values = pandas.concat([new_metric.metric_values,other.metric_values],ignore_index=True, axis=0)
new_metric.metric_values = new_metric.metric_values.dropna()
new_metric.metric_values = (
new_metric.metric_values.drop_duplicates("ds")
.sort_values(by=["ds"])
.reset_index(drop=True)
)
# if oldest_data_datetime is set, trim the dataframe and only keep the newer data
if new_metric.oldest_data_datetime:
if isinstance(new_metric.oldest_data_datetime, datetime.timedelta):
# create a time range mask
mask = new_metric.metric_values["ds"] >= (
new_metric.metric_values.iloc[-1, 0] - abs(new_metric.oldest_data_datetime)
)
else:
# create a time range mask
mask = new_metric.metric_values["ds"] >= new_metric.oldest_data_datetime
# truncate the df within the mask
new_metric.metric_values = new_metric.metric_values.loc[mask]
# Update the metric start time and the metric end time for the new Metric
new_metric.start_time = new_metric.metric_values.iloc[0, 0]
new_metric.end_time = new_metric.metric_values.iloc[-1, 0]
return new_metric
if self.metric_name != other.metric_name:
error_string = "Different metric names"
else:
error_string = "Different metric labels"
raise TypeError("Cannot Add different metric types. " + error_string)
_metric_plot = None
[docs]
def plot(self, *args, **kwargs):
"""Plot a very simple line graph for the metric time-series."""
if not Metric._metric_plot:
from prometheus_api_client.metric_plot import MetricPlot
Metric._metric_plot = MetricPlot(*args, **kwargs)
metric = self
Metric._metric_plot.plot_date(metric)
[docs]
def show(self, block=None):
"""Plot a very simple line graph for the metric time-series."""
if not Metric._metric_plot:
# can't show before plot
TypeError("Invalid operation: Can't show() before plot()")
Metric._metric_plot.show(block)