Add botocore db.statement sanitization for dynamoDB

CHANGELOG.md

@@ -14,6 +14,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,9 @@ 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)
 
@@ -243,9 +262,9 @@ def _apply_response_attributes(span: Span, result):
         headers = metadata.get("HTTPHeaders")
         if headers is not None:
             request_id = (
-                headers.get("x-amzn-RequestId")
-                or headers.get("x-amz-request-id")
-                or headers.get("x-amz-id-2")
+                    headers.get("x-amzn-RequestId")
+                    or headers.get("x-amz-request-id")
+                    or headers.get("x-amz-id-2")
             )
     if request_id:
         # TODO: update when semantic conventions exist
@@ -262,7 +281,7 @@ 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]]
 ) -> Optional[_AwsSdkCallContext]:
     try:
         call_context = _AwsSdkCallContext(client, args)

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

@@ -62,14 +62,44 @@ 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 str(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)
 
@@ -350,12 +380,25 @@ 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:
+            attributes[SpanAttributes.DB_STATEMENT] = str(
+                self._call_context.params
+            )
+
+            if self._configuration.get("sanitize_query"):
+                attributes[
+                    SpanAttributes.DB_STATEMENT
+                ] = _conv_params_to_sanitized_str(self._call_context.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,6 +42,7 @@ 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]]):
@@ -54,12 +53,19 @@ def __init__(self, client: _BotoClientT, args: Tuple[str, Dict[str, Any]]):
             _logger.warning("Could not get request params.")
             params = {}
 
+        try:
+            configurations = args[2]
+        except (IndexError, TypeError):
+            _logger.warning("Could not get request configurations.")
+            configurations = None
+
         boto_meta = client.meta
         service_model = boto_meta.service_model
 
         self.service = service_model.service_name.lower()  # type: str
         self.operation = operation  # type: str
         self.params = params  # type: Dict[str, Any]
+        self.configuration = configurations
 
         # 'operation' and 'service' are essential for instrumentation.
         # for all other attributes we extract them defensively. All of them should