Fix elastic-search sanitization for bulk queries

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fix elastic-search instrumentation sanitization to support bulk queries
+  ([#1870](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1870))
 - Update falcon instrumentation to follow semantic conventions
   ([#1824](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1824))
 

instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py

@@ -11,6 +11,7 @@
 # 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 json
 
 sanitized_keys = (
     "message",
@@ -51,6 +52,9 @@ def _unflatten_dict(d):
 
 
 def sanitize_body(body) -> str:
+    if isinstance(body, str):
+        body = json.loads(body)
+
     flatten_body = _flatten_dict(body)
 
     for key in flatten_body:

instrumentation/opentelemetry-instrumentation-elasticsearch/tests/test_elasticsearch.py

@@ -479,3 +479,7 @@ def test_body_sanitization(self, _):
             sanitize_body(sanitization_queries.filter_query),
             str(sanitization_queries.filter_query_sanitized),
         )
+        self.assertEqual(
+            sanitize_body(json.dumps(sanitization_queries.interval_query)),
+            str(sanitization_queries.interval_query_sanitized),
+        )