satella.instrumentation.metrics.metric_types package

Submodules

satella.instrumentation.metrics.metric_types.base module

class satella.instrumentation.metrics.metric_types.base.EmbeddedSubmetrics(name, root_metric=None, metric_level=None, labels=None, internal=False, *args, **kwargs)

Bases: LeafMetric

A metric that can optionally accept some labels in it’s handle, and this will be counted as a separate metric. For example:

>>> metric = getMetric('root.test.IntValue', 'int', enable_timestamp=False)
>>> metric.handle(2, label='key')
>>> metric.handle(3, label='value')

If you try to inherit from it, refer to simple.IntegerMetric to see how to do it. All please pass all the arguments received from child class into this constructor, as this constructor actually stores them! Refer to cps.ClicksPerTimeUnitMetric on how to do that.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • root_metric (Metric) –

  • metric_level (str) –

  • labels (Optional[dict]) –

  • internal (bool) –

args: List
children_mapping: Dict[Any, Metric]
clone(labels)

Return a fresh instance of this metric, with it’s parent being set to this metric and having a particular set of labels, and being of level INHERIT.

Parameters:

labels (dict) –

Return type:

LeafMetric

embedded_submetrics_enabled: bool
get_specific_metric_data(labels)

Return a MetricDataCollection for a child with given labels

Parameters:

labels (dict) –

Return type:

MetricDataCollection

kwargs: Dict
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.base.LeafMetric(name, root_metric=None, metric_level=None, labels=None, internal=False, *args, **kwargs)

Bases: Metric

A metric capable of generating only leaf entries.

You cannot hook up any children to a leaf metric.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • root_metric (Metric) –

  • metric_level (str) –

  • labels (Optional[dict]) –

  • internal (bool) –

append_child(metric)
Parameters:

metric (Metric) –

labels
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.base.Metric(name, root_metric=None, metric_level=None, internal=False, *args, **kwargs)

Bases: object

Container for child metrics. A base metric class, as well as the default metric.

Switch levels by setting metric.level to a proper value

Parameters:
  • enable_timestamp – append timestamp of last update to the metric

  • internal (bool) – if True, this metric won’t be visible in exporters

  • root_metric (Metric) –

  • metric_level (Optional[Union[MetricLevel, int]]) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'base'
append_child(metric)
Parameters:

metric (Metric) –

can_process_this_level(target_level)
Parameters:

target_level (MetricLevel) –

Return type:

bool

children: List[Metric]
debug(*args, **kwargs)
enable_timestamp
get_fully_qualified_name()
get_timestamp()

Return this timestamp, or None if no timestamp support is enabled

Return type:

Optional[float]

handle(level, *args, **kwargs)
Parameters:

level (Union[int, MetricLevel]) –

Return type:

None

internal
last_updated: Optional[float]
property level: MetricLevel
name
reset()

Delete all child metrics that this metric contains.

Also, if called on root metric, sets the runlevel to RUNTIME

Return type:

None

root_metric
runtime(*args, **kwargs)
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.base.MetricLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

DEBUG = 3
DISABLED = 1
INHERIT = 4
RUNTIME = 2

satella.instrumentation.metrics.metric_types.callable module

class satella.instrumentation.metrics.metric_types.callable.CallableMetric(name, root_metric=None, metric_level=None, labels=None, internal=False, value_getter=None, *args, **kwargs)

Bases: LeafMetric

A metric whose value at any given point in time is the result of it’s callable.

Parameters:
  • value_getter (Optional[Callable[[], float]]) – a callable() that returns a float - the current value of this metric. It should be easy and cheap to compute, as this callable will be called each time a snapshot of metric state is requested

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • labels (Optional[dict]) –

  • internal (bool) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'callable'
callable
labeled_metrics
register_labeled_metric(labeled_metric)
to_metric_data()
Return type:

MetricDataCollection

satella.instrumentation.metrics.metric_types.counter module

class satella.instrumentation.metrics.metric_types.counter.CounterMetric(name, root_metric=None, metric_level=None, internal=False, sum_children=True, count_calls=False, *args, **kwargs)

