# 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.
# type: ignore
"""
OpenTelemetry Base Instrumentor
"""
from abc import ABC, abstractmethod
from logging import getLogger
from typing import Collection, Optional
from opentelemetry.instrumentation._semconv import (
_OpenTelemetrySemanticConventionStability,
)
from opentelemetry.instrumentation.dependencies import (
DependencyConflict,
get_dependency_conflicts,
)
_LOG = getLogger(__name__)
[docs]class BaseInstrumentor(ABC):
"""An ABC for instrumentors
Child classes of this ABC should instrument specific third
party libraries or frameworks either by using the
``opentelemetry-instrument`` command or by calling their methods
directly.
Since every third party library or framework is different and has different
instrumentation needs, more methods can be added to the child classes as
needed to provide practical instrumentation to the end user.
"""
_instance = None
_is_instrumented_by_opentelemetry = False
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = object.__new__(cls)
return cls._instance
@property
def is_instrumented_by_opentelemetry(self):
return self._is_instrumented_by_opentelemetry
[docs] @abstractmethod
def instrumentation_dependencies(self) -> Collection[str]:
"""Return a list of python packages with versions that the will be instrumented.
The format should be the same as used in requirements.txt or pyproject.toml.
For example, if an instrumentation instruments requests 1.x, this method should look
like:
def instrumentation_dependencies(self) -> Collection[str]:
return ['requests ~= 1.0']
This will ensure that the instrumentation will only be used when the specified library
is present in the environment.
"""
def _instrument(self, **kwargs):
"""Instrument the library"""
@abstractmethod
def _uninstrument(self, **kwargs):
"""Uninstrument the library"""
def _check_dependency_conflicts(self) -> Optional[DependencyConflict]:
dependencies = self.instrumentation_dependencies()
return get_dependency_conflicts(dependencies)
[docs] def instrument(self, **kwargs):
"""Instrument the library
This method will be called without any optional arguments by the
``opentelemetry-instrument`` command.
This means that calling this method directly without passing any
optional values should do the very same thing that the
``opentelemetry-instrument`` command does.
"""
if self._is_instrumented_by_opentelemetry:
_LOG.warning("Attempting to instrument while already instrumented")
return None
# check if instrumentor has any missing or conflicting dependencies
skip_dep_check = kwargs.pop("skip_dep_check", False)
if not skip_dep_check:
conflict = self._check_dependency_conflicts()
if conflict:
_LOG.error(conflict)
return None
# initialize semantic conventions opt-in if needed
_OpenTelemetrySemanticConventionStability._initialize()
result = self._instrument( # pylint: disable=assignment-from-no-return
**kwargs
)
self._is_instrumented_by_opentelemetry = True
return result
[docs] def uninstrument(self, **kwargs):
"""Uninstrument the library
See ``BaseInstrumentor.instrument`` for more information regarding the
usage of ``kwargs``.
"""
if self._is_instrumented_by_opentelemetry:
result = self._uninstrument(**kwargs)
self._is_instrumented_by_opentelemetry = False
return result
_LOG.warning("Attempting to uninstrument while already uninstrumented")
return None
__all__ = ["BaseInstrumentor"]