How to use opentelemetry-sdk - 10 common examples

def add_event(
        self,
        name: str,
        attributes: types.Attributes = None,
        timestamp: Optional[int] = None,
    ) -> None:
        self.add_lazy_event(
            trace_api.Event(
                name,
                Span.empty_attributes if attributes is None else attributes,
                time_ns() if timestamp is None else timestamp,
            )

self._lock = threading.Lock()

        if attributes is None:
            self.attributes = Span.empty_attributes
        else:
            self.attributes = BoundedDict.from_map(
                MAX_NUM_ATTRIBUTES, attributes
            )

        if events is None:
            self.events = Span.empty_events
        else:
            self.events = BoundedList.from_seq(MAX_NUM_EVENTS, events)

        if links is None:
            self.links = Span.empty_links
        else:
            self.links = BoundedList.from_seq(MAX_NUM_LINKS, links)

        self.end_time = None  # type: Optional[int]
        self.start_time = None  # type: Optional[int]
        self.instrumentation_info = instrumentation_info

self.kind = kind
        self._set_status_on_exception = set_status_on_exception

        self.span_processor = span_processor
        self.status = None
        self._lock = threading.Lock()

        if attributes is None:
            self.attributes = Span.empty_attributes
        else:
            self.attributes = BoundedDict.from_map(
                MAX_NUM_ATTRIBUTES, attributes
            )

        if events is None:
            self.events = Span.empty_events
        else:
            self.events = BoundedList.from_seq(MAX_NUM_EVENTS, events)

        if links is None:
            self.links = Span.empty_links
        else:
            self.links = BoundedList.from_seq(MAX_NUM_LINKS, links)

        self.end_time = None  # type: Optional[int]
        self.start_time = None  # type: Optional[int]
        self.instrumentation_info = instrumentation_info

parent_context,
            context.trace_id,
            context.span_id,
            name,
            attributes,
            links,
        )

        if sampling_decision.sampled:
            if attributes is None:
                span_attributes = sampling_decision.attributes
            else:
                # apply sampling decision attributes after initial attributes
                span_attributes = attributes.copy()
                span_attributes.update(sampling_decision.attributes)
            span = Span(
                name=name,
                context=context,
                parent=parent,
                sampler=self.source.sampler,
                attributes=span_attributes,
                span_processor=self.source._active_span_processor,  # pylint:disable=protected-access
                kind=kind,
                links=links,
                instrumentation_info=self.instrumentation_info,
                set_status_on_exception=set_status_on_exception,
            )
            span.start(start_time=start_time)
        else:
            span = trace_api.DefaultSpan(context=context)
        return span

self.last_update_timestamp = time_ns()
            self.data += value


