Quelle metrics.py Sprache: Python

# -*- 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/.

"""
Classes for each of the high-level metric types.
"""

import enum
from typing import Any, Dict, List, Optional, Type, Union  # noqa

from . import pings
from . import tags
from . import util

# Important: if the values are ever changing here, make sure
# to also fix mozilla/glean. Otherwise language bindings may
# break there.
class Lifetime(enum.Enum):
    ping = 0
    application = 1
    user = 2

class DataSensitivity(enum.Enum):
    technical = 1
    interaction = 2
    stored_content = 3
    web_activity = 3  # Old, deprecated name
    highly_sensitive = 4

class Metric:
    typename: str = "ERROR"
    glean_internal_metric_cat: str = "glean.internal.metrics"
    metric_types: Dict[str, Any] = {}
    default_store_names: List[str] = ["metrics"]

    def __init__(
        self,
        type: str,
        category: str,
        name: str,
        bugs: List[str],
        description: str,
        notification_emails: List[str],
        expires: Any,
        metadata: Optional[Dict] = None,
        data_reviews: Optional[List[str]] = None,
        version: int = 0,
        disabled: bool = False,
        lifetime: str = "ping",
        send_in_pings: Optional[List[str]] = None,
        unit: Optional[str] = None,
        gecko_datapoint: str = "",
        no_lint: Optional[List[str]] = None,
        data_sensitivity: Optional[List[str]] = None,
        defined_in: Optional[Dict] = None,
        telemetry_mirror: Optional[str] = None,
        _config: Optional[Dict[str, Any]] = None,
        _validated: bool = False,
    ):
        # Avoid cyclical import
        from . import parser

        self.type = type
        self.category = category
        self.name = name
        self.bugs = bugs
        self.description = description
        self.notification_emails = notification_emails
        self.expires = expires
        if metadata is None:
            metadata = {}
        self.metadata = metadata
        if data_reviews is None:
            data_reviews = []
        self.data_reviews = data_reviews
        self.version = version
        self.disabled = disabled
        self.lifetime = getattr(Lifetime, lifetime)
        if send_in_pings is None:
            send_in_pings = ["default"]
        self.send_in_pings = send_in_pings
        self.unit = unit
        self.gecko_datapoint = gecko_datapoint
        if no_lint is None:
            no_lint = []
        self.no_lint = no_lint
        if data_sensitivity is not None:
            self.data_sensitivity = [
                getattr(DataSensitivity, x) for x in data_sensitivity
            ]
        self.defined_in = defined_in
        if telemetry_mirror is not None:
            self.telemetry_mirror = telemetry_mirror

        # _validated indicates whether this metric has already been jsonschema
        # validated (but not any of the Python-level validation).
        if not _validated:
            data = {
                "$schema": parser.METRICS_ID,
                self.category: {self.name: self._serialize_input()},
            }  # type: Dict[str, util.JSONType]
            for error in parser.validate(data):
                raise ValueError(error)

        # Store the config, but only after validation.
        if _config is None:
            _config = {}
        self._config = _config

        # Metrics in the special category "glean.internal.metrics" need to have
        # an empty category string when identifying the metrics in the ping.
        if self.category == Metric.glean_internal_metric_cat:
            self.category = ""

    def __init_subclass__(cls, **kwargs):
        # Create a mapping of all of the subclasses of this class
        if cls not in Metric.metric_types and hasattr(cls, "typename"):
            Metric.metric_types[cls.typename] = cls
        super().__init_subclass__(**kwargs)

    @classmethod
    def make_metric(
        cls,
        category: str,
        name: str,
        metric_info: Dict[str, util.JSONType],
        config: Optional[Dict[str, Any]] = None,
        validated: bool = False,
    ):
        """
        Given a metric_info dictionary from metrics.yaml, return a metric
        instance.

        :param: category The category the metric lives in
        :param: name The name of the metric
        :param: metric_info A dictionary of the remaining metric parameters
        :param: config A dictionary containing commandline configuration
            parameters
        :param: validated True if the metric has already gone through
            jsonschema validation
        :return: A new Metric instance.
        """
        if config is None:
            config = {}

        metric_type = metric_info["type"]
        if not isinstance(metric_type, str):
            raise TypeError(f"Unknown metric type {metric_type}")
        return cls.metric_types[metric_type](
            category=category,
            name=name,
            defined_in=getattr(metric_info, "defined_in", None),
            _validated=validated,
            _config=config,
            **metric_info,
        )

    def serialize(self) -> Dict[str, util.JSONType]:
        """
        Serialize the metric back to JSON object model.
        """
        d = self.__dict__.copy()
        # Convert enum fields back to strings
        for key, val in d.items():
            if isinstance(val, enum.Enum):
                d[key] = d[key].name
            if isinstance(val, set):
                d[key] = sorted(list(val))
            if isinstance(val, list) and len(val) and isinstance(val[0], enum.Enum):
                d[key] = [x.name for x in val]
        del d["name"]
        del d["category"]
        if not d["unit"]:
            d.pop("unit")
        d.pop("_config", None)
        d.pop("_generate_enums", None)
        d.pop("_generate_structure", None)
        return d

    def _serialize_input(self) -> Dict[str, util.JSONType]:
        d = self.serialize()
        modified_dict = util.remove_output_params(d, "defined_in")
        return modified_dict

    def identifier(self) -> str:
        """
        Create an identifier unique for this metric.
        Generally, category.name; however, Glean internal
        metrics only use name.
        """
        if not self.category:
            return self.name
        return ".".join((self.category, self.name))

    def is_disabled(self) -> bool:
        return self.disabled or self.is_expired()

    def is_expired(self) -> bool:
        def default_handler(expires) -> bool:
            return util.is_expired(expires, self._config.get("expire_by_version"))

        return self._config.get("custom_is_expired", default_handler)(self.expires)

    def validate_expires(self):
        def default_handler(expires):
            return util.validate_expires(expires, self._config.get("expire_by_version"))

        return self._config.get("custom_validate_expires", default_handler)(
            self.expires
        )

    def is_internal_metric(self) -> bool:
        return self.category in (Metric.glean_internal_metric_cat, "")