Bases: EmbeddedSubmetrics, MeasurableMixin

A counter that can be adjusted by a given value.

Parameters:
  • sum_children (bool) – whether to sum up all calls to children

  • count_calls (bool) – count the amount of calls to handle()

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • internal (bool) –

  • args (List) –

  • kwargs (Dict) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'counter'
calls: int
count_calls: bool
sum_children: bool
to_metric_data()
Return type:

MetricDataCollection

value: float

satella.instrumentation.metrics.metric_types.cps module

class satella.instrumentation.metrics.metric_types.cps.ClicksPerTimeUnitMetric(*args, time_unit_vectors=None, aggregate_children=True, internal=False, **kwargs)

Bases: EmbeddedSubmetrics

This tracks the amount of calls to handle() during the last time periods, as specified by time_unit_vectors (in seconds). You may specify multiple time periods as consequent entries in the list.

By default (if you do not specify otherwise) this will track calls made during the last second.

This was once deprecated but out of platforms which suck at calculating derivatives of their series (AWS, I’m looking at you!) this was decided to be undeprecated.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • args (List) –

  • time_unit_vectors (Optional[List[float]]) –

  • aggregate_children (bool) –

  • internal (bool) –

  • kwargs (Dict) –

CLASS_NAME = 'cps'
aggregate_children: bool
count_vectors(last_clicks)
Return type:

MetricDataCollection

cutoff_period: int
last_clicks: tp.List[float]
time_unit_vectors: tp.List[int]
to_metric_data()
Return type:

List[int]

satella.instrumentation.metrics.metric_types.empty module

class satella.instrumentation.metrics.metric_types.empty.EmptyMetric(name, root_metric=None, metric_level=None, internal=False, *args, **kwargs)

Bases: Metric

A metric that disregards all data that it’s fed, and outputs nothing.

A placeholder for the times when you configure metrics and decide to leave some of them out blank.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • root_metric (Metric) –

  • metric_level (Optional[Union[MetricLevel, int]]) –

  • internal (bool) –

CLASS_NAME = 'empty'
to_metric_data()
Return type:

MetricDataCollection

satella.instrumentation.metrics.metric_types.histogram module

class satella.instrumentation.metrics.metric_types.histogram.HistogramMetric(name, root_metric=None, metric_level=None, internal=False, buckets=(0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0), aggregate_children=True, *args, **kwargs)

Bases: EmbeddedSubmetrics, MeasurableMixin

A histogram, by Prometheus’ interpretation.

Parameters:
  • buckets (Sequence[float]) – buckets to add. First bucket will be from zero to first value, second from first value to second, last bucket will be from last value to infinity. So there are len(buckets)+1 buckets. Buckets are expected to be passed in sorted!

  • aggregate_children (bool) – whether to accept child calls to be later presented as total

  • name (str) –

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • internal (bool) –

  • args (List) –

  • kwargs (Dict) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'histogram'
aggregate_children: bool
bucket_limits: tp.List[float]
buckets: tp.List[int]
containers_to_metric_data()
Return type:

MetricDataCollection

count: int
sum: float
to_metric_data()
Return type:

MetricDataCollection

satella.instrumentation.metrics.metric_types.linkfail module

class satella.instrumentation.metrics.metric_types.linkfail.LinkfailMetric(name, root_metric=None, metric_level=None, labels=None, internal=False, consecutive_failures_to_offline=100, consecutive_successes_to_online=10, callback_on_online=<function LinkfailMetric.<lambda>>, callback_on_offline=<function LinkfailMetric.<lambda>>, *args, **kwargs)

Bases: EmbeddedSubmetrics

Metric that measures whether given link is operable.

Parameters:
  • consecutive_failures_to_offline (int) – consecutive failures needed for link to become offline

  • consecutive_successes_to_online (int) – consecutive successes needed for link to become online after a failure

  • callback_on_online (Callable[[int, dict], None]) – callback that accepts an address of a link that becomes online and labels

  • callback_on_offline (Callable[[int, dict], None]) – callback that accepts an address of a link that becomes offline and labels

  • name (str) –

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • labels (Optional[dict]) –

  • internal (bool) –

  • args (List) –

  • kwargs (Dict) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'linkfail'