class GaugeHandle(metrics_api.GaugeHandle, BaseHandle):
    def set(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.GaugeHandle.set`."""
        if self._validate_update(value):
            if self.monotonic and value < self.data:
                logger.warning("Monotonic gauge cannot descend.")
                return
            self.last_update_timestamp = time_ns()
            self.data = value


class MeasureHandle(metrics_api.MeasureHandle, BaseHandle):
    def record(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.MeasureHandle.record`."""
        if self._validate_update(value):
            if self.monotonic and value < 0:
                logger.warning("Monotonic measure cannot accept negatives.")
                return
            self.last_update_timestamp = time_ns()
            # TODO: record


class Metric(metrics_api.Metric):
    """See `opentelemetry.metrics.Metric`."""

    HANDLE_TYPE = BaseHandle

    def __init__(

type(self).__name__, self.data, self.last_update_timestamp
        )


class CounterHandle(metrics_api.CounterHandle, BaseHandle):
    def add(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.CounterHandle.add`."""
        if self._validate_update(value):
            if self.monotonic and value < 0:
                logger.warning("Monotonic counter cannot descend.")
                return
            self.last_update_timestamp = time_ns()
            self.data += value


class GaugeHandle(metrics_api.GaugeHandle, BaseHandle):
    def set(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.GaugeHandle.set`."""
        if self._validate_update(value):
            if self.monotonic and value < self.data:
                logger.warning("Monotonic gauge cannot descend.")
                return
            self.last_update_timestamp = time_ns()
            self.data = value


class MeasureHandle(metrics_api.MeasureHandle, BaseHandle):
    def record(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.MeasureHandle.record`."""
        if self._validate_update(value):
            if self.monotonic and value < 0:
                logger.warning("Monotonic measure cannot accept negatives.")

if not self.enabled:
            return False
        if not isinstance(value, self.value_type):
            logger.warning(
                "Invalid value passed for %s.", self.value_type.__name__
            )
            return False
        return True

    def __repr__(self):
        return '{}(data="{}", last_update_timestamp={})'.format(
            type(self).__name__, self.data, self.last_update_timestamp
        )


class CounterHandle(metrics_api.CounterHandle, BaseHandle):
    def add(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.CounterHandle.add`."""
        if self._validate_update(value):
            if self.monotonic and value < 0:
                logger.warning("Monotonic counter cannot descend.")
                return
            self.last_update_timestamp = time_ns()
            self.data += value


class GaugeHandle(metrics_api.GaugeHandle, BaseHandle):
    def set(self, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.GaugeHandle.set`."""
        if self._validate_update(value):
            if self.monotonic and value < self.data:
                logger.warning("Monotonic gauge cannot descend.")

description,
            unit,
            value_type,
            label_keys=label_keys,
            enabled=enabled,
            monotonic=monotonic,
        )

    def set(self, label_set: LabelSet, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.Gauge.set`."""
        self.get_handle(label_set).set(value)

    UPDATE_FUNCTION = set


class Measure(Metric, metrics_api.Measure):
    """See `opentelemetry.metrics.Measure`.

    By default, measure metrics can accept both positive and negatives.
    Negative inputs will be discarded when monotonic is True.
    """

    HANDLE_TYPE = MeasureHandle

    def __init__(
        self,
        name: str,
        description: str,
        unit: str,
        value_type: Type[metrics_api.ValueT],
        label_keys: Sequence[str] = (),
        enabled: bool = False,

description,
            unit,
            value_type,
            label_keys=label_keys,
            enabled=enabled,
            monotonic=monotonic,
        )

    def add(self, label_set: LabelSet, value: metrics_api.ValueT) -> None:
        """See `opentelemetry.metrics.Counter.add`."""
        self.get_handle(label_set).add(value)

    UPDATE_FUNCTION = add


class Gauge(Metric, metrics_api.Gauge):
    """See `opentelemetry.metrics.Gauge`.

    By default, gauge values can go both up and down (non-monotonic).
    Negative inputs will be discarded for monotonic gauge metrics.
    """

    HANDLE_TYPE = GaugeHandle

    def __init__(
        self,
        name: str,
        description: str,
        unit: str,
        value_type: Type[metrics_api.ValueT],
        label_keys: Sequence[str] = (),
        enabled: bool = True,

if not handle:
            handle = self.HANDLE_TYPE(
                self.value_type, self.enabled, self.monotonic
            )
        self.handles[label_set] = handle
        return handle

    def __repr__(self):
        return '{}(name="{}", description={})'.format(
            type(self).__name__, self.name, self.description
        )

    UPDATE_FUNCTION = lambda x, y: None  # noqa: E731


class Counter(Metric, metrics_api.Counter):
    """See `opentelemetry.metrics.Counter`.

    By default, counter values can only go up (monotonic). Negative inputs
    will be discarded for monotonic counter metrics. Counter metrics that
    have a monotonic option set to False allows negative inputs.
    """

    HANDLE_TYPE = CounterHandle

    def __init__(
        self,
        name: str,
        description: str,
        unit: str,
        value_type: Type[metrics_api.ValueT],
        label_keys: Sequence[str] = (),