Merge branch 'main' into metrics-instrumentation-celery

Author: shalevr

Diff for: .github/workflows/test.yml

@@ -6,7 +6,7 @@ on:
     - 'release/*'
   pull_request:
 env:
-  CORE_REPO_SHA: d0bb12b34b0c487198c935001636b6163485a50f
+  CORE_REPO_SHA: 2d1f0b9f5fce62549d1338882f37b91b95881c75
 
 jobs:
   build:

Diff for: CHANGELOG.md

@@ -10,13 +10,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add metric instrumentation for celery
   ([#1679](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1679))
 
+- Fix exception in Urllib3 when dealing with filelike body.
+  ([#1399](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1399))
+
+### Added
+
+- Add connection attributes to sqlalchemy connect span
+  ([#1608](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1608))
+- Add support for enabling Redis sanitization from environment variable
+  ([#1690](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1690))
+
+### Fixed
+
+- Fix Flask instrumentation to only close the span if it was created by the same thread.
+  ([#1654](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1654))
+
 ## Version 1.16.0/0.37b0 (2023-02-17)
 
 ### Added
 
 - Support `aio_pika` 9.x (([#1670](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1670])
 - `opentelemetry-instrumentation-redis` Add `sanitize_query` config option to allow query sanitization.  ([#1572](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1572))
-- `opentelemetry-instrumentation-elasticsearch` Add optional db.statement query sanitization. 
+- `opentelemetry-instrumentation-elasticsearch` Add optional db.statement query sanitization.
   ([#1598](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1598))
 - `opentelemetry-instrumentation-celery` Record exceptions as events on the span.
   ([#1573](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1573))

Diff for: docs-requirements.txt

@@ -24,6 +24,7 @@ aiopg>=0.13.0,<1.3.0
 asyncpg>=0.12.0
 boto~=2.0
 botocore~=1.0
+boto3~=1.0
 celery>=4.0
 confluent-kafka>= 1.8.2,< 2.0.0
 elasticsearch>=2.0,<9.0

Diff for: docs/conf.py

@@ -126,7 +126,7 @@ def getlistcfg(strval):
     ]
 
 
-ignore_categories = ["py-class", "py-func", "py-exc", "any"]
+ignore_categories = ["py-class", "py-func", "py-exc", "py-obj", "any"]
 
 for category in ignore_categories:
     if category in mcfg:

Diff for: docs/instrumentation/boto3sqs/boto3sqs.rst

@@ -0,0 +1,6 @@
+.. include:: ../../../instrumentation/opentelemetry-instrumentation-boto3sqs/README.rst
+
+.. automodule:: opentelemetry.instrumentation.boto3sqs
+    :members:
+    :undoc-members:
+    :show-inheritance:

Diff for: docs/nitpick-exceptions.ini

@@ -13,6 +13,7 @@ py-class=
     opentelemetry.sdk.trace.id_generator.IdGenerator
     opentelemetry.instrumentation.confluent_kafka.ProxiedProducer
     opentelemetry.instrumentation.confluent_kafka.ProxiedConsumer
+    opentelemetry.instrumentation.instrumentor.BaseInstrumentor
     ; - AwsXRayIdGenerator
     TextMapPropagator
     CarrierT
@@ -54,7 +55,16 @@ any=
     ; - instrumentation.*
     Setter
     httpx
-;
+    instrument
+    __iter__
+    list.__iter__
+    __getitem__
+    list.__getitem__
+    SQS.ReceiveMessage
+
+py-obj=
+    opentelemetry.propagators.textmap.CarrierT
+    
 py-func=
     poll
     flush

Diff for: exporter/opentelemetry-exporter-prometheus-remote-write/example/Dockerfile

@@ -6,6 +6,6 @@ WORKDIR /code
 COPY . .
 
 RUN pip install -e .
-RUN pip install -r ./examples/requirements.txt
+RUN pip install -r ./example/requirements.txt
 
-CMD ["python", "./examples/sampleapp.py"]
+CMD ["python", "./example/sampleapp.py"]

Diff for: exporter/opentelemetry-exporter-prometheus-remote-write/example/README.md

@@ -14,7 +14,7 @@ data
 *Users do not need to install Python as the app will be run in the Docker Container*
 
 ## Instructions
-1. Run `docker-compose up -d` in the the `examples/` directory
+1. Run `docker-compose up -d` in the the `example/` directory
 
 The `-d` flag causes all services to run in detached mode and frees up your
 terminal session. This also causes no logs to show up. Users can attach themselves to the service's logs manually using `docker logs ${CONTAINER_ID} --follow`
@@ -39,4 +39,4 @@ terminal session. This also causes no logs to show up. Users can attach themselv
    * Click the refresh button and data should show up on the graph
 
 6. Shutdown the services when finished
-   * Run `docker-compose down` in the examples directory
+   * Run `docker-compose down` in the example directory

Diff for: exporter/opentelemetry-exporter-prometheus-remote-write/example/docker-compose.yml

@@ -30,4 +30,4 @@ services:
   sample_app:
     build:
       context: ../
-      dockerfile: ./examples/Dockerfile
+      dockerfile: ./example/Dockerfile

Diff for: instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py

@@ -16,17 +16,17 @@
 
 .. _boto3sqs: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sqs.html
 
-
 Usage
 -----
 
-.. code:: python
+.. code-block:: python
 
     import boto3
     from opentelemetry.instrumentation.boto3sqs import Boto3SQSInstrumentor
 
-
     Boto3SQSInstrumentor().instrument()
+
+---
 """
 import logging
 from typing import Any, Collection, Dict, Generator, List, Mapping, Optional

Diff for: instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py

@@ -238,8 +238,8 @@ def response_hook(span: Span, status: str, response_headers: List):
 API
 ---
 """
-
 from logging import getLogger
+from threading import get_ident
 from time import time_ns
 from timeit import default_timer
 from typing import Collection
@@ -265,6 +265,7 @@ def response_hook(span: Span, status: str, response_headers: List):
 _ENVIRON_STARTTIME_KEY = "opentelemetry-flask.starttime_key"
 _ENVIRON_SPAN_KEY = "opentelemetry-flask.span_key"
 _ENVIRON_ACTIVATION_KEY = "opentelemetry-flask.activation_key"
+_ENVIRON_THREAD_ID_KEY = "opentelemetry-flask.thread_id_key"
 _ENVIRON_TOKEN = "opentelemetry-flask.token"
 
 _excluded_urls_from_env = get_excluded_urls("FLASK")
@@ -398,6 +399,7 @@ def _before_request():
         activation = trace.use_span(span, end_on_exit=True)
         activation.__enter__()  # pylint: disable=E1101
         flask_request_environ[_ENVIRON_ACTIVATION_KEY] = activation
+        flask_request_environ[_ENVIRON_THREAD_ID_KEY] = get_ident()
         flask_request_environ[_ENVIRON_SPAN_KEY] = span
         flask_request_environ[_ENVIRON_TOKEN] = token
 
@@ -437,10 +439,17 @@ def _teardown_request(exc):
             return
 
         activation = flask.request.environ.get(_ENVIRON_ACTIVATION_KEY)
-        if not activation:
+        thread_id = flask.request.environ.get(_ENVIRON_THREAD_ID_KEY)
+        if not activation or thread_id != get_ident():
             # This request didn't start a span, maybe because it was created in
             # a way that doesn't run `before_request`, like when it is created
             # with `app.test_request_context`.
+            #
+            # Similarly, check the thread_id against the current thread to ensure
+            # tear down only happens on the original thread. This situation can
+            # arise if the original thread handling the request spawn children
+            # threads and then uses something like copy_current_request_context
+            # to copy the request context.
             return
         if exc is None:
             activation.__exit__(None, None, None)

Diff for: instrumentation/opentelemetry-instrumentation-flask/tests/base_test.py

@@ -12,6 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from random import randint
+
 import flask
 from werkzeug.test import Client
 from werkzeug.wrappers import Response
@@ -34,6 +37,25 @@ def _sqlcommenter_endpoint():
         )
         return sqlcommenter_flask_values
 
+    @staticmethod
+    def _multithreaded_endpoint(count):
+        def do_random_stuff():
+            @flask.copy_current_request_context
+            def inner():
+                return randint(0, 100)
+
+            return inner
+
+        executor = ThreadPoolExecutor(count)
+        futures = []
+        for _ in range(count):
+            futures.append(executor.submit(do_random_stuff()))
+        numbers = []
+        for future in as_completed(futures):
+            numbers.append(future.result())
+
+        return " ".join([str(i) for i in numbers])
+
     @staticmethod
     def _custom_response_headers():
         resp = flask.Response("test response")
@@ -61,6 +83,7 @@ def excluded2_endpoint():
         # pylint: disable=no-member
         self.app.route("/hello/<int:helloid>")(self._hello_endpoint)
         self.app.route("/sqlcommenter")(self._sqlcommenter_endpoint)
+        self.app.route("/multithreaded")(self._multithreaded_endpoint)
         self.app.route("/excluded/<int:helloid>")(self._hello_endpoint)
         self.app.route("/excluded")(excluded_endpoint)
         self.app.route("/excluded2")(excluded2_endpoint)

Diff for: instrumentation/opentelemetry-instrumentation-flask/tests/test_multithreading.py

@@ -0,0 +1,52 @@
+# 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.
+
+import flask
+from werkzeug.test import Client
+from werkzeug.wrappers import Response
+
+from opentelemetry.instrumentation.flask import FlaskInstrumentor
+from opentelemetry.test.wsgitestutil import WsgiTestBase
+
+# pylint: disable=import-error
+from .base_test import InstrumentationTest
+
+
+class TestMultiThreading(InstrumentationTest, WsgiTestBase):
+    def setUp(self):
+        super().setUp()
+        FlaskInstrumentor().instrument()
+        self.app = flask.Flask(__name__)
+        self._common_initialization()
+
+    def tearDown(self):
+        super().tearDown()
+        with self.disable_logging():
+            FlaskInstrumentor().uninstrument()
+
+    def test_multithreaded(self):
+        """Test that instrumentation tear down does not blow up
+        when the request thread spawn children threads and the request
+        context is copied to the children threads
+        """
+        self.app = flask.Flask(__name__)
+        self.app.route("/multithreaded/<int:count>")(
+            self._multithreaded_endpoint
+        )
+        client = Client(self.app, Response)
+        count = 5
+        resp = client.get(f"/multithreaded/{count}")
+        self.assertEqual(200, resp.status_code)
+        # Should return the specified number of random integers
+        self.assertEqual(count, len(resp.text.split(" ")))

Diff for: instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py

@@ -88,17 +88,37 @@ def response_hook(span, instance, response):
     client = redis.StrictRedis(host="localhost", port=6379)
     client.get("my-key")
 
+Configuration
+-------------
+
+Query sanitization
+******************
+To enable query sanitization with an environment variable, set
+``OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS`` to "true".
+
+For example,
+
+::
+
+    export OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS="true"
+
+will result in traced queries like "SET ? ?".
+
 API
 ---
 """
 import typing
+from os import environ
 from typing import Any, Collection
 
 import redis
 from wrapt import wrap_function_wrapper
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+from opentelemetry.instrumentation.redis.environment_variables import (
+    OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS,
+)
 from opentelemetry.instrumentation.redis.package import _instruments
 from opentelemetry.instrumentation.redis.util import (
     _extract_conn_attributes,
@@ -287,7 +307,15 @@ def _instrument(self, **kwargs):
             tracer,
             request_hook=kwargs.get("request_hook"),
             response_hook=kwargs.get("response_hook"),
-            sanitize_query=kwargs.get("sanitize_query", False),
+            sanitize_query=kwargs.get(
+                "sanitize_query",
+                environ.get(
+                    OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS, "false"
+                )
+                .lower()
+                .strip()
+                == "true",
+            ),
         )
 
     def _uninstrument(self, **kwargs):

Diff for: instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/environment_variables.py

@@ -0,0 +1,17 @@
+# 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.
+
+OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS = (
+    "OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS"
+)

Diff for: instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py

@@ -168,6 +168,32 @@ def test_query_sanitizer_enabled(self):
         span = spans[0]
         self.assertEqual(span.attributes.get("db.statement"), "SET ? ?")
 
+    def test_query_sanitizer_enabled_env(self):
+        redis_client = redis.Redis()
+        connection = redis.connection.Connection()
+        redis_client.connection = connection
+
+        RedisInstrumentor().uninstrument()
+
+        env_patch = mock.patch.dict(
+            "os.environ",
+            {"OTEL_PYTHON_INSTRUMENTATION_SANITIZE_REDIS": "true"},
+        )
+        env_patch.start()
+        RedisInstrumentor().instrument(
+            tracer_provider=self.tracer_provider,
+        )
+
+        with mock.patch.object(redis_client, "connection"):
+            redis_client.set("key", "value")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+
+        span = spans[0]
+        self.assertEqual(span.attributes.get("db.statement"), "SET ? ?")
+        env_patch.stop()
+
     def test_query_sanitizer_disabled(self):
         redis_client = redis.Redis()
         connection = redis.connection.Connection()