callback_on_offline: tp.Callable
callback_on_online: tp.Callable
consecutive_failures: tp.Dict[int, int]
consecutive_failures_to_offline: int
consecutive_successes: tp.Dict[int, int]
consecutive_successes_to_online: int
to_metric_data()
Return type:

MetricDataCollection

working: tp.Dict[int, bool]

satella.instrumentation.metrics.metric_types.measurable_mixin module

class satella.instrumentation.metrics.metric_types.measurable_mixin.MeasurableMixin

Bases: object

Add a .measure() method, useful for HistogramMetric and SummaryMetric

measure(include_exceptions=True, logging_level=MetricLevel.RUNTIME, value_getter=<built-in function monotonic>, **labels)

A decorator to measure a difference between some value after the method call and before it.

By default, it will measure the execution time.

Use like:

>>> call_time = getMetric('root.metric_name.execution_time', 'summary')
>>> @call_time.measure()
>>> def measure_my_execution(args):
>>>     ...

If wrapped around p_gen, it will time it from the first element to the last, so beware that it will depend on the speed of the consumer.

It also can be used as a context manager:

>>> with call_time.measure(logging_level=MetricLevel.DEBUG, label='key'):
>>>     ...
Parameters:
  • include_exceptions (bool) – whether to include exceptions

  • logging_level (MetricLevel) – one of RUNTIME or DEBUG

  • value_getter (Callable[[], float]) – a callable that takes no arguments and returns a float, which is the value

  • labels – extra labels to call handle() with

measure_future(future, logging_level=MetricLevel.RUNTIME, value_getter=<built-in function monotonic>, **labels)

A function to measure a difference between some value after the method call and before it.

The value will be taken at the moment this function executes, and the moment the future completes (with or without an exception)

Parameters:
  • future (Future) – future that is considered

  • logging_level (MetricLevel) – one of RUNTIME or DEBUG

  • value_getter (Callable[[], float]) – a callable that takes no arguments and returns a float, which is the value

  • labels – extra labels to call handle() with

satella.instrumentation.metrics.metric_types.registry module

satella.instrumentation.metrics.metric_types.registry.register_metric(cls)

Decorator to register your custom metrics

satella.instrumentation.metrics.metric_types.simple module

class satella.instrumentation.metrics.metric_types.simple.FloatMetric(*args, **kwargs)

Bases: SimpleMetric, MeasurableMixin

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • args (List) –

  • kwargs (Dict) –

CLASS_NAME = 'float'
CONSTRUCTOR

alias of float

DEFAULT_VALUE = 0.0
data: tp.Any
class satella.instrumentation.metrics.metric_types.simple.IntegerMetric(*args, **kwargs)

Bases: SimpleMetric

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • args (List) –

  • kwargs (Dict) –

CLASS_NAME = 'int'
CONSTRUCTOR

alias of int

DEFAULT_VALUE = 0
data: tp.Any
class satella.instrumentation.metrics.metric_types.simple.SimpleMetric(*args, **kwargs)

Bases: EmbeddedSubmetrics

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • args (List) –

  • kwargs (Dict) –

CLASS_NAME = 'string'
CONSTRUCTOR

alias of str

DEFAULT_VALUE = ''
data: tp.Any
to_metric_data()
Return type:

Optional[Union[list, dict, str, int, float]]

satella.instrumentation.metrics.metric_types.summary module

class satella.instrumentation.metrics.metric_types.summary.QuantileMetric(*args, **kwargs)

Bases: SummaryMetric

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • args (List) –

  • kwargs (Dict) –

CLASS_NAME = 'quantile'
aggregate_children: bool
calls_queue: tp.List[float]
count_calls: bool
last_calls: int
quantiles: tp.List[float]
tot_calls: int
tot_time: float
class satella.instrumentation.metrics.metric_types.summary.SummaryMetric(name, root_metric=None, metric_level=None, internal=False, last_calls=100, quantiles=(0.5, 0.95), aggregate_children=True, count_calls=True, *args, **kwargs)

