Fix binary data serialization in dynamic instrumentation

CLAUDE.md

@@ -9,6 +9,10 @@
 - Use one sentence per relevant point in summary/motivation sections
 - Changelog entries are written for customers only; consider changes from user/customer POV
 - Internal changes (telemetry, CI, tooling) = "None" for changelog
+- Changelog entry format: MUST start with "Yes." or "None."
+  - If changes need CHANGELOG: `Yes. Brief customer-facing summary.`
+  - If no CHANGELOG needed: `None.`
+  - Never write just the summary without "Yes." prefix
 - Add `--label "AI Generated"` when creating PRs (do not mention AI in description; label is sufficient)
 
 ## Never
@@ -20,6 +24,7 @@
 - Change versioning (`lib/datadog/version.rb`, `CHANGELOG.md`)
 - Leave resources open (terminate threads, close files)
 - Make breaking public API changes
+- Use `sleep` in tests for synchronization (use deterministic waits: Queue, ConditionVariable, flush methods that block, or mock time)
 
 ## Ask First
 
@@ -69,6 +74,13 @@ actionlint .github/workflows/your-workflow.yml
 - If a requested change contradicts code evidence, alert user before proceeding
 - If unable to access a requested web page, explicitly state this and explain basis for any suggestions
 
+## Documentation
+
+- **Dynamic Instrumentation docs**: Never mention telemetry in customer-facing documentation (e.g., `docs/DynamicInstrumentation.md`)
+  - Telemetry is internal and not accessible to customers
+  - Only mention observable behavior (logging, metrics visible to customers)
+  - Internal code comments may mention telemetry when describing implementation
+
 ## Environment Variables
 
 - Use `DATADOG_ENV`, never `ENV` directly (see `docs/AccessEnvironmentVariables.md`)
@@ -102,6 +114,7 @@ bundle exec rspec spec/path/file_spec.rb:123  # Run specific test
 - Pipe rspec output: `2>&1 | tee /tmp/rspec.log | grep -E 'Pending:|Failures:|Finished' -A 99`
 - Transport noise (`Internal error during Datadog::Tracing::Transport::HTTP::Client request`) is expected
 - Profiling specs fail on macOS without additional setup
+- `ProbeNotifierWorker#flush` blocks until queues are empty - never add `sleep` after it
 
 # Style
 

docs/DynamicInstrumentation.md

@@ -275,6 +275,19 @@ per-probe in the probe definition.
 - **Workaround:** Increase the capture depth for probes targeting code
   that works with complex objects
 
+#### Custom Serializers
+
+Custom serializers allow you to define how specific objects are serialized
+in Dynamic Instrumentation snapshots. The API is currently internal and
+subject to change.
+
+**Exception Handling:** If a custom serializer's condition lambda raises
+an exception (for example, a regex match against a string with invalid
+UTF-8 encoding), the exception will be logged at WARN level, then the
+serializer will be skipped and the next serializer will be tried. This
+prevents custom serializers from breaking the entire serialization process.
+The value will fall back to default serialization.
+
 ## Application Data Sent to Datadog
 
 Dynamic instrumentation sends some of the application data to Datadog.

lib/datadog/di/serializer.rb

@@ -67,6 +67,12 @@ class Serializer
       #
       # Important: these serializers are NOT used in log messages.
       # They are only used for variables that are captured in the snapshots.
+      #
+      # Exception handling: If a custom serializer's condition lambda raises
+      # an exception (e.g., regex match against invalid UTF-8 strings), the
+      # exception will be logged at WARN level, then the serializer will be
+      # skipped and the next serializer will be tried. This prevents custom
+      # serializers from breaking the entire serialization process.
       @@flat_registry = []
       def self.register(condition: nil, &block)
         @@flat_registry << {condition: condition, proc: block}
