Skip to content

Commit 6acb1c5

Browse files
oxeye-nikolayoxeye-nikolay
committed
Added initial code
1 parent c279ee5 commit 6acb1c5

File tree

8 files changed

+330
-0
lines changed

8 files changed

+330
-0
lines changed

‎instrumentation/opentelemetry-instrumentation-pika/README.rst

+23
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
OpenTelemetry <REPLACE ME> Instrumentation
2+
===========================
3+
4+
|pypi|
5+
6+
.. |pypi| image:: https://badge.fury.io/py/opentelemetry-instrumentation-pika.svg
7+
:target: https://pypi.org/project/opentelemetry-instrumentation-pika/
8+
9+
This library allows tracing requests made by the <REPLACE ME> library.
10+
11+
Installation
12+
------------
13+
14+
::
15+
16+
pip install opentelemetry-instrumentation-pika
17+
18+
19+
References
20+
----------
21+
22+
* `OpenTelemetry pika/ Tracing <https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/pika/pika.html>`_
23+
* `OpenTelemetry Project <https://opentelemetry.io/>`_

‎instrumentation/opentelemetry-instrumentation-pika/setup.cfg

+50
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
# Copyright The OpenTelemetry Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
#
15+
[metadata]
16+
name = opentelemetry-instrumentation-pika
17+
description = OpenTelemetry pika instrumentation
18+
long_description = file: README.rst
19+
long_description_content_type = text/x-rst
20+
author = OpenTelemetry Authors
21+
author_email = [email protected]
22+
url = https://github.com/open-telemetry/opentelemetry-python-contrib/instrumentation/opentelemetry-instrumentation-pika
23+
platforms = any
24+
license = Apache-2.0
25+
classifiers =
26+
Development Status :: 4 - Beta
27+
Intended Audience :: Developers
28+
License :: OSI Approved :: Apache Software License
29+
Programming Language :: Python
30+
Programming Language :: Python :: 3
31+
Programming Language :: Python :: 3.6
32+
Programming Language :: Python :: 3.7
33+
Programming Language :: Python :: 3.8
34+
35+
[options]
36+
python_requires = >=3.6
37+
package_dir=
38+
=src
39+
packages=find_namespace:
40+
41+
install_requires =
42+
opentelemetry-api ~= 1.5
43+
pika >= 1.1.0
44+
45+
[options.extras_require]
46+
test =
47+
48+
[options.packages.find]
49+
where = src
50+

‎instrumentation/opentelemetry-instrumentation-pika/setup.py

+38
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
# Copyright The OpenTelemetry Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
import os
15+
16+
import setuptools
17+
18+
BASE_DIR = os.path.dirname(__file__)
19+
VERSION_FILENAME = os.path.join(
20+
BASE_DIR,
21+
"src",
22+
"opentelemetry",
23+
"instrumentation",
24+
"pika",
25+
"version.py",
26+
)
27+
PACKAGE_INFO = {}
28+
with open(VERSION_FILENAME) as f:
29+
exec(f.read(), PACKAGE_INFO)
30+
31+
setuptools.setup(
32+
version=PACKAGE_INFO["__version__"],
33+
entry_points={
34+
"opentelemetry_instrumentor": [
35+
"pika = opentelemetry.instrumentation.pika:PikaInstrumentor"
36+
]
37+
},
38+
)

‎instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/__init__.py

+2
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
from .version import __version__
2+
from .pika_instrumentor import PikaInstrumentation

‎instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/package.py

+16
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
# Copyright The OpenTelemetry Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
16+
_instruments = ("pika >= 1.1.0",)

‎instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/pika_instrumentor.py

