SolarManager/mysql/opentelemetry/sdk/metrics/_internal/view.py

# Copyright The OpenTelemetry Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from fnmatch import fnmatch
from logging import getLogger
from typing import Optional, Set, Type

from mysql.opentelemetry.metrics import Instrument
from mysql.opentelemetry.sdk.metrics._internal.aggregation import (
    Aggregation,
    DefaultAggregation,
)

# FIXME import from typing when support for 3.7 is removed
from typing_extensions import final

_logger = getLogger(__name__)


class View:
    """
    A `View` configuration parameters can be used for the following
    purposes:

    1. Match instruments: When an instrument matches a view, measurements
       received by that instrument will be processed.
    2. Customize metric streams: A metric stream is identified by a match
       between a view and an instrument and a set of attributes. The metric
       stream can be customized by certain attributes of the corresponding view.

    The attributes documented next serve one of the previous two purposes.

    Args:
        instrument_type: This is an instrument matching attribute: the class the
            instrument must be to match the view.

        instrument_name: This is an instrument matching attribute: the name the
            instrument must have to match the view. Wild card characters are supported. Wild
            card characters should not be used with this attribute if the view has also a
            ``name`` defined.

        meter_name: This is an instrument matching attribute: the name the
            instrument meter must have to match the view.

        meter_version: This is an instrument matching attribute: the version
            the instrument meter must have to match the view.

        meter_schema_url: This is an instrument matching attribute: the schema
            URL the instrument meter must have to match the view.

        name: This is a metric stream customizing attribute: the name of the
            metric stream. If `None`, the name of the instrument will be used.

        description: This is a metric stream customizing attribute: the
            description of the metric stream. If `None`, the description of the instrument will
            be used.

        attribute_keys: This is a metric stream customizing attribute: this is
            a set of attribute keys. If not `None` then only the measurement attributes that
            are in ``attribute_keys`` will be used to identify the metric stream.

        aggregation: This is a metric stream customizing attribute: the
            aggregation instance to use when data is aggregated for the
            corresponding metrics stream. If `None` an instance of
            `DefaultAggregation` will be used.

    This class is not intended to be subclassed by the user.
    """

    _default_aggregation = DefaultAggregation()

    def __init__(
        self,
        instrument_type: Optional[Type[Instrument]] = None,
        instrument_name: Optional[str] = None,
        meter_name: Optional[str] = None,
        meter_version: Optional[str] = None,
        meter_schema_url: Optional[str] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        attribute_keys: Optional[Set[str]] = None,
        aggregation: Optional[Aggregation] = None,
    ):
        if (
            instrument_type
            is instrument_name
            is meter_name
            is meter_version
            is meter_schema_url
            is None
        ):
            raise Exception(
                "Some instrument selection "
                f"criteria must be provided for View {name}"
            )

        if (
            name is not None
            and instrument_name is not None
            and ("*" in instrument_name or "?" in instrument_name)
        ):
            raise Exception(
                f"View {name} declared with wildcard " "characters in instrument_name"
            )

        # _name, _description, _aggregation and _attribute_keys will be
        # accessed when instantiating a _ViewInstrumentMatch.
        self._name = name
        self._instrument_type = instrument_type
        self._instrument_name = instrument_name
        self._meter_name = meter_name
        self._meter_version = meter_version
        self._meter_schema_url = meter_schema_url

        self._description = description
        self._attribute_keys = attribute_keys
        self._aggregation = aggregation or self._default_aggregation

    # pylint: disable=too-many-return-statements
    # pylint: disable=too-many-branches
    @final
    def _match(self, instrument: Instrument) -> bool:
        if self._instrument_type is not None:
            if not isinstance(instrument, self._instrument_type):
                return False

        if self._instrument_name is not None:
            if not fnmatch(instrument.name, self._instrument_name):
                return False

        if self._meter_name is not None:
            if instrument.instrumentation_scope.name != self._meter_name:
                return False

        if self._meter_version is not None:
            if instrument.instrumentation_scope.version != self._meter_version:
                return False

        if self._meter_schema_url is not None:
            if instrument.instrumentation_scope.schema_url != self._meter_schema_url:
                return False

        return True