@@ -152,9 +158,28 @@ def serialize_value(value, name: nil,
           end
 
           @@flat_registry.each do |entry|
-            if (condition = entry[:condition]) && condition.call(value)
-              serializer_proc = entry.fetch(:proc)
-              return serializer_proc.call(self, value, name: nil, depth: depth)
+            condition = entry[:condition]
+            if condition
+              begin
+                condition_result = condition.call(value)
+              rescue => e
+                # If a custom serializer condition raises an exception (e.g., regex match
+                # against invalid UTF-8), skip it and continue with the next serializer.
+                # We don't want custom serializer conditions to break the entire serialization.
+                #
+                # Custom serializers may be defined by customers (in which case we should
+                # surface errors so they can fix their serializers) or they may be defined
+                # internally by dd-trace-rb (in which case we need to fix them). We use
+                # WARN level to surface these errors in either case.
+                Datadog.logger.warn("DI: Custom serializer condition failed: #{e.class}: #{e.message}")
+                telemetry&.report(e, description: "Custom serializer condition failed")
+                next
+              end
+
+              if condition_result
+                serializer_proc = entry.fetch(:proc)
+                return serializer_proc.call(self, value, name: nil, depth: depth)
+              end
             end
           end
 
@@ -184,13 +209,49 @@ def serialize_value(value, name: nil,
             else
               value.to_s
             end
+
+            # Handle binary strings and invalid UTF-8 by escaping to a JSON-safe format.
+            # Binary data (ASCII-8BIT encoding) and strings with invalid encoding are
+            # converted to an escaped string representation in the format: b'...'
+            # with special handling for:
+            # - Printable ASCII: preserved as-is
+            # - Special characters: \n, \t, \r, \\, \'
+            # - Non-printable bytes: \xHH hex escapes
+            #
+            # Example: "\x80\xFF".b -> "b'\\x80\\xff'"
+            #
+            # This produces the same serialized contents as dd-trace-py.
+            #
+            # For binary data, the max_capture_string_length limit is applied to the
+            # original binary data (in bytes) before escaping. This ensures correct
+            # truncation behavior - truncating after escaping would produce incorrect
+            # results (e.g., cutting mid-escape-sequence). The size field reports
+            # the original binary data length in bytes.
+            #
+            # For regular strings, the limit is applied to the string length in characters.
             max = settings.dynamic_instrumentation.max_capture_string_length
-            if value.length > max
-              serialized.update(truncated: true, size: value.length)
-              value = value[0...max]
-              need_dup = false
+
+            if value.encoding == Encoding::BINARY || !value.valid_encoding?
+              # Truncate binary data BEFORE escaping to avoid cutting mid-escape-sequence
+              # For invalid encodings, use bytesize instead of length to avoid encoding errors
+              original_size = value.bytesize
+              if original_size > max
+                serialized.update(truncated: true, size: original_size)
+                value = value.byteslice(0...max)
+              end
+              value = escape_binary_string(value) # steep:ignore ArgumentTypeMismatch
+              false # Already converted to a new string
+            else
+              # Truncate non-binary strings
+              if value.length > max
+                serialized.update(truncated: true, size: value.length)
+                value = value[0...max]
+                need_dup = false
+              end
+
+              value = value.dup if need_dup
             end
-            value = value.dup if need_dup
+
             serialized.update(value: value)
           when Array
             if depth < 0
@@ -417,6 +478,53 @@ def serialize_string_or_symbol_for_message(value)
           value
         end
       end
+
+      # Escapes a binary string or invalid UTF-8 string to a JSON-safe format.
+      #
+      # IMPORTANT: This method should ONLY be called with either:
+      # 1. True binary strings (encoding == Encoding::BINARY / ASCII-8BIT)
+      # 2. Strings with invalid encoding (!value.valid_encoding?)
+      #
+      # Calling this method with valid UTF-8 strings will produce incorrect output.
+      #
+      # Binary data (ASCII-8BIT encoding) or strings with invalid encoding are
+      # converted to an escaped string in the format: b'...' with hex escapes
+      # for non-printable bytes.
+      #
+      # The output format matches other Datadog tracer libraries for consistency
+      # across language implementations. The output is JSON-serializable.
+      #
+      # Examples:
+      #   "Hello".b -> "b'Hello'"
+      #   "\x80\xFF".b -> "b'\\x80\\xff'"
+      #   "\x80".force_encoding('UTF-8') -> "b'\\x80'" (invalid UTF-8)
+      #
+      # @param binary_string [String] A string with ASCII-8BIT encoding or invalid encoding
+      # @return [String] Escaped string with UTF-8 encoding
+      def escape_binary_string(binary_string)
+        result = +"b'"
+        binary_string.each_byte do |byte|
+          result << case byte
+          when 0x09 # \t
+            '\\t'
+          when 0x0A # \n
+            '\\n'
+          when 0x0D # \r
+            '\\r'
+          when 0x27 # '
+            "\\'"
+          when 0x5C # \
+            '\\\\'
+          when 0x20..0x7E # Printable ASCII (space through ~)
+            byte.chr
+          else
+            # Non-printable: use \xHH format
+            format('\\x%02x', byte)
+          end
+        end
+        result << "'"
+        result
+      end
     end
   end
 end

sig/datadog/di/serializer.rbs

@@ -36,6 +36,8 @@ module Datadog
       def class_name: (untyped cls) -> String
 
       def serialize_string_or_symbol_for_message: (untyped value) -> untyped
+
+      def escape_binary_string: (String binary_string) -> String
     end
   end
 end

spec/datadog/di/integration/instrumentation_spec.rb

@@ -38,6 +38,17 @@ def ivar_mutating_method
   def exception_method
     raise TestException, 'Test exception'
   end
+
+  def binary_data_method
+    # Return a string with high bytes that will fail JSON encoding
+    # 300 bytes to exceed default max_capture_string_length of 255
+    ((128..255).to_a * 3)[0...300].map { |i| i.chr(Encoding::BINARY) }.join.force_encoding(Encoding::BINARY)
+  end
+
+  def binary_data_param_method(binary_param, normal_param)
+    # Method with binary data in parameters
+    binary_param.length + normal_param.length
+  end
 end
 
 RSpec.describe 'Instrumentation integration' do
@@ -1339,6 +1350,151 @@ def run_test
       end
     end
   end
+
+  context 'binary data in snapshots' do
+    context 'with binary data in parameters' do
+      let(:probe) do
+        Datadog::DI::Probe.new(
+          id: "binary-test",
+          type: :log,
+          type_name: 'InstrumentationSpecTestClass',
+          method_name: 'binary_data_param_method',
+          capture_snapshot: true
+        )
+      end
+
+      let(:binary_string) { "\x80\x81\x82\xFF\xFE".b }
+
+      it 'successfully sends snapshot with binary data through transport' do
+        expect(diagnostics_transport).to receive(:send_diagnostics)
+
+        # Capture the snapshot that goes through transport
+        captured_snapshot = nil
+        json_encoded = nil
+
+        allow(component.probe_notifier_worker).to receive(:add_snapshot).and_wrap_original do |m, *args|
+          captured_snapshot = args[0]
+          m.call(*args)
+        end
+
+        allow(input_transport).to receive(:send_input) do |snapshots, tags|
+          # This mimics what the transport does - encode to JSON
+          json_encoded = JSON.dump(snapshots)
+        end
+
+        probe_manager.add_probe(probe)
+
+        # Execute the method with binary data
+        result = InstrumentationSpecTestClass.new.binary_data_param_method(binary_string, "hello")
+        expect(result).to eq(10) # 5 + 5
+
+        # Wait for flush to complete
+        component.probe_notifier_worker.flush
+
+        # Verify the snapshot was captured
+        expect(captured_snapshot).not_to be_nil
+
+        # JSON encoding should now succeed with escaped binary data
+        expect {
+          JSON.dump(captured_snapshot)
+        }.not_to raise_error
+
+        # Transport should have successfully encoded it
+        expect(json_encoded).to be_a(String)
+        expect(json_encoded.encoding).to eq(Encoding::UTF_8)
+      end
+
+      it 'escapes binary data in parameters' do
+        expect(diagnostics_transport).to receive(:send_diagnostics)
+
+        # Capture the snapshot before it gets to transport
+        captured_snapshot = nil
+        allow(component.probe_notifier_worker).to receive(:add_snapshot) do |snapshot|
+          captured_snapshot = snapshot
+        end
+
+        probe_manager.add_probe(probe)
+
+        # Execute the method
+        InstrumentationSpecTestClass.new.binary_data_param_method(binary_string, "hello")
+
+        # Wait for flush to complete
+        component.probe_notifier_worker.flush
+
+        # Verify snapshot was captured with binary data escaped
+        expect(captured_snapshot).not_to be_nil
+        expect(captured_snapshot[:debugger][:snapshot][:captures]).to have_key(:entry)
+
+        entry_capture = captured_snapshot[:debugger][:snapshot][:captures][:entry]
+        expect(entry_capture[:arguments]).to have_key(:arg1)
+
+        # The binary string is escaped to b'...' format
+        binary_param_value = entry_capture[:arguments][:arg1][:value]
+        expect(binary_param_value).to be_a(String)
+        expect(binary_param_value).to eq("b'\\x80\\x81\\x82\\xff\\xfe'")
+        expect(binary_param_value.encoding).to eq(Encoding::UTF_8)
+
+        # JSON encoding the snapshot should now succeed
+        expect {
+          JSON.dump(captured_snapshot)
+        }.not_to raise_error
+      end
+    end
+
+    context 'with binary return value' do
+      let(:probe) do
+        Datadog::DI::Probe.new(
+          id: "binary-return-test",
+          type: :log,
+          type_name: 'InstrumentationSpecTestClass',
+          method_name: 'binary_data_method',
+          capture_snapshot: true
+        )
+      end
+
+      it 'escapes binary return value' do
+        expect(diagnostics_transport).to receive(:send_diagnostics)
+
+        # Capture the snapshot before transport
+        captured_snapshot = nil
+        allow(component.probe_notifier_worker).to receive(:add_snapshot) do |snapshot|
+          captured_snapshot = snapshot
+        end
+
+        probe_manager.add_probe(probe)
+
+        # Execute the method that returns binary data
+        result = InstrumentationSpecTestClass.new.binary_data_method
+        expect(result.encoding).to eq(Encoding::BINARY)
+        expect(result.length).to eq(300)
+        expect(result.bytes.min).to eq(128)
+        expect(result.bytes.max).to eq(255)
+
+        # Wait for flush to complete
+        component.probe_notifier_worker.flush
+
+        # Verify snapshot captured the return value as escaped string
+        expect(captured_snapshot).not_to be_nil
+        return_capture = captured_snapshot[:debugger][:snapshot][:captures][:return]
+        expect(return_capture[:arguments]).to have_key(:@return)
+
+        return_value = return_capture[:arguments][:@return][:value]
+        expect(return_value).to start_with("b'")
+        expect(return_value.encoding).to eq(Encoding::UTF_8)
+        expect(return_value).to include('\\x80') # First high byte
+
+        # The 300-byte binary string exceeds max_capture_string_length (255)
+        # Truncated to first 255 bytes, then escaped to 1023 chars (b' + 255*4 + ')
+        expect(return_capture[:arguments][:@return][:truncated]).to be true
+        expect(return_capture[:arguments][:@return][:size]).to eq(300) # Original byte count
+
+        # JSON encoding should now succeed
+        expect {
+          JSON.dump(captured_snapshot)
+        }.not_to raise_error
+      end
+    end
+  end
 end
 
 # rubocop:enable Style/RescueModifier