class Boolean(Metric):
    typename = "boolean"

class String(Metric):
    typename = "string"

class StringList(Metric):
    typename = "string_list"

class Counter(Metric):
    typename = "counter"

class Quantity(Metric):
    typename = "quantity"

class TimeUnit(enum.Enum):
    nanosecond = 0
    microsecond = 1
    millisecond = 2
    second = 3
    minute = 4
    hour = 5
    day = 6

class TimeBase(Metric):
    def __init__(self, *args, **kwargs):
        self.time_unit = getattr(TimeUnit, kwargs.pop("time_unit", "millisecond"))
        super().__init__(*args, **kwargs)

class Timespan(TimeBase):
    typename = "timespan"

class TimingDistribution(TimeBase):
    typename = "timing_distribution"

    def __init__(self, *args, **kwargs):
        self.time_unit = getattr(TimeUnit, kwargs.pop("time_unit", "nanosecond"))
        Metric.__init__(self, *args, **kwargs)

class MemoryUnit(enum.Enum):
    byte = 0
    kilobyte = 1
    megabyte = 2
    gigabyte = 3

class MemoryDistribution(Metric):
    typename = "memory_distribution"

    def __init__(self, *args, **kwargs):
        self.memory_unit = getattr(MemoryUnit, kwargs.pop("memory_unit", "byte"))
        super().__init__(*args, **kwargs)

class HistogramType(enum.Enum):
    linear = 0
    exponential = 1

class CustomDistribution(Metric):
    typename = "custom_distribution"

    def __init__(self, *args, **kwargs):
        self.range_min = kwargs.pop("range_min", 1)
        self.range_max = kwargs.pop("range_max")
        self.bucket_count = kwargs.pop("bucket_count")
        self.histogram_type = getattr(
            HistogramType, kwargs.pop("histogram_type", "exponential")
        )
        super().__init__(*args, **kwargs)

class Datetime(TimeBase):
    typename = "datetime"

class Event(Metric):
    typename = "event"

    default_store_names = ["events"]

    def __init__(self, *args, **kwargs):
        self.extra_keys = kwargs.pop("extra_keys", {})
        self.validate_extra_keys(self.extra_keys, kwargs.get("_config", {}))
        super().__init__(*args, **kwargs)
        self._generate_enums = [("allowed_extra_keys_with_types", "Extra")]

    @property
    def allowed_extra_keys(self):
        # Sort keys so that output is deterministic
        return sorted(list(self.extra_keys.keys()))

    @property
    def allowed_extra_keys_with_types(self):
        # Sort keys so that output is deterministic
        return sorted(
            [(k, v.get("type", "string")) for (k, v) in self.extra_keys.items()],
            key=lambda x: x[0],
        )

    @staticmethod
    def validate_extra_keys(extra_keys: Dict[str, str], config: Dict[str, Any]) -> None:
        if not config.get("allow_reserved") and any(
            k.startswith("glean.") for k in extra_keys.keys()
        ):
            raise ValueError(
                "Extra keys beginning with 'glean.' are reserved for "
                "Glean internal use."
            )

class Uuid(Metric):
    typename = "uuid"

class Url(Metric):
    typename = "url"

class Jwe(Metric):
    typename = "jwe"

    def __init__(self, *args, **kwargs):
        raise ValueError(
            "JWE support was removed. "
            "If you require this send an email to glean-team@mozilla.com."
        )

