fix(botocore): fix kinesis typeerror [backports #5317 to 1.9] (#5333)

Backports: #5317

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/contributing.html#Release-Note-Guidelines)
are followed.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Author is aware of the performance implications of this PR as
reported in the benchmarks PR comment.

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer is aware of, and discussed the performance implications
of this PR as reported in the benchmarks PR comment.

Co-authored-by: Jordan Gonzalez <[email protected]>
Co-authored-by: Emmett Butler <[email protected]>

Author: 3 people

ddtrace/contrib/botocore/patch.py

@@ -186,7 +186,7 @@ def get_json_from_str(data_str):
 
     if data_str.endswith(LINE_BREAK):
         return LINE_BREAK, data_obj
-    return "", data_obj
+    return None, data_obj
 
 
 def get_kinesis_data_object(data):
@@ -195,24 +195,34 @@ def get_kinesis_data_object(data):
     :data: the data from a kinesis stream
 
     The data from a kinesis stream comes as a string (could be json, base64 encoded, etc.)
-    We support injecting our trace context in the following two cases:
+    We support injecting our trace context in the following three cases:
     - json string
+    - byte encoded json string
     - base64 encoded json string
     If it's neither of these, then we leave the message as it is.
     """
 
     # check if data is a json string
     try:
         return get_json_from_str(data)
-    except ValueError:
-        pass
+    except Exception:
+        log.debug("Kinesis data is not a JSON string. Trying Byte encoded JSON string.")
+
+    # check if data is an encoded json string
+    try:
+        data_str = data.decode("ascii")
+        return get_json_from_str(data_str)
+    except Exception:
+        log.debug("Kinesis data is not a JSON string encoded. Trying Base64 encoded JSON string.")
 
     # check if data is a base64 encoded json string
     try:
         data_str = base64.b64decode(data).decode("ascii")
         return get_json_from_str(data_str)
-    except ValueError:
-        raise TraceInjectionDecodingError("Unable to parse kinesis streams data string")
+    except Exception:
+        log.error("Unable to parse payload, unable to inject trace context.")
+
+    return None, None
 
 
 def inject_trace_to_kinesis_stream_data(record, span):
@@ -231,22 +241,23 @@ def inject_trace_to_kinesis_stream_data(record, span):
 
     data = record["Data"]
     line_break, data_obj = get_kinesis_data_object(data)
-    data_obj["_datadog"] = {}
-    HTTPPropagator.inject(span.context, data_obj["_datadog"])
-    data_json = json.dumps(data_obj)
-
-    # if original string had a line break, add it back
-    if line_break:
-        data_json += line_break
-
-    # check if data size will exceed max size with headers
-    data_size = len(data_json)
-    if data_size >= MAX_KINESIS_DATA_SIZE:
-        raise TraceInjectionSizeExceed(
-            "Data including trace injection ({}) exceeds ({})".format(data_size, MAX_KINESIS_DATA_SIZE)
-        )
-
-    record["Data"] = data_json
+    if data_obj is not None:
+        data_obj["_datadog"] = {}
+        HTTPPropagator.inject(span.context, data_obj["_datadog"])
+        data_json = json.dumps(data_obj)
+
+        # if original string had a line break, add it back
+        if line_break is not None:
+            data_json += line_break
+
+        # check if data size will exceed max size with headers
+        data_size = len(data_json)
+        if data_size >= MAX_KINESIS_DATA_SIZE:
+            raise TraceInjectionSizeExceed(
+                "Data including trace injection ({}) exceeds ({})".format(data_size, MAX_KINESIS_DATA_SIZE)
+            )
+
+        record["Data"] = data_json
 
 
 def inject_trace_to_kinesis_stream(params, span):

releasenotes/notes/fix-botocore-newline-str-issues-fab81e395207b0ed.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    botocore: Fix TypeError raised by injecting trace context into Kinesis messages.