Bases: EmbeddedSubmetrics, MeasurableMixin

A metric that can register some values, sequentially, and then calculate quantiles from it. It calculates configurable quantiles over a sliding window of amount of measurements.

Parameters:
  • last_calls (int) – last calls to handle() to take into account

  • quantiles (Sequence[float]) – a sequence of quantiles to return in to_metric_data

  • aggregate_children (bool) – whether to sum up children values (if present)

  • count_calls (bool) – whether to count total amount of calls and total time

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • internal (bool) –

  • args (List) –

  • kwargs (Dict) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'summary'
aggregate_children: bool
calculate_quantiles(calls_queue)
Return type:

MetricDataCollection

calls_queue: tp.List[float]
count_calls: bool
last_calls: int
quantiles: tp.List[float]
to_metric_data()
Return type:

MetricDataCollection

tot_calls: int
tot_time: float

satella.instrumentation.metrics.metric_types.uptime module

class satella.instrumentation.metrics.metric_types.uptime.UptimeMetric(*args, time_getter=<built-in function monotonic>, **kwargs)

Bases: LeafMetric

A metric that gives the difference between current value of time_getter and it’s value at the initialization of this metric

Parameters:

time_getter (Callable[[], float]) – a callable/0 that returns a float, the notion of the time passing. By default it’s a safe time.monotonic

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'uptime'
basic_time
time_getter
to_metric_data()
Return type:

MetricDataCollection

Module contents

class satella.instrumentation.metrics.metric_types.CallableMetric(name, root_metric=None, metric_level=None, labels=None, internal=False, value_getter=None, *args, **kwargs)

Bases: LeafMetric

A metric whose value at any given point in time is the result of it’s callable.

Parameters:
  • value_getter (Optional[Callable[[], float]]) – a callable() that returns a float - the current value of this metric. It should be easy and cheap to compute, as this callable will be called each time a snapshot of metric state is requested

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • labels (Optional[dict]) –

  • internal (bool) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'callable'
callable
labeled_metrics
register_labeled_metric(labeled_metric)
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.ClicksPerTimeUnitMetric(*args, time_unit_vectors=None, aggregate_children=True, internal=False, **kwargs)

Bases: EmbeddedSubmetrics

This tracks the amount of calls to handle() during the last time periods, as specified by time_unit_vectors (in seconds). You may specify multiple time periods as consequent entries in the list.

By default (if you do not specify otherwise) this will track calls made during the last second.

This was once deprecated but out of platforms which suck at calculating derivatives of their series (AWS, I’m looking at you!) this was decided to be undeprecated.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • time_unit_vectors (Optional[List[float]]) –

  • aggregate_children (bool) –

  • internal (bool) –

CLASS_NAME = 'cps'
aggregate_children: bool
count_vectors(last_clicks)
Return type:

MetricDataCollection

cutoff_period: int
last_clicks: tp.List[float]
time_unit_vectors: tp.List[int]
to_metric_data()
Return type:

List[int]

class satella.instrumentation.metrics.metric_types.CounterMetric(name, root_metric=None, metric_level=None, internal=False, sum_children=True, count_calls=False, *args, **kwargs)

Bases: EmbeddedSubmetrics, MeasurableMixin

A counter that can be adjusted by a given value.

Parameters:
  • sum_children (bool) – whether to sum up all calls to children

  • count_calls (bool) – count the amount of calls to handle()

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • internal (bool) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'counter'
calls: int
count_calls: bool
sum_children: bool
to_metric_data()
Return type:

MetricDataCollection

value: float
class satella.instrumentation.metrics.metric_types.EmbeddedSubmetrics(name, root_metric=None, metric_level=None, labels=None, internal=False, *args, **kwargs)

Bases: LeafMetric

A metric that can optionally accept some labels in it’s handle, and this will be counted as a separate metric. For example:

