Add botocore db.statement sanitization for dynamoDB

CHANGELOG.md

@@ -50,6 +50,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `opentelemetry-instrumentation-botocore` Add optional db.statement query sanitization.
+  ([#1662](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1662))
 - 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.

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

@@ -66,6 +66,8 @@ def request_hook(span, service_name, operation_name, api_params):
     def response_hook(span, service_name, operation_name, result):
         # response hook logic
 
+    dynamodb_sanitize_query (bool) - an optional dynamodb query sanitization flag, default is False
+
     # Instrument Botocore with hooks
     BotocoreInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)
 
@@ -76,6 +78,9 @@ def response_hook(span, service_name, operation_name, result):
     )
     ec2 = self.session.create_client("ec2", region_name="us-west-2")
     ec2.describe_instances()
+
+    # Instrument Botocore with dynamoDB sanitization enabled - default is False
+    BotocoreInstrumentor().instrument(dynamodb_sanitize_query=True)
 """
 
 import logging
@@ -117,6 +122,17 @@ def _patched_endpoint_prepare_request(wrapped, instance, args, kwargs):
     return wrapped(*args, **kwargs)
 
 
+class _ExtensionsConfiguration:
+    def __init__(self, **kwargs):
+        dynamodb_sanitize_query = kwargs.get("dynamodb_sanitize_query", False)
+        self.dynamodb_configuration = {
+            "sanitize_query": dynamodb_sanitize_query
+        }
+
+    def get_dynamodb_configuration(self):
+        return self.dynamodb_configuration
+
+
 class BotocoreInstrumentor(BaseInstrumentor):
     """An instrumentor for Botocore.
 
@@ -139,6 +155,7 @@ def _instrument(self, **kwargs):
 
         self.request_hook = kwargs.get("request_hook")
         self.response_hook = kwargs.get("response_hook")
+        self.configuration = _ExtensionsConfiguration(**kwargs)
 
         wrap_function_wrapper(
             "botocore.client",
@@ -161,7 +178,11 @@ def _patched_api_call(self, original_func, instance, args, kwargs):
         if context_api.get_value(_SUPPRESS_INSTRUMENTATION_KEY):
             return original_func(*args, **kwargs)
 
-        call_context = _determine_call_context(instance, args)
+        call_context = _determine_call_context(
+            instance,
+            args,
+            self.configuration,
+        )
         if call_context is None:
             return original_func(*args, **kwargs)
 
@@ -262,10 +283,12 @@ def _apply_response_attributes(span: Span, result):
 
 
 def _determine_call_context(
-    client: BaseClient, args: Tuple[str, Dict[str, Any]]
+    client: BaseClient,
+    args: Tuple[str, Dict[str, Any]],
+    configuration: _ExtensionsConfiguration,
 ) -> Optional[_AwsSdkCallContext]:
     try:
-        call_context = _AwsSdkCallContext(client, args)
+        call_context = _AwsSdkCallContext(client, args, configuration)
 
         logger.debug(
             "AWS SDK invocation: %s %s",

instrumentation/opentelemetry-instrumentation-botocore/src/opentelemetry/instrumentation/botocore/extensions/dynamodb.py

@@ -62,14 +62,43 @@ def _conv_val_to_len(value) -> Optional[int]:
     return len(value) if value is not None else None
 
 
+_db_statement_operations = {
+    "BatchGetItem",
+    "BatchWriteItem",
+    "DeleteItem",
+    "GetItem",
+    "PutItem",
+    "Query",
+    "Scan",
+    "UpdateItem",
+}
+
+_sanitized_keys = (
+    "Item",
+    "Key",
+    "KeyConditionExpression",
+    "ExpressionAttributeValues",
+    "FilterExpression",
+)
+_sanitized_value = {"?": "?"}
+
+
+# pylint: disable=C0103
+def _conv_params_to_sanitized_str(params) -> str:
+    p = params.copy()
+    for key in p:
+        if key in _sanitized_keys:
+            p[key] = _sanitized_value
+    return p
+
+
 ################################################################################
 # common request attributes
 ################################################################################
 
 _REQ_TABLE_NAME = ("TableName", _conv_val_to_single_attr_tuple)
 _REQ_REQITEMS_TABLE_NAMES = ("RequestItems", _conv_dict_to_key_tuple)
 
-
 _REQ_GLOBAL_SEC_INDEXES = ("GlobalSecondaryIndexes", _conv_list_to_json_list)
 _REQ_LOCAL_SEC_INDEXES = ("LocalSecondaryIndexes", _conv_list_to_json_list)
 
@@ -83,7 +112,6 @@ def _conv_val_to_len(value) -> Optional[int]:
 _REQ_SELECT = ("Select", None)
 _REQ_INDEX_NAME = ("IndexName", None)
 
-
 ################################################################################
 # common response attributes
 ################################################################################
@@ -350,12 +378,27 @@ class _DynamoDbExtension(_AwsSdkExtension):
     def __init__(self, call_context: _AwsSdkCallContext):
         super().__init__(call_context)
         self._op = _OPERATION_MAPPING.get(call_context.operation)
+        self._configuration = (
+            call_context.configuration.get_dynamodb_configuration()
+        )
 
     def extract_attributes(self, attributes: _AttributeMapT):
         attributes[SpanAttributes.DB_SYSTEM] = DbSystemValues.DYNAMODB.value
         attributes[SpanAttributes.DB_OPERATION] = self._call_context.operation
         attributes[SpanAttributes.NET_PEER_NAME] = self._get_peer_name()
 
+        if (
+            self._call_context.operation in _db_statement_operations
+            and self._call_context.params
+        ):
+            params = self._call_context.params
+            if self._configuration.get("sanitize_query"):
+                params = _conv_params_to_sanitized_str(params)
+
+            attributes[SpanAttributes.DB_STATEMENT] = _conv_dict_to_json_str(
+                params
+            )
+
         if self._op is None:
             return
 

instrumentation/opentelemetry-instrumentation-botocore/src/opentelemetry/instrumentation/botocore/extensions/types.py

@@ -24,8 +24,6 @@
 _BotoClientT = "botocore.client.BaseClient"
 _BotoResultT = Dict[str, Any]
 _BotoClientErrorT = "botocore.exceptions.ClientError"
-
-_OperationParamsT = Dict[str, Any]
 _AttributeMapT = Dict[str, AttributeValue]
 
 
@@ -44,9 +42,15 @@ class _AwsSdkCallContext:
         api_version: the API version of the called AWS service.
         span_name: the name used to create the span.
         span_kind: the kind used to create the span.
+        configuration: a class represents additional configuration for botocore extensions.
     """
 
-    def __init__(self, client: _BotoClientT, args: Tuple[str, Dict[str, Any]]):
+    def __init__(
+        self,
+        client: _BotoClientT,
+        args: Tuple[str, Dict[str, Any]],
+        configuration,
+    ):
         operation = args[0]
         try:
             params = args[1]
@@ -60,6 +64,7 @@ def __init__(self, client: _BotoClientT, args: Tuple[str, Dict[str, Any]]):
         self.service = service_model.service_name.lower()  # type: str
         self.operation = operation  # type: str
         self.params = params  # type: Dict[str, Any]
+        self.configuration = configuration
 
         # 'operation' and 'service' are essential for instrumentation.
         # for all other attributes we extract them defensively. All of them should