1 # Copyright (c) 2022 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
14 """Config executor library."""
16 from importlib import import_module
17 from logging import getLogger
22 Executor class responsible for executing configuration.
26 Config Executor init.=
28 self.metric_registry = dict()
30 def create(self, metrics):
32 Create metrics based on input configuration.
34 :param metrics: Metric list to create.
37 for metric_type, metric_list in metrics.items():
38 for metric in metric_list:
39 module = import_module(
40 name=u"telemetry.metrics", package=metric_type.capitalize()
42 self.metric_registry[metric[u"name"]] = getattr(
43 module, metric_type.capitalize()
46 def serialize(self, metric, labels, item):
48 Serialize metric into destination format.
50 :param metrics: Metric name.
51 :param labels: Metric labels.
52 :param item: Metric dict.
57 if type(self.metric_registry[metric]).__name__ == u"Counter":
58 self.metric_registry[metric].labels(**labels).inc(
61 if type(self.metric_registry[metric]).__name__ == u"Gauge":
62 self.metric_registry[metric].labels(**labels).set(
65 if type(self.metric_registry[metric]).__name__ == u"Info":
66 self.metric_registry[metric].labels(**labels).info(
72 Publish metric into logger.
75 for _, metric_list in self.metric_registry.items():
76 for metric in metric_list.collect():
80 # Adjust from OpenMetrics into Prometheus format.
81 mname = f"{mname}_total" if mtype == u"counter" else mname
82 mname = f"{mname}_info" if mtype == u"info" else mname
83 if mtype in (u"info", u"stateset"):
85 if mtype in (u"gaugehistogram", u"histogram"):
88 mdocumentation = metric.documentation.replace(u"\\", r"\\")
89 mdocumentation = mdocumentation.replace(u"\n", r"\n")
90 output.append(f"# HELP {mname} {mdocumentation}\n")
91 output.append(f"# TYPE {mname} {mtype}\n")
93 for line in metric.samples:
96 for k, value in sorted(line.labels.items()):
98 value = value.replace(u"\\", r"\\")
99 value = value.replace(u"\n", r"\n")
100 value = value.replace(u'"', r'\"')
101 llabel.append(f'{k}="{value}"')
102 labelstr = f"{{{','.join(llabel)}}}"
106 timestamp = f" {int(float(line.timestamp) * 1000):d}" \
107 if line.timestamp else u""
109 f"{line.name}{labelstr} {line.value}{timestamp}\n"
111 getLogger(u"prom").info(u"".join(output))