+108
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
import pika
2+
from logging import getLogger
3+
from opentelemetry import trace
4+
from typing import Dict, Callable
5+
from typing import Collection, Any
6+
from pika.adapters import BaseConnection
7+
from opentelemetry.propagate import inject
8+
from opentelemetry.instrumentation.pika import utils
9+
from opentelemetry.trace import Tracer, TracerProvider
10+
from opentelemetry.semconv.trace import MessagingOperationValues
11+
from opentelemetry.instrumentation.pika.package import _instruments
12+
from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
13+
14+
15+
_LOG = getLogger(__name__)
16+
CTX_KEY = "__otel_task_span"
17+
18+
19+
class PikaInstrumentation(BaseInstrumentor):
20+
@staticmethod
21+
def _instrument_consumers(
22+
consumers_dict: Dict[str, Callable[..., Any]], tracer: Tracer
23+
) -> Any:
24+
for key, callback in consumers_dict.items():
25+
26+
def decorated_callback(
27+
channel: pika.channel.Channel,
28+
method: pika.spec.Basic.Deliver,
29+
properties: pika.spec.BasicProperties,
30+
body: bytes,
31+
) -> Any:
32+
if not properties:
33+
properties = pika.spec.BasicProperties()
34+
span = utils.get_span(
35+
tracer,
36+
channel,
37+
properties,
38+
task_name=key,
39+
operation=MessagingOperationValues.RECEIVE,
40+
)
41+
with trace.use_span(span, end_on_exit=True):
42+
inject(properties.headers)
43+
retval = callback(channel, method, properties, body)
44+
return retval
45+
46+
decorated_callback.__setattr__("_original_callback", callback)
47+
consumers_dict[key] = decorated_callback
48+
49+
@staticmethod
50+
def _instrument_publish(channel: Any, tracer: Tracer) -> None:
51+
original_basic_publish = channel.basic_publish
52+
53+
def decorated_basic_publish(
54+
exchange, routing_key, body, properties=None, mandatory=False
55+
):
56+
if not properties:
57+
properties = pika.spec.BasicProperties()
58+
span = utils.get_span(
59+
tracer,
60+
channel,
61+
properties,
62+
task_name="(temporary)",
63+
operation=None,
64+
)
65+
with trace.use_span(span, end_on_exit=True):
66+
inject(properties.headers)
67+
retval = original_basic_publish(
68+
exchange, routing_key, body, properties, mandatory
69+
)
70+
return retval
71+
72+
decorated_basic_publish.__setattr__(
73+
"_original_function", original_basic_publish
74+
)
75+
channel.basic_publish = decorated_basic_publish
76+
77+
@staticmethod
78+
def instrument_channel(
79+
channel: Any, tracer_provider: TracerProvider
80+
) -> None:
81+
if not hasattr(channel, "_impl") or not isinstance(
82+
channel._impl, pika.channel.Channel
83+
):
84+
_LOG.error("Could not find implementation for provided channel!")
85+
return
86+
tracer = trace.get_tracer(__name__, pika.__version__, tracer_provider)
87+
if channel._impl._consumers:
88+
PikaInstrumentation._instrument_consumers(
89+
channel._impl._consumers, tracer
90+
)
91+
PikaInstrumentation._instrument_publish(channel, tracer)
92+
93+
def _uninstrument(self, connection: Any, **kwargs: Dict[str, Any]) -> None:
94+
if not hasattr(connection, "_impl") or not isinstance(
95+
connection._impl, BaseConnection
96+
):
97+
_LOG.error("Could not find implementation for provided channel!")
98+
return
99+
for key, callback in connection._impl._consumers:
100+
if hasattr(callback, "_original_callback"):
101+
connection._consumers[key] = callback._original_callback
102+
if hasattr(connection.basic_publish, "_original_function"):
103+
connection.basic_publish = (
104+
connection.basic_publish._original_function
105+
)
106+
107+
def instrumentation_dependencies(self) -> Collection[str]:
108+
return _instruments

‎instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/utils.py

+78
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
from typing import Optional
2+
from pika.channel import Channel
3+
from pika.spec import BasicProperties
4+
from opentelemetry.trace import Tracer
5+
from opentelemetry.trace.span import Span
6+
from opentelemetry.propagate import extract
7+
from opentelemetry.propagators.textmap import Getter
8+
from opentelemetry.semconv.trace import (
9+
SpanAttributes,
10+
MessagingOperationValues,
11+
)
12+
13+
14+
class PikaGetter(Getter):
15+
def get(self, carrier, key):
16+
value = carrier.get(key, None)
17+
if value is None:
18+
return None
19+
return (value,)
20+
21+
def keys(self, carrier):
22+
return []
23+
24+
25+
pika_getter = PikaGetter()
26+
27+
28+
def get_span(
29+
tracer: Tracer,
30+
channel: Channel,
31+
properties: BasicProperties,
32+
task_name: str,
33+
operation: Optional[MessagingOperationValues],
34+
) -> Span:
35+
if properties.headers is None:
36+
properties.headers = {}
37+
ctx = extract(properties.headers, getter=pika_getter)
38+
span = tracer.start_span(
39+
context=ctx, name=generate_span_name(task_name, operation)
40+
)
41+
enrich_span(span, channel, properties, task_name, operation)
42+
return span
43+
44+
45+
def generate_span_name(
46+
task_name: str, operation: MessagingOperationValues
47+
) -> str:
48+
return f"{task_name} {operation.value}"
49+
50+
51+
def enrich_span(
52+
span: Span,
53+
channel: Channel,
54+
properties: BasicProperties,
55+
task_destination: str,
56+
operation: Optional[MessagingOperationValues],
57+
) -> None:
58+
span.set_attribute(SpanAttributes.MESSAGING_SYSTEM, "rabbitmq")
59+
if operation:
60+
span.set_attribute(SpanAttributes.MESSAGING_OPERATION, operation.value)
61+
else:
62+
span.set_attribute(SpanAttributes.MESSAGING_TEMP_DESTINATION, True)
63+
span.set_attribute(SpanAttributes.MESSAGING_DESTINATION, task_destination)
64+
span.set_attribute(
65+
SpanAttributes.MESSAGING_DESTINATION_KIND, properties.type
66+
)
67+
span.set_attribute(
68+
SpanAttributes.MESSAGING_MESSAGE_ID, properties.message_id
69+
)
70+
span.set_attribute(
71+
SpanAttributes.MESSAGING_CONVERSATION_ID, properties.correlation_id
72+
)
73+
span.set_attribute(
74+
SpanAttributes.NET_PEER_NAME, channel.connection.params.host
75+
)
76+
span.set_attribute(
77+
SpanAttributes.NET_PEER_PORT, channel.connection.params.port
78+
)

‎instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/version.py

+15
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
# Copyright The OpenTelemetry Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
__version__ = "0.24b0"

0 commit comments

Comments
 (0)