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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
|
# -*- coding: utf-8 -*-
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""
High-level interface for translating `metrics.yaml` into other formats.
"""
from pathlib import Path
import os
import shutil
import sys
import tempfile
from typing import Any, Callable, Dict, Iterable, List, Optional
from . import lint
from . import parser
from . import javascript
from . import javascript_server
from . import kotlin
from . import markdown
from . import metrics
from . import ruby_server
from . import rust
from . import swift
from . import util
class Outputter:
"""
Class to define an output format.
Each outputter in the table has the following member values:
- output_func: the main function of the outputter, the one which
does the actual translation.
- clear_patterns: A list of glob patterns to clear in the directory before
writing new results to it.
"""
def __init__(
self,
output_func: Callable[[metrics.ObjectTree, Path, Dict[str, Any]], None],
clear_patterns: Optional[List[str]] = None,
):
if clear_patterns is None:
clear_patterns = []
self.output_func = output_func
self.clear_patterns = clear_patterns
OUTPUTTERS = {
"javascript": Outputter(javascript.output_javascript, []),
"typescript": Outputter(javascript.output_typescript, []),
"javascript_server": Outputter(javascript_server.output_javascript, []),
"typescript_server": Outputter(javascript_server.output_typescript, []),
"ruby_server": Outputter(ruby_server.output_ruby, []),
"kotlin": Outputter(kotlin.output_kotlin, ["*.kt"]),
"markdown": Outputter(markdown.output_markdown, []),
"swift": Outputter(swift.output_swift, ["*.swift"]),
"rust": Outputter(rust.output_rust, []),
}
def transform_metrics(objects):
"""
Transform the object model from one that represents the YAML definitions
to one that reflects the type specifics needed by code generators.
e.g. This will transform a `rate` to be a `numerator` if its denominator is
external.
"""
counters = {}
numerators_by_denominator: Dict[str, Any] = {}
for category_name, category_val in objects.items():
if category_name == "tags":
continue
for metric in category_val.values():
fqmn = metric.identifier()
if getattr(metric, "type", None) == "counter":
counters[fqmn] = metric
denominator_name = getattr(metric, "denominator_metric", None)
if denominator_name:
metric.type = "numerator"
numerators_by_denominator.setdefault(denominator_name, [])
numerators_by_denominator[denominator_name].append(metric)
for denominator_name, numerators in numerators_by_denominator.items():
if denominator_name not in counters:
raise ValueError(
f"No `counter` named {denominator_name} found to be used as"
"denominator for {numerators}",
file=sys.stderr,
)
counters[denominator_name].__class__ = metrics.Denominator
counters[denominator_name].type = "denominator"
counters[denominator_name].numerators = numerators
def translate_metrics(
input_filepaths: Iterable[Path],
output_dir: Path,
translation_func: Callable[[metrics.ObjectTree, Path, Dict[str, Any]], None],
clear_patterns: Optional[List[str]] = None,
options: Optional[Dict[str, Any]] = None,
parser_config: Optional[Dict[str, Any]] = None,
):
"""
Translate the files in `input_filepaths` by running the metrics through a
translation function and writing the results in `output_dir`.
:param input_filepaths: list of paths to input metrics.yaml files
:param output_dir: the path to the output directory
:param translation_func: the function that actually performs the translation.
It is passed the following arguments:
- metrics_objects: The tree of metrics as pings as returned by
`parser.parse_objects`.
- output_dir: The path to the output directory.
- options: A dictionary of output format-specific options.
Examples of translation functions are in `kotlin.py` and `swift.py`.
:param clear_patterns: a list of glob patterns of files to clear before
generating the output files. By default, no files will be cleared (i.e.
the directory should be left alone).
:param options: dictionary of options. The available options are backend
format specific. These are passed unchanged to `translation_func`.
:param parser_config: A dictionary of options that change parsing behavior.
See `parser.parse_metrics` for more info.
"""
if clear_patterns is None:
clear_patterns = []
if options is None:
options = {}
if parser_config is None:
parser_config = {}
input_filepaths = util.ensure_list(input_filepaths)
allow_missing_files = parser_config.get("allow_missing_files", False)
if not input_filepaths and not allow_missing_files:
print("❌ No metric files specified. ", end="")
print("Use `--allow-missing-files` to not treat this as an error.")
return 1
if lint.glinter(input_filepaths, parser_config):
return 1
all_objects = parser.parse_objects(input_filepaths, parser_config)
if util.report_validation_errors(all_objects):
return 1
# allow_reserved is also relevant to the translators, so copy it there
if parser_config.get("allow_reserved"):
options["allow_reserved"] = True
# We don't render tags anywhere yet.
all_objects.value.pop("tags", None)
# Apply additional general transformations to all metrics
transform_metrics(all_objects.value)
# Write everything out to a temporary directory, and then move it to the
# real directory, for transactional integrity.
with tempfile.TemporaryDirectory() as tempdir:
tempdir_path = Path(tempdir)
translation_func(all_objects.value, tempdir_path, options)
if output_dir.is_file():
output_dir.unlink()
elif output_dir.is_dir() and len(clear_patterns):
for clear_pattern in clear_patterns:
for filepath in output_dir.glob(clear_pattern):
filepath.unlink()
if len(list(output_dir.iterdir())):
print(f"Extra contents found in '{output_dir}'.")
# We can't use shutil.copytree alone if the directory already exists.
# However, if it doesn't exist, make sure to create one otherwise
# shutil.copy will fail.
os.makedirs(str(output_dir), exist_ok=True)
for filename in tempdir_path.glob("*"):
shutil.copy(str(filename), str(output_dir))
return 0
def translate(
input_filepaths: Iterable[Path],
output_format: str,
output_dir: Path,
options: Optional[Dict[str, Any]] = None,
parser_config: Optional[Dict[str, Any]] = None,
):
"""
Translate the files in `input_filepaths` to the given `output_format` and
put the results in `output_dir`.
:param input_filepaths: list of paths to input metrics.yaml files
:param output_format: the name of the output format
:param output_dir: the path to the output directory
:param options: dictionary of options. The available options are backend
format specific.
:param parser_config: A dictionary of options that change parsing behavior.
See `parser.parse_metrics` for more info.
"""
if options is None:
options = {}
if parser_config is None:
parser_config = {}
format_desc = OUTPUTTERS.get(output_format, None)
if format_desc is None:
raise ValueError(f"Unknown output format '{output_format}'")
return translate_metrics(
input_filepaths,
output_dir,
format_desc.output_func,
format_desc.clear_patterns,
options,
parser_config,
)
|