>>> metric = getMetric('root.test.IntValue', 'int', enable_timestamp=False)
>>> metric.handle(2, label='key')
>>> metric.handle(3, label='value')

If you try to inherit from it, refer to simple.IntegerMetric to see how to do it. All please pass all the arguments received from child class into this constructor, as this constructor actually stores them! Refer to cps.ClicksPerTimeUnitMetric on how to do that.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • root_metric (Metric) –

  • metric_level (str) –

  • labels (Optional[dict]) –

  • internal (bool) –

  • args (List) –

  • kwargs (Dict) –

args: List
children_mapping: Dict[Any, Metric]
clone(labels)

Return a fresh instance of this metric, with it’s parent being set to this metric and having a particular set of labels, and being of level INHERIT.

Parameters:

labels (dict) –

Return type:

LeafMetric

embedded_submetrics_enabled: bool
get_specific_metric_data(labels)

Return a MetricDataCollection for a child with given labels

Parameters:

labels (dict) –

Return type:

MetricDataCollection

kwargs: Dict
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.EmptyMetric(name, root_metric=None, metric_level=None, internal=False, *args, **kwargs)

Bases: Metric

A metric that disregards all data that it’s fed, and outputs nothing.

A placeholder for the times when you configure metrics and decide to leave some of them out blank.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • root_metric (Metric) –

  • metric_level (Optional[Union[MetricLevel, int]]) –

  • internal (bool) –

CLASS_NAME = 'empty'
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.FloatMetric(*args, **kwargs)

Bases: SimpleMetric, MeasurableMixin

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'float'
CONSTRUCTOR

alias of float

DEFAULT_VALUE = 0.0
data: tp.Any
class satella.instrumentation.metrics.metric_types.HistogramMetric(name, root_metric=None, metric_level=None, internal=False, buckets=(0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0), aggregate_children=True, *args, **kwargs)

Bases: EmbeddedSubmetrics, MeasurableMixin

A histogram, by Prometheus’ interpretation.

Parameters:
  • buckets (Sequence[float]) – buckets to add. First bucket will be from zero to first value, second from first value to second, last bucket will be from last value to infinity. So there are len(buckets)+1 buckets. Buckets are expected to be passed in sorted!

  • aggregate_children (bool) – whether to accept child calls to be later presented as total

  • name (str) –

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • internal (bool) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'histogram'
aggregate_children: bool
bucket_limits: tp.List[float]
buckets: tp.List[int]
containers_to_metric_data()
Return type:

MetricDataCollection

count: int
sum: float
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.IntegerMetric(*args, **kwargs)

Bases: SimpleMetric

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'int'
CONSTRUCTOR

alias of int

DEFAULT_VALUE = 0
data: tp.Any
class satella.instrumentation.metrics.metric_types.LeafMetric(name, root_metric=None, metric_level=None, labels=None, internal=False, *args, **kwargs)

Bases: Metric

A metric capable of generating only leaf entries.

You cannot hook up any children to a leaf metric.

When reimplementing the method, remember to pass kwargs here!

Parameters:
  • root_metric (Metric) –

  • metric_level (str) –

  • labels (Optional[dict]) –

  • internal (bool) –

append_child(metric)
Parameters:

metric (Metric) –

labels
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.LinkfailMetric(name, root_metric=None, metric_level=None, labels=None, internal=False, consecutive_failures_to_offline=100, consecutive_successes_to_online=10, callback_on_online=<function LinkfailMetric.<lambda>>, callback_on_offline=<function LinkfailMetric.<lambda>>, *args, **kwargs)

Bases: EmbeddedSubmetrics

Metric that measures whether given link is operable.

Parameters:
  • consecutive_failures_to_offline (int) – consecutive failures needed for link to become offline

  • consecutive_successes_to_online (int) – consecutive successes needed for link to become online after a failure

  • callback_on_online (Callable[[int, dict], None]) – callback that accepts an address of a link that becomes online and labels

  • callback_on_offline (Callable[[int, dict], None]) – callback that accepts an address of a link that becomes offline and labels

  • name (str) –

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • labels (Optional[dict]) –

  • internal (bool) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'linkfail'
