Merge branch 'master' into record-exceptions

Author: owais

docs/examples/basic_meter/http.py

@@ -0,0 +1,42 @@
+# 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.
+#
+"""
+This module shows how you can enable collection and exporting of http metrics
+related to instrumentations.
+"""
+import requests
+
+from opentelemetry import metrics
+from opentelemetry.instrumentation.requests import RequestsInstrumentor
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import ConsoleMetricsExporter
+
+# Sets the global MeterProvider instance
+metrics.set_meter_provider(MeterProvider())
+
+# Exporter to export metrics to the console
+exporter = ConsoleMetricsExporter()
+
+# Instrument the requests library
+RequestsInstrumentor().instrument()
+
+# Indicate to start collecting and exporting requests related metrics
+metrics.get_meter_provider().start_pipeline(
+    RequestsInstrumentor().meter, exporter, 5
+)
+
+response = requests.get("http://example.com")
+
+input("...\n")

docs/instrumentation/instrumentation.rst

@@ -13,3 +13,4 @@ Submodules
     :maxdepth: 1
 
     instrumentor
+    metric

docs/instrumentation/metric.rst

@@ -0,0 +1,7 @@
+opentelemetry.instrumentation.metric package
+============================================
+
+.. automodule:: opentelemetry.instrumentation.metric
+    :members:
+    :undoc-members:
+    :show-inheritance:

exporter/opentelemetry-exporter-zipkin/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Zipkin exporter now accepts a ``max_tag_value_length`` attribute to customize the
   maximum allowed size a tag value can have. ([#1151](https://github.com/open-telemetry/opentelemetry-python/pull/1151)) 
+- Fixed OTLP events to Zipkin annotations translation. ([#1161](https://github.com/open-telemetry/opentelemetry-python/pull/1161))
 
 ## Version 0.13b0
 

exporter/opentelemetry-exporter-zipkin/src/opentelemetry/exporter/zipkin/__init__.py

@@ -237,20 +237,25 @@ def _extract_tags_from_span(self, span: Span):
             tags.update(self._extract_tags_from_dict(span.resource.attributes))
         return tags
 
-    def _extract_annotations_from_events(
-        self, events
-    ):  # pylint: disable=R0201
-        return (
-            [
+    def _extract_annotations_from_events(self, events):
+        if not events:
+            return None
+
+        annotations = []
+        for event in events:
+            attrs = {}
+            for key, value in event.attributes.items():
+                if isinstance(value, str):
+                    value = value[: self.max_tag_value_length]
+                attrs[key] = value
+
+            annotations.append(
                 {
-                    "timestamp": _nsec_to_usec_round(e.timestamp),
-                    "value": e.name,
+                    "timestamp": _nsec_to_usec_round(event.timestamp),
+                    "value": json.dumps({event.name: attrs}),
                 }
-                for e in events
-            ]
-            if events
-            else None
-        )
+            )
+        return annotations
 
 
 def _nsec_to_usec_round(nsec):

exporter/opentelemetry-exporter-zipkin/tests/test_zipkin_exporter.py

@@ -98,9 +98,8 @@ def test_constructor_explicit(self):
         self.assertEqual(exporter.ipv6, ipv6)
         self.assertEqual(exporter.url, url)
 
-    # pylint: disable=too-many-locals
+    # pylint: disable=too-many-locals,too-many-statements
     def test_export(self):
-
         span_names = ("test1", "test2", "test3", "test4")
         trace_id = 0x6E0C63257DE34C926F9EFCD03927272E
         span_id = 0x34BF92DEEFC58C92
@@ -204,7 +203,7 @@ def test_export(self):
         local_endpoint = {"serviceName": service_name, "port": 9411}
 
         exporter = ZipkinSpanExporter(service_name)
-        expected = [
+        expected_spans = [
             {
                 "traceId": format(trace_id, "x"),
                 "id": format(span_id, "x"),
@@ -220,14 +219,20 @@ def test_export(self):
                     "otel.status_code": "2",
                     "otel.status_description": "Example description",
                 },
+                "debug": True,
+                "parentId": format(parent_id, "x"),
                 "annotations": [
                     {
                         "timestamp": event_timestamp // 10 ** 3,
-                        "value": "event0",
+                        "value": {
+                            "event0": {
+                                "annotation_bool": True,
+                                "annotation_string": "annotation_test",
+                                "key_float": 0.3,
+                            }
+                        },
                     }
                 ],
-                "debug": True,
-                "parentId": format(parent_id, "x"),
             },
             {
                 "traceId": format(trace_id, "x"),
@@ -281,11 +286,21 @@ def test_export(self):
             status = exporter.export(otel_spans)
             self.assertEqual(SpanExportResult.SUCCESS, status)
 
-        mock_post.assert_called_with(
-            url="http://localhost:9411/api/v2/spans",
-            data=json.dumps(expected),
-            headers={"Content-Type": "application/json"},
+        # pylint: disable=unsubscriptable-object
+        kwargs = mock_post.call_args[1]
+
+        self.assertEqual(kwargs["url"], "http://localhost:9411/api/v2/spans")
+        actual_spans = sorted(
+            json.loads(kwargs["data"]), key=lambda span: span["timestamp"]
         )
+        for expected, actual in zip(expected_spans, actual_spans):
+            expected_annotations = expected.pop("annotations", None)
+            actual_annotations = actual.pop("annotations", None)
+            if actual_annotations:
+                for annotation in actual_annotations:
+                    annotation["value"] = json.loads(annotation["value"])
+            self.assertEqual(expected, actual)
+            self.assertEqual(expected_annotations, actual_annotations)
 
     # pylint: disable=too-many-locals
     def test_zero_padding(self):

instrumentation/opentelemetry-instrumentation-celery/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Span operation names now include the task type. ([#1135](https://github.com/open-telemetry/opentelemetry-python/pull/1135))
+- Added automatic context propagation. ([#1135](https://github.com/open-telemetry/opentelemetry-python/pull/1135))
+
 ## Version 0.12b0
 
 Released 2020-08-14

instrumentation/opentelemetry-instrumentation-celery/README.rst

@@ -29,11 +29,20 @@ Usage
 
 .. code-block:: python
 
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchExportSpanProcessor
     from opentelemetry.instrumentation.celery import CeleryInstrumentor
 
-    CeleryInstrumentor().instrument()
-
     from celery import Celery
+    from celery.signals import worker_process_init
+
+    @worker_process_init.connect(weak=False)
+    def init_celery_tracing(*args, **kwargs):
+        trace.set_tracer_provider(TracerProvider())
+        span_processor = BatchExportSpanProcessor(ConsoleSpanExporter())
+        trace.get_tracer_provider().add_span_processor(span_processor)
+        CeleryInstrumentor().instrument()
 
     app = Celery("tasks", broker="amqp://localhost")
 
@@ -43,6 +52,15 @@ Usage
 
     add.delay(42, 50)
 
+
+Setting up tracing 
+--------------------
+
+When tracing a celery worker process, tracing and instrumention both must be initialized after the celery worker
+process is initialized. This is required for any tracing components that might use threading to work correctly
+such as the BatchExportSpanProcessor. Celery provides a signal called ``worker_process_init`` that can be used to
+accomplish this as shown in the example above.
+
 References
 ----------
 * `OpenTelemetry Celery Instrumentation <https://opentelemetry-python.readthedocs.io/en/latest/ext/celery/celery.html>`_

instrumentation/opentelemetry-instrumentation-celery/setup.cfg

@@ -46,6 +46,7 @@ install_requires =
 [options.extras_require]
 test =
     pytest
+    celery ~= 4.0
     opentelemetry-test == 0.14.dev0
 
 [options.packages.find]

instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py

@@ -30,11 +30,20 @@
 
 .. code:: python
 
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchExportSpanProcessor
     from opentelemetry.instrumentation.celery import CeleryInstrumentor
 
-    CeleryInstrumentor().instrument()
-
     from celery import Celery
+    from celery.signals import worker_process_init
+
+    @worker_process_init.connect(weak=False)
+    def init_celery_tracing(*args, **kwargs):
+        trace.set_tracer_provider(TracerProvider())
+        span_processor = BatchExportSpanProcessor(ConsoleSpanExporter())
+        trace.get_tracer_provider().add_span_processor(span_processor)
+        CeleryInstrumentor().instrument()
 
     app = Celery("tasks", broker="amqp://localhost")
 
@@ -50,13 +59,15 @@ def add(x, y):
 
 import logging
 import signal
+from collections.abc import Iterable
 
 from celery import signals  # pylint: disable=no-name-in-module
 
-from opentelemetry import trace
+from opentelemetry import propagators, trace
 from opentelemetry.instrumentation.celery import utils
 from opentelemetry.instrumentation.celery.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+from opentelemetry.trace.propagation import get_current_span
 from opentelemetry.trace.status import Status, StatusCanonicalCode
 
 logger = logging.getLogger(__name__)
@@ -106,9 +117,16 @@ def _trace_prerun(self, *args, **kwargs):
         if task is None or task_id is None:
             return
 
+        request = task.request
+        tracectx = propagators.extract(carrier_extractor, request) or {}
+        parent = get_current_span(tracectx)
+
         logger.debug("prerun signal start task_id=%s", task_id)
 
-        span = self._tracer.start_span(task.name, kind=trace.SpanKind.CONSUMER)
+        operation_name = "{0}/{1}".format(_TASK_RUN, task.name)
+        span = self._tracer.start_span(
+            operation_name, parent=parent, kind=trace.SpanKind.CONSUMER
+        )
 
         activation = self._tracer.use_span(span, end_on_exit=True)
         activation.__enter__()
@@ -146,7 +164,10 @@ def _trace_before_publish(self, *args, **kwargs):
         if task is None or task_id is None:
             return
 
-        span = self._tracer.start_span(task.name, kind=trace.SpanKind.PRODUCER)
+        operation_name = "{0}/{1}".format(_TASK_APPLY_ASYNC, task.name)
+        span = self._tracer.start_span(
+            operation_name, kind=trace.SpanKind.PRODUCER
+        )
 
         # apply some attributes here because most of the data is not available
         span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)
@@ -158,6 +179,10 @@ def _trace_before_publish(self, *args, **kwargs):
         activation.__enter__()
         utils.attach_span(task, task_id, (span, activation), is_publish=True)
 
+        headers = kwargs.get("headers")
+        if headers:
+            propagators.inject(type(headers).__setitem__, headers)
+
     @staticmethod
     def _trace_after_publish(*args, **kwargs):
         task = utils.retrieve_task_from_sender(kwargs)
@@ -221,3 +246,10 @@ def _trace_retry(*args, **kwargs):
         # Use `str(reason)` instead of `reason.message` in case we get
         # something that isn't an `Exception`
         span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))
+
+
+def carrier_extractor(carrier, key):
+    value = getattr(carrier, key, [])
+    if isinstance(value, str) or not isinstance(value, Iterable):
+        value = (value,)
+    return value