class CowString(str):
    """
    Wrapper class for strings that should be represented
    as a `Cow<'static, str>` in Rust,
    or `String` in other target languages.

    This wraps `str`, so unless `CowString` is specifically
    handled it acts (and serializes)
    as a string.
    """

    def __init__(self, val: str):
        self.inner: str = val

    def __eq__(self, other):
        return self.inner == other.inner

    def __hash__(self):
        return self.inner.__hash__()

    def __lt__(self, other):
        return self.inner.__lt__(other.inner)

class Labeled(Metric):
    labeled = True

    def __init__(self, *args, **kwargs):
        labels = kwargs.pop("labels", None)
        if labels is not None:
            self.ordered_labels = labels
            self.labels = set([CowString(label) for label in labels])
        else:
            self.ordered_labels = None
            self.labels = None
        super().__init__(*args, **kwargs)

    def serialize(self) -> Dict[str, util.JSONType]:
        """
        Serialize the metric back to JSON object model.
        """
        d = super().serialize()
        d["labels"] = self.ordered_labels
        del d["ordered_labels"]
        return d

class LabeledBoolean(Labeled, Boolean):
    typename = "labeled_boolean"

class LabeledString(Labeled, String):
    typename = "labeled_string"

class LabeledCounter(Labeled, Counter):
    typename = "labeled_counter"

class LabeledCustomDistribution(Labeled, CustomDistribution):
    typename = "labeled_custom_distribution"

class LabeledMemoryDistribution(Labeled, MemoryDistribution):
    typename = "labeled_memory_distribution"

class LabeledTimingDistribution(Labeled, TimingDistribution):
    typename = "labeled_timing_distribution"

class LabeledQuantity(Labeled, Quantity):
    typename = "labeled_quantity"

class Rate(Metric):
    typename = "rate"

    def __init__(self, *args, **kwargs):
        self.denominator_metric = kwargs.pop("denominator_metric", None)
        super().__init__(*args, **kwargs)

class Denominator(Counter):
    typename = "denominator"
    # A denominator is a counter with an additional list of numerators.
    numerators: List[Rate] = []

class Text(Metric):
    typename = "text"

class Object(Metric):
    typename = "object"

    def __init__(self, *args, **kwargs):
        structure = kwargs.pop("structure", None)
        if not structure:
            raise ValueError("`object` is missing required parameter `structure`")

        self._generate_structure = self.validate_structure(structure)
        super().__init__(*args, **kwargs)

    ALLOWED_TOPLEVEL = {"type", "properties", "items"}
    ALLOWED_TYPES = ["object", "array", "number", "string", "boolean"]

    @staticmethod
    def _validate_substructure(structure):
        extra = set(structure.keys()) - Object.ALLOWED_TOPLEVEL
        if extra:
            extra = ", ".join(extra)
            allowed = ", ".join(Object.ALLOWED_TOPLEVEL)
            raise ValueError(
                f"Found additional fields: {extra}. Only allowed: {allowed}"
            )

        if "type" not in structure:
            raise ValueError(
                f"missing `type` in object structure. Allowed: {Object.ALLOWED_TYPES}"
            )
        if structure["type"] not in Object.ALLOWED_TYPES:
            raise ValueError(
                "invalid `type` in object structure. found: {}, allowed: {}".format(
                    structure["type"], Object.ALLOWED_TYPES
                )
            )

        if structure["type"] == "object":
            if "items" in structure:
                raise ValueError("`items` not allowed in object structure")

            if "properties" not in structure:
                raise ValueError("`properties` missing for type `object`")

            for key in structure["properties"]:
                value = structure["properties"][key]
                structure["properties"][key] = Object._validate_substructure(value)

        if structure["type"] == "array":
            if "properties" in structure:
                raise ValueError("`properties` not allowed in array structure")

            if "items" not in structure:
                raise ValueError("`items` missing for type `array`")

            value = structure["items"]
            structure["items"] = Object._validate_substructure(value)

        return structure

    @staticmethod
    def validate_structure(structure):
        if None:
            raise ValueError("`structure` needed for object metric.")

        # Different from `ALLOWED_TYPES`:
        # We _require_ the root type to be an object or array.
        allowed_types = ["object", "array"]
        if "type" not in structure:
            raise ValueError(
                f"missing `type` in object structure. Allowed: {allowed_types}"
            )
        if structure["type"] not in allowed_types:
            raise ValueError(
                "invalid `type` in object structure. found: {}, allowed: {}".format(
                    structure["type"], allowed_types
                )
            )

        structure = Object._validate_substructure(structure)
        return structure

ObjectTree = Dict[str, Dict[str, Union[Metric, pings.Ping, tags.Tag]]]

quality92%

¤ Dauer der Verarbeitung: 0.16 Sekunden (vorverarbeitet) ¤

Wurzel

Suchen

Beweissystem der NASA

Beweissystem Isabelle

NIST Cobol Testsuite

Cephes Mathematical Library

Wiener Entwicklungsmethode

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung ist noch experimentell.