-
Notifications
You must be signed in to change notification settings - Fork 421
/
Copy pathmetric.py
116 lines (90 loc) · 3.57 KB
/
metric.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import json
import logging
from contextlib import contextmanager
from typing import Dict
from aws_lambda_powertools.helper.models import MetricUnit
from aws_lambda_powertools.metrics.base import MetricManager
logger = logging.getLogger(__name__)
class SingleMetric(MetricManager):
"""SingleMetric creates an EMF object with a single metric.
EMF specification doesn't allow metrics with different dimensions.
SingleMetric overrides MetricManager's add_metric method to do just that.
Use `single_metric` when you need to create metrics with different dimensions,
otherwise `aws_lambda_powertools.metrics.metrics.Metrics` is
a more cost effective option
Environment variables
---------------------
POWERTOOLS_METRICS_NAMESPACE : str
metric namespace
Example
-------
**Creates cold start metric with function_version as dimension**
from aws_lambda_powertools.metrics import SingleMetric, MetricUnit
import json
metric = Single_Metric(namespace="ServerlessAirline")
metric.add_metric(name="ColdStart", unit=MetricUnit.Count, value=1)
metric.add_dimension(name="function_version", value=47)
print(json.dumps(metric.serialize_metric_set(), indent=4))
Parameters
----------
MetricManager : MetricManager
Inherits from `aws_lambda_powertools.metrics.base.MetricManager`
"""
def add_metric(self, name: str, unit: MetricUnit, value: float):
"""Method to prevent more than one metric being created
Parameters
----------
name : str
Metric name (e.g. BookingConfirmation)
unit : MetricUnit
Metric unit (e.g. "Seconds", MetricUnit.Seconds)
value : float
Metric value
"""
if len(self.metric_set) > 0:
logger.debug(f"Metric {name} already set, skipping...")
return
return super().add_metric(name, unit, value)
@contextmanager
def single_metric(name: str, unit: MetricUnit, value: float, namespace: str = None):
"""Context manager to simplify creation of a single metric
Example
-------
**Creates cold start metric with function_version as dimension**
from aws_lambda_powertools.metrics import single_metric, MetricUnit
with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1, namespace="ServerlessAirline") as metric:
metric.add_dimension(name="function_version", value=47)
**Same as above but set namespace using environment variable**
$ export POWERTOOLS_METRICS_NAMESPACE="ServerlessAirline"
from aws_lambda_powertools.metrics import single_metric, MetricUnit
with single_metric(name="ColdStart", unit=MetricUnit.Count, value=1) as metric:
metric.add_dimension(name="function_version", value=47)
Parameters
----------
name : str
Metric name
unit : MetricUnit
`aws_lambda_powertools.helper.models.MetricUnit`
value : float
Metric value
namespace: str
Namespace for metrics
Yields
-------
SingleMetric
SingleMetric class instance
Raises
------
e
Propagate error received
"""
metric_set = None
try:
metric: SingleMetric = SingleMetric(namespace=namespace)
metric.add_metric(name=name, unit=unit, value=value)
yield metric
logger.debug("Serializing single metric")
metric_set: Dict = metric.serialize_metric_set()
finally:
logger.debug("Publishing single metric", {"metric": metric})
print(json.dumps(metric_set))