# -*- 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/.
"""
Code
for parsing metrics.yaml files.
"""
import functools
from pathlib
import Path
import textwrap
from typing
import Any, cast, Dict, Generator, Iterable, Optional, Set, Tuple, Union
import jsonschema
# type: ignore
from jsonschema.exceptions
import ValidationError
# type: ignore
from .metrics
import Metric, ObjectTree
from .pings
import Ping, RESERVED_PING_NAMES
from .tags
import Tag
from .
import util
from .util
import DictWrapper
ROOT_DIR = Path(__file__).parent
SCHEMAS_DIR = ROOT_DIR /
"schemas"
METRICS_ID =
"moz://mozilla.org/schemas/glean/metrics/2-0-0"
PINGS_ID =
"moz://mozilla.org/schemas/glean/pings/2-0-0"
TAGS_ID =
"moz://mozilla.org/schemas/glean/tags/1-0-0"
def _update_validator(validator):
"""
Adds some custom validators to the jsonschema validator that produce
nicer error messages.
"""
def required(validator, required, instance, schema):
if not validator.is_type(instance,
"object"):
return
missing_properties = set(
property
for property
in required
if property
not in instance
)
if len(missing_properties):
missing_properties = sorted(list(missing_properties))
yield ValidationError(
f
"Missing required properties: {', '.join(missing_properties)}"
)
validator.VALIDATORS[
"required"] = required
def _load_file(
filepath: Path, parser_config: Dict[str, Any]
) -> Generator[str,
None, Tuple[Dict[str, util.JSONType], Optional[str]]]:
"""
Load a metrics.yaml
or pings.yaml format file.
If the `filepath` does
not exist, raises `FileNotFoundError`, unless
`parser_config[
"allow_missing_files"]`
is `
True`.
"""
try:
content = util.load_yaml_or_json(filepath)
except FileNotFoundError:
if not parser_config.get(
"allow_missing_files",
False):
raise
else:
return {},
None
except Exception
as e:
yield util.format_error(filepath,
"", textwrap.fill(str(e)))
return {},
None
if content
is None:
yield util.format_error(filepath,
"", f
"'{filepath}' file can not be empty.")
return {},
None
if not isinstance(content, dict):
return {},
None
if content == {}:
return {},
None
schema_key = content.get(
"$schema")
if not isinstance(schema_key, str):
raise TypeError(f
"Invalid schema key {schema_key}")
filetype: Optional[str] =
None
try:
filetype = schema_key.split(
"/")[-2]
except IndexError:
filetype =
None
if filetype
not in (
"metrics",
"pings",
"tags"):
filetype =
None
for error
in validate(content, filepath):
content = {}
yield error
return content, filetype
@functools.lru_cache(maxsize=1)
def _load_schemas() -> Dict[str, Tuple[Any, Any]]:
"""
Load all of the known schemas
from disk,
and put them
in a map based on the
schema
's $id.
"""
schemas = {}
for schema_path
in SCHEMAS_DIR.glob(
"*.yaml"):
schema = util.load_yaml_or_json(schema_path)
resolver = util.get_null_resolver(schema)
validator_class = jsonschema.validators.validator_for(schema)
_update_validator(validator_class)
validator_class.check_schema(schema)
validator = validator_class(schema, resolver=resolver)
schemas[schema[
"$id"]] = (schema, validator)
return schemas
def _get_schema(
schema_id: str, filepath: Union[str, Path] =
"<input>"
) -> Tuple[Any, Any]:
"""
Get the schema
for the given schema $id.
"""
schemas = _load_schemas()
if schema_id
not in schemas:
raise ValueError(
util.format_error(
filepath,
"",
f
"$schema key must be one of {', '.join(schemas.keys())}",
)
)
return schemas[schema_id]
def _get_schema_for_content(
content: Dict[str, util.JSONType], filepath: Union[str, Path]
) -> Tuple[Any, Any]:
"""
Get the appropriate schema
for the given JSON content.
"""
schema_url = content.get(
"$schema")
if not isinstance(schema_url, str):
raise TypeError(
"Invalid $schema type {schema_url}")
return _get_schema(schema_url, filepath)
def validate(
content: Dict[str, util.JSONType], filepath: Union[str, Path] =
"<input>"
) -> Generator[str,
None,
None]:
"""
Validate the given content against the appropriate schema.
"""
try:
schema, validator = _get_schema_for_content(content, filepath)
except ValueError
as e:
yield str(e)
else:
yield from (
util.format_error(filepath,
"", util.pprint_validation_error(e))
for e
in validator.iter_errors(content)
)
def _instantiate_metrics(
all_objects: ObjectTree,
sources: Dict[Any, Path],
content: Dict[str, util.JSONType],
filepath: Path,
config: Dict[str, Any],
) -> Generator[str,
None,
None]:
"""
Load a list of metrics.yaml files, convert the JSON information into Metric
objects,
and merge them into a single tree.
"""
global_no_lint = content.get(
"no_lint", [])
global_tags = content.get(
"$tags", [])
assert isinstance(global_tags, list)
for category_key, category_val
in sorted(content.items()):
if category_key.startswith(
"$"):
continue
if category_key ==
"no_lint":
continue
if not config.get(
"allow_reserved")
and category_key.split(
".")[0] ==
"glean":
yield util.format_error(
filepath,
f
"For category '{category_key}'",
"Categories beginning with 'glean' are reserved for "
"Glean internal use.",
)
continue
all_objects.setdefault(category_key, DictWrapper())
if not isinstance(category_val, dict):
raise TypeError(f
"Invalid content for {category_key}")
for metric_key, metric_val
in sorted(category_val.items()):
try:
metric_obj = Metric.make_metric(
category_key, metric_key, metric_val, validated=
True, config=config
)
except Exception
as e:
yield util.format_error(
filepath,
f
"On instance {category_key}.{metric_key}",
str(e),
metric_val.defined_in[
"line"],
)
metric_obj =
None
else:
if (
not config.get(
"allow_reserved")
and "all-pings" in metric_obj.send_in_pings
):
yield util.format_error(
filepath,
f
"On instance {category_key}.{metric_key}",
'Only internal metrics may specify "all-pings" '
'in "send_in_pings"',
metric_val.defined_in[
"line"],
)
metric_obj =
None
if metric_obj
is not None:
metric_obj.no_lint = sorted(set(metric_obj.no_lint + global_no_lint))
if len(global_tags):
metric_obj.metadata[
"tags"] = sorted(
set(metric_obj.metadata.get(
"tags", []) + global_tags)
)
if isinstance(filepath, Path):
metric_obj.defined_in[
"filepath"] = str(filepath)
already_seen = sources.get((category_key, metric_key))
if already_seen
is not None:
# We've seen this metric name already
yield util.format_error(
filepath,
"",
(
f
"Duplicate metric name '{category_key}.{metric_key}' "
f
"already defined in '{already_seen}'"
),
metric_obj.defined_in[
"line"],
)
else:
all_objects[category_key][metric_key] = metric_obj
sources[(category_key, metric_key)] = filepath
def _instantiate_pings(
all_objects: ObjectTree,
sources: Dict[Any, Path],
content: Dict[str, util.JSONType],
filepath: Path,
config: Dict[str, Any],
) -> Generator[str,
None,
None]:
"""
Load a list of pings.yaml files, convert the JSON information into Ping
objects.
"""
global_no_lint = content.get(
"no_lint", [])
assert isinstance(global_no_lint, list)
ping_schedule_reverse_map: Dict[str, Set[str]] = dict()
for ping_key, ping_val
in sorted(content.items()):
if ping_key.startswith(
"$"):
continue
if ping_key ==
"no_lint":
continue
if not config.get(
"allow_reserved"):
if ping_key
in RESERVED_PING_NAMES:
yield util.format_error(
filepath,
f
"For ping '{ping_key}'",
f
"Ping uses a reserved name ({RESERVED_PING_NAMES})",
)
continue
if not isinstance(ping_val, dict):
raise TypeError(f
"Invalid content for ping {ping_key}")
ping_val[
"name"] = ping_key
if "metadata" in ping_val
and "ping_schedule" in ping_val[
"metadata"]:
if ping_key
in ping_val[
"metadata"][
"ping_schedule"]:
yield util.format_error(
filepath,
f
"For ping '{ping_key}'",
"ping_schedule contains its own ping name",
)
continue
for ping_schedule
in ping_val[
"metadata"][
"ping_schedule"]:
if ping_schedule
not in ping_schedule_reverse_map:
ping_schedule_reverse_map[ping_schedule] = set()
ping_schedule_reverse_map[ping_schedule].add(ping_key)
try:
ping_obj = Ping(
defined_in=getattr(ping_val,
"defined_in",
None),
_validated=
True,
**ping_val,
)
except Exception
as e:
yield util.format_error(filepath, f
"On instance '{ping_key}'", str(e))
continue
if ping_obj
is not None:
ping_obj.no_lint = sorted(set(ping_obj.no_lint + global_no_lint))
if isinstance(filepath, Path)
and ping_obj.defined_in
is not None:
ping_obj.defined_in[
"filepath"] = str(filepath)
already_seen = sources.get(ping_key)
if already_seen
is not None:
# We've seen this ping name already
yield util.format_error(
filepath,
"",
f
"Duplicate ping name '{ping_key}' "
f
"already defined in '{already_seen}'",
)
else:
all_objects.setdefault(
"pings", {})[ping_key] = ping_obj
sources[ping_key] = filepath
for scheduler, scheduled
in ping_schedule_reverse_map.items():
if scheduler
in all_objects[
"pings"]
and isinstance(
all_objects[
"pings"][scheduler], Ping
):
scheduler_obj: Ping = cast(Ping, all_objects[
"pings"][scheduler])
scheduler_obj.schedules_pings = sorted(list(scheduled))
def _instantiate_tags(
all_objects: ObjectTree,
sources: Dict[Any, Path],
content: Dict[str, util.JSONType],
filepath: Path,
config: Dict[str, Any],
) -> Generator[str,
None,
None]:
"""
Load a list of tags.yaml files, convert the JSON information into Tag
objects.
"""
global_no_lint = content.get(
"no_lint", [])
assert isinstance(global_no_lint, list)
for tag_key, tag_val
in sorted(content.items()):
if tag_key.startswith(
"$"):
continue
if tag_key ==
"no_lint":
continue
if not isinstance(tag_val, dict):
raise TypeError(f
"Invalid content for tag {tag_key}")
tag_val[
"name"] = tag_key
try:
tag_obj = Tag(
defined_in=getattr(tag_val,
"defined_in",
None),
_validated=
True,
**tag_val,
)
except Exception
as e:
yield util.format_error(filepath, f
"On instance '{tag_key}'", str(e))
continue
if tag_obj
is not None:
tag_obj.no_lint = sorted(set(tag_obj.no_lint + global_no_lint))
if isinstance(filepath, Path)
and tag_obj.defined_in
is not None:
tag_obj.defined_in[
"filepath"] = str(filepath)
already_seen = sources.get(tag_key)
if already_seen
is not None:
# We've seen this tag name already
yield util.format_error(
filepath,
"",
f
"Duplicate tag name '{tag_key}' "
f
"already defined in '{already_seen}'",
)
else:
all_objects.setdefault(
"tags", {})[tag_key] = tag_obj
sources[tag_key] = filepath
def _preprocess_objects(objs: ObjectTree, config: Dict[str, Any]) -> ObjectTree:
"""
Preprocess the object tree to better set defaults.
"""
for category
in objs.values():
for obj
in category.values():
if not isinstance(obj, Metric):
continue
if not config.get(
"do_not_disable_expired",
False)
and hasattr(
obj,
"is_disabled"
):
obj.disabled = obj.is_disabled()
if hasattr(obj,
"send_in_pings"):
if "default" in obj.send_in_pings:
obj.send_in_pings = obj.default_store_names + [
x
for x
in obj.send_in_pings
if x !=
"default"
]
obj.send_in_pings = sorted(list(set(obj.send_in_pings)))
return objs
@util.keep_value
def parse_objects(
filepaths: Iterable[Path], config: Optional[Dict[str, Any]] =
None
) -> Generator[str,
None, ObjectTree]:
"""
Parse one
or more metrics.yaml
and/
or pings.yaml files, returning a tree of
`metrics.Metric`, `pings.Ping`,
and `tags.Tag` instances.
The result
is a generator over any errors.
If there are no errors, the
actual metrics can be obtained
from `result.value`.
For example::
result = metrics.parse_metrics(filepaths)
for err
in result:
print(err)
all_metrics = result.value
The result value
is a dictionary of category names to categories, where
each category
is a dictionary
from metric name to `metrics.Metric`
instances. There are also the special categories `pings`
and `tags`
containing all of the `pings.Ping`
and `tags.Tag` instances, respectively.
:param filepaths: list of Path objects to metrics.yaml, pings.yaml,
and/
or
tags.yaml files
:param config: A dictionary of options that change parsing behavior.
Supported keys are:
- `allow_reserved`: Allow values reserved
for internal Glean use.
- `do_not_disable_expired`: Don
't mark expired metrics as disabled.
This
is useful when you want to retain the original
"disabled"
value
from the `metrics.yaml`, rather than having it overridden when
the metric expires.
- `allow_missing_files`: Do
not raise a `FileNotFoundError`
if any of
the input `filepaths` do
not exist.
- `interesting`: Contains an array of interesting metrics/ping files.
Probes
not included
in these files will be marked
as disabled.
"""
if config
is None:
config = {}
all_objects: ObjectTree = DictWrapper()
sources: Dict[Any, Path] = {}
filepaths = util.ensure_list(filepaths)
for filepath
in filepaths:
content, filetype =
yield from _load_file(filepath, config)
if filetype ==
"metrics":
yield from _instantiate_metrics(
all_objects, sources, content, filepath, config
)
elif filetype ==
"pings":
yield from _instantiate_pings(
all_objects, sources, content, filepath, config
)
elif filetype ==
"tags":
yield from _instantiate_tags(
all_objects, sources, content, filepath, config
)
if config.get(
"interesting"):
# We're configured to disable probes not included in the interesting list.
filepaths = util.ensure_list(config.get(
"interesting"))
interesting_metrics_dict: Dict[str, Dict[str, Any]] = dict()
interesting_metrics_dict.setdefault(
"metrics", DictWrapper())
interesting_metrics_dict.setdefault(
"pings", DictWrapper())
for filepath
in filepaths:
content, filetype =
yield from _load_file(filepath, config)
if not isinstance(content, dict):
raise TypeError(f
"Invalid content for {filepath}")
for category_key, category_val
in sorted(content.items()):
if category_key.startswith(
"$"):
continue
interesting_metrics_dict.setdefault(category_key, DictWrapper())
if not isinstance(category_val, dict):
raise TypeError(f
"Invalid category_val for {category_key}")
for metric_key, metric_val
in sorted(category_val.items()):
interesting_metrics_dict[category_key][metric_key] = metric_val
for category_key, category_val
in all_objects.items():
if category_key ==
"tags":
continue
for metric_key, metric_val
in sorted(category_val.items()):
category_dict = interesting_metrics_dict.get(category_key, {})
if metric_key
not in category_dict:
obj = all_objects[category_key][metric_key]
if hasattr(obj,
"disabled"):
obj.disabled =
True
return _preprocess_objects(all_objects, config)
