How to use opentelemetry-api - 10 common examples

def __init__(self, name: str, default: "object"):
            # pylint: disable=super-init-not-called
            self.name = name
            self.default = base_context.wrap_callable(
                default
            )  # type: typing.Callable[..., object]

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

try:
    from contextvars import ContextVar
except ImportError:
    pass
else:
    import typing  # pylint: disable=unused-import
    from . import base_context

    class AsyncRuntimeContext(base_context.BaseRuntimeContext):
        class Slot(base_context.BaseRuntimeContext.Slot):
            def __init__(self, name: str, default: object):
                # pylint: disable=super-init-not-called
                self.name = name
                self.contextvar = ContextVar(name)  # type: ContextVar[object]
                self.default = base_context.wrap_callable(
                    default
                )  # type: typing.Callable[..., object]

            def clear(self) -> None:
                self.contextvar.set(self.default())

            def get(self) -> object:
                try:
                    return self.contextvar.get()
                except LookupError:

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

try:
    from contextvars import ContextVar
except ImportError:
    pass
else:
    import typing  # pylint: disable=unused-import
    from . import base_context

    class AsyncRuntimeContext(base_context.BaseRuntimeContext):
        class Slot(base_context.BaseRuntimeContext.Slot):
            def __init__(self, name: str, default: object):
                # pylint: disable=super-init-not-called
                self.name = name
                self.contextvar = ContextVar(name)  # type: ContextVar[object]
                self.default = base_context.wrap_callable(
                    default
                )  # type: typing.Callable[..., object]

            def clear(self) -> None:
                self.contextvar.set(self.default())

            def get(self) -> object:
                try:
                    return self.contextvar.get()
                except LookupError:
                    value = self.default()

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

import threading
import typing  # pylint: disable=unused-import

from . import base_context


class ThreadLocalRuntimeContext(base_context.BaseRuntimeContext):
    class Slot(base_context.BaseRuntimeContext.Slot):
        _thread_local = threading.local()

        def __init__(self, name: str, default: "object"):
            # pylint: disable=super-init-not-called
            self.name = name
            self.default = base_context.wrap_callable(
                default
            )  # type: typing.Callable[..., object]

        def clear(self) -> None:
            setattr(self._thread_local, self.name, self.default())

        def get(self) -> "object":
            try:
                got = getattr(self._thread_local, self.name)  # type: object

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

import threading
import typing  # pylint: disable=unused-import

from . import base_context


class ThreadLocalRuntimeContext(base_context.BaseRuntimeContext):
    class Slot(base_context.BaseRuntimeContext.Slot):
        _thread_local = threading.local()

        def __init__(self, name: str, default: "object"):
            # pylint: disable=super-init-not-called
            self.name = name
            self.default = base_context.wrap_callable(
                default
            )  # type: typing.Callable[..., object]

        def clear(self) -> None:
            setattr(self._thread_local, self.name, self.default())

        def get(self) -> "object":
            try:
                got = getattr(self._thread_local, self.name)  # type: object
                return got

def __init__(self, name: str, default: object):
                # pylint: disable=super-init-not-called
                self.name = name
                self.contextvar = ContextVar(name)  # type: ContextVar[object]
                self.default = base_context.wrap_callable(
                    default
                )  # type: typing.Callable[..., object]

"""


class DefaultMetric(Metric):
    """The default Metric used when no Metric implementation is available."""

    def get_handle(self, label_set: LabelSet) -> "DefaultMetricHandle":
        """Gets a `DefaultMetricHandle`.

        Args:
            label_set: `LabelSet` to associate with the returned handle.
        """
        return DefaultMetricHandle()


class Counter(Metric):
    """A counter type metric that expresses the computation of a sum."""

    def get_handle(self, label_set: LabelSet) -> "CounterHandle":
        """Gets a `CounterHandle`."""
        return CounterHandle()

    def add(self, label_set: LabelSet, value: ValueT) -> None:
        """Increases the value of the counter by ``value``.

        Args:
            label_set: `LabelSet` to associate with the returned handle.
            value: The value to add to the counter metric.
        """


class Gauge(Metric):

"""A counter type metric that expresses the computation of a sum."""

    def get_handle(self, label_set: LabelSet) -> "CounterHandle":
        """Gets a `CounterHandle`."""
        return CounterHandle()

    def add(self, label_set: LabelSet, value: ValueT) -> None:
        """Increases the value of the counter by ``value``.

        Args:
            label_set: `LabelSet` to associate with the returned handle.
            value: The value to add to the counter metric.
        """


class Gauge(Metric):
    """A gauge type metric that expresses a pre-calculated value.

    Gauge metrics have a value that is either ``Set`` by explicit
    instrumentation or observed through a callback. This kind of metric
    should be used when the metric cannot be expressed as a sum or because
    the measurement interval is arbitrary.
    """

    def get_handle(self, label_set: LabelSet) -> "GaugeHandle":
        """Gets a `GaugeHandle`."""
        return GaugeHandle()

    def set(self, label_set: LabelSet, value: ValueT) -> None:
        """Sets the value of the gauge to ``value``.

        Args:

def get_handle(self, label_set: LabelSet) -> "object":
        """Gets a handle, used for repeated-use of metrics instruments.

        Handles are useful to reduce the cost of repeatedly recording a metric
        with a pre-defined set of label values. All metric kinds (counter,
        gauge, measure) support declaring a set of required label keys. The
        values corresponding to these keys should be specified in every handle.
        "Unspecified" label values, in cases where a handle is requested but
        a value was not provided are permitted.

        Args:
            label_set: `LabelSet` to associate with the returned handle.
        """


class DefaultMetric(Metric):
    """The default Metric used when no Metric implementation is available."""

    def get_handle(self, label_set: LabelSet) -> "DefaultMetricHandle":
        """Gets a `DefaultMetricHandle`.

        Args:
            label_set: `LabelSet` to associate with the returned handle.
        """
        return DefaultMetricHandle()


class Counter(Metric):
    """A counter type metric that expresses the computation of a sum."""

    def get_handle(self, label_set: LabelSet) -> "CounterHandle":
        """Gets a `CounterHandle`."""

"""

    def get_handle(self, label_set: LabelSet) -> "GaugeHandle":
        """Gets a `GaugeHandle`."""
        return GaugeHandle()

    def set(self, label_set: LabelSet, value: ValueT) -> None:
        """Sets the value of the gauge to ``value``.

        Args:
            label_set: `LabelSet` to associate with the returned handle.
            value: The value to set the gauge metric to.
        """


class Measure(Metric):
    """A measure type metric that represent raw stats that are recorded.

    Measure metrics represent raw statistics that are recorded. By
    default, measure metrics can accept both positive and negatives.
    Negative inputs will be discarded when monotonic is True.
    """

    def get_handle(self, label_set: LabelSet) -> "MeasureHandle":
        """Gets a `MeasureHandle` with a float value."""
        return MeasureHandle()

    def record(self, label_set: LabelSet, value: ValueT) -> None:
        """Records the ``value`` to the measure.

        Args:
            label_set: `LabelSet` to associate with the returned handle.