callback_on_offline: tp.Callable
callback_on_online: tp.Callable
consecutive_failures: tp.Dict[int, int]
consecutive_failures_to_offline: int
consecutive_successes: tp.Dict[int, int]
consecutive_successes_to_online: int
to_metric_data()
Return type:

MetricDataCollection

working: tp.Dict[int, bool]
class satella.instrumentation.metrics.metric_types.Metric(name, root_metric=None, metric_level=None, internal=False, *args, **kwargs)

Bases: object

Container for child metrics. A base metric class, as well as the default metric.

Switch levels by setting metric.level to a proper value

Parameters:
  • enable_timestamp – append timestamp of last update to the metric

  • internal (bool) – if True, this metric won’t be visible in exporters

  • root_metric (Metric) –

  • metric_level (Optional[Union[MetricLevel, int]]) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'base'
append_child(metric)
Parameters:

metric (Metric) –

can_process_this_level(target_level)
Parameters:

target_level (MetricLevel) –

Return type:

bool

children: List[Metric]
debug(*args, **kwargs)
enable_timestamp
get_fully_qualified_name()
get_timestamp()

Return this timestamp, or None if no timestamp support is enabled

Return type:

Optional[float]

handle(level, *args, **kwargs)
Parameters:

level (Union[int, MetricLevel]) –

Return type:

None

internal
last_updated: Optional[float]
property level: MetricLevel
name
reset()

Delete all child metrics that this metric contains.

Also, if called on root metric, sets the runlevel to RUNTIME

Return type:

None

root_metric
runtime(*args, **kwargs)
to_metric_data()
Return type:

MetricDataCollection

class satella.instrumentation.metrics.metric_types.MetricLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

DEBUG = 3
DISABLED = 1
INHERIT = 4
RUNTIME = 2
class satella.instrumentation.metrics.metric_types.QuantileMetric(*args, **kwargs)

Bases: SummaryMetric

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'quantile'
aggregate_children: bool
calls_queue: tp.List[float]
count_calls: bool
last_calls: int
quantiles: tp.List[float]
tot_calls: int
tot_time: float
class satella.instrumentation.metrics.metric_types.SummaryMetric(name, root_metric=None, metric_level=None, internal=False, last_calls=100, quantiles=(0.5, 0.95), aggregate_children=True, count_calls=True, *args, **kwargs)

Bases: EmbeddedSubmetrics, MeasurableMixin

A metric that can register some values, sequentially, and then calculate quantiles from it. It calculates configurable quantiles over a sliding window of amount of measurements.

Parameters:
  • last_calls (int) – last calls to handle() to take into account

  • quantiles (Sequence[float]) – a sequence of quantiles to return in to_metric_data

  • aggregate_children (bool) – whether to sum up children values (if present)

  • count_calls (bool) – whether to count total amount of calls and total time

  • root_metric (Metric) –

  • metric_level (Optional[MetricLevel]) –

  • internal (bool) –

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'summary'
aggregate_children: bool
calculate_quantiles(calls_queue)
Return type:

MetricDataCollection

calls_queue: tp.List[float]
count_calls: bool
last_calls: int
quantiles: tp.List[float]
to_metric_data()
Return type:

MetricDataCollection

tot_calls: int
tot_time: float
class satella.instrumentation.metrics.metric_types.UptimeMetric(*args, time_getter=<built-in function monotonic>, **kwargs)

Bases: LeafMetric

A metric that gives the difference between current value of time_getter and it’s value at the initialization of this metric

Parameters:

time_getter (Callable[[], float]) – a callable/0 that returns a float, the notion of the time passing. By default it’s a safe time.monotonic

When reimplementing the method, remember to pass kwargs here!

CLASS_NAME = 'uptime'
basic_time
time_getter
to_metric_data()
Return type:

MetricDataCollection

satella.instrumentation.metrics.metric_types.register_metric(cls)

Decorator to register your custom metrics