[Elastic Search] Fix sanitization for bulk queries

[GilTeixeira]

Description

Fix sanitization for bulk queries for newer versions (>7.x). Continues the work of #1870

Fixes #1868

Type of change

Please delete options that are not relevant.

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• This change requires a documentation update

How Has This Been Tested?

from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor
from elasticsearch import Elasticsearch, helpers
 
ElasticsearchInstrumentor().instrument()
 
es = Elasticsearch(["http://localhost:9200"])
 
 
def gendata():
    mywords = ["foo", "bar", "baz"]
    for word in mywords:
        yield {
            "_index": "mywords",
            "word": word,
        }
 
 
helpers.bulk(es, gendata())

Currently it fails with this traceback:

Traceback (most recent call last):
  File "/Users/gil.teixeira/dev/elastic-telemetry/a.py", line 18, in <module>
    helpers.bulk(es, gendata())
  File "/Users/gil.teixeira/dev/elastic-telemetry/venv/lib/python3.11/site-packages/elasticsearch/helpers/actions.py", line 521, in bulk
    for ok, item in streaming_bulk(
  File "/Users/gil.teixeira/dev/elastic-telemetry/venv/lib/python3.11/site-packages/elasticsearch/helpers/actions.py", line 436, in streaming_bulk
    for data, (ok, info) in zip(
  File "/Users/gil.teixeira/dev/elastic-telemetry/venv/lib/python3.11/site-packages/elasticsearch/helpers/actions.py", line 339, in _process_bulk_chunk
    resp = client.bulk(*args, operations=bulk_actions, **kwargs)  # type: ignore[arg-type]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/gil.teixeira/dev/elastic-telemetry/venv/lib/python3.11/site-packages/elasticsearch/_sync/client/utils.py", line 414, in wrapped
    return api(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^
  File "/Users/gil.teixeira/dev/elastic-telemetry/venv/lib/python3.11/site-packages/elasticsearch/_sync/client/__init__.py", line 708, in bulk
    return self.perform_request(  # type: ignore[return-value]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/gil.teixeira/dev/elastic-telemetry/venv/lib/python3.11/site-packages/elasticsearch/_sync/client/_base.py", line 285, in perform_request
    meta, resp_body = self.transport.perform_request(
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/gil.teixeira/dev/elastic-telemetry/opentelemetry-python-contrib/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py", line 242, in wrapper
    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
                                              ^^^^^^^^^^^^^^
  File "/Users/gil.teixeira/dev/elastic-telemetry/opentelemetry-python-contrib/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py", line 58, in sanitize_body
    flatten_body = _flatten_dict(body)
                   ^^^^^^^^^^^^^^^^^^^
  File "/Users/gil.teixeira/dev/elastic-telemetry/opentelemetry-python-contrib/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py", line 31, in _flatten_dict
    for k, v in d.items():
                ^^^^^^^
AttributeError: 'list' object has no attribute 'items'

Does This PR Require a Core Repo Change?

• Yes. - Link to PR:
• No.

Checklist:

See contributing.md for styleguide, changelog guidelines, and more.

• Followed the style guidelines of this project
• Changelogs have been updated
• Unit tests have been added
• Documentation has been updated

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: GilTeixeira / name: Gil Teixeira (5e576df, 06acb8f, 90ee6a1, 63b4507, 07b1331, 5a2d6d2, 1e0b1f3, 2b72ac2, e488a3d, 8c54c5e, f84f8ee)
• ✅ login: ocelotl / name: Diego Hurtado (6db5340)
• ✅ login: srikanthccv / name: Srikanth Chekuri (1e0b1f3)

[ocelotl]

@GilTeixeira please add at least one test case for this changes :V:

[anpr]

This is blocking our upgrade to opentelemetry, could somebody please review and merge this?

[GilTeixeira]

@ocelotl I added a test. And this works with elasticsearch==8.x, since the body is a List[bytes].
But this will still throw an exception with elasticsearch<=7.x, since the body that is passed is formed like this: resp = client.bulk(*args, body="\n".join(bulk_actions) + "\n", **kwargs)
I am not sure what to do in these cases, do I wrap the whole sanitize_body in a try except?

[srikanthccv]

But this will still throw an exception with elasticsearch<=7.x .... I am not sure what to do in these cases

Inspect the version of the elasticsearch (import VERSION from elasticsearch) and use appropriate logic for sanitization. Something like

[GilTeixeira]

@srikanthccv I think you forgot to add the suggestion 😛
I splitted the string by newline and sanitize each line, lmk if you have a better idea

[anpr]

Would be great to have the final steps to be taken soon, thanks 🙏

[oliverholworthy]

Looks like this has been addressed by #2355 and this PR can be now closed

[srikanthccv]

Thanks @oliverholworthy. Please check with the latest release and close if it fixes your issue @GilTeixeira

[GilTeixeira]

That's it, good work 🚀

[anpr]

Thanks a lot 🙏 👍