DEBUG-2334 Dynamic instrumentation transport component (#3981)

lib/datadog/di/transport.rb

@@ -0,0 +1,67 @@
+# frozen_string_literal: true
+
+require_relative 'error'
+
+module Datadog
+  module DI
+    # Transport for sending probe statuses and snapshots to local agent.
+    #
+    # Handles encoding of the payloads into multipart posts if necessary,
+    # body formatting/encoding, setting correct headers, etc.
+    #
+    # The transport does not handle batching of statuses or snapshots -
+    # the batching should be implemented upstream of this class.
+    #
+    # Timeout settings are forwarded from agent settings to the Net adapter.
+    #
+    # The send_* methods raise Error::AgentCommunicationError on errors
+    # (network errors and HTTP protocol errors). It is the responsibility
+    # of upstream code to rescue these exceptions appropriately to prevent them
+    # from being propagated to the application.
+    #
+    # @api private
+    class Transport
+      DIAGNOSTICS_PATH = '/debugger/v1/diagnostics'
+      INPUT_PATH = '/debugger/v1/input'
+
+      def initialize(agent_settings)
+        # Note that this uses host, port, timeout and TLS flag from
+        # agent settings.
+        @client = Core::Transport::HTTP::Adapters::Net.new(agent_settings)
+      end
+
+      def send_diagnostics(payload)
+        event_payload = Core::Vendor::Multipart::Post::UploadIO.new(
+          StringIO.new(JSON.dump(payload)), 'application/json', 'event.json'
+        )
+        payload = {'event' => event_payload}
+        send_request('Probe status submission', DIAGNOSTICS_PATH, payload)
+      end
+
+      def send_input(payload)
+        send_request('Probe snapshot submission', INPUT_PATH, payload,
+          headers: {'content-type' => 'application/json'},)
+      end
+
+      private
+
+      attr_reader :client
+
+      def send_request(desc, path, payload, headers: {})
+        # steep:ignore:start
+        env = OpenStruct.new(
+          path: path,
+          form: payload,
+          headers: headers,
+        )
+        # steep:ignore:end
+        response = client.post(env)
+        unless response.ok?
+          raise Error::AgentCommunicationError, "#{desc} failed: #{response.code}: #{response.payload}"
+        end
+      rescue IOError, SystemCallError => exc
+        raise Error::AgentCommunicationError, "#{desc} failed: #{exc.class}: #{exc}"
+      end
+    end
+  end
+end

sig/datadog/di/transport.rbs

@@ -0,0 +1,23 @@
+module Datadog
+  module DI
+    class Transport
+      @client: untyped
+
+      DIAGNOSTICS_PATH: "/debugger/v1/diagnostics"
+
+      INPUT_PATH: "/debugger/v1/input"
+
+      def initialize: (untyped agent_settings) -> void
+
+      def send_diagnostics: (Hash[untyped,untyped] payload) -> untyped
+
+      def send_input: (Hash[untyped,untyped] payload) -> untyped
+
+      private
+
+      attr_reader client: untyped
+
+      def send_request: (String desc, String path, Hash[untyped,untyped] payload, ?headers: ::Hash[untyped, untyped]) -> void
+    end
+  end
+end

spec/datadog/di/transport_spec.rb

@@ -0,0 +1,108 @@
+require "datadog/di/spec_helper"
+require "datadog/di/transport"
+
+RSpec.describe Datadog::DI::Transport do
+  di_test
+
+  let(:agent_settings) do
+    instance_double(Datadog::Core::Configuration::AgentSettingsResolver::AgentSettings)
+  end
+
+  describe '.new' do
+    it 'creates an instance using agent settings' do
+      expect(agent_settings).to receive(:hostname).and_return('localhost')
+      expect(agent_settings).to receive(:port).and_return(8126)
+      expect(agent_settings).to receive(:timeout_seconds).and_return(1)
+      expect(agent_settings).to receive(:ssl).and_return(false)
+
+      expect(described_class.new(agent_settings)).to be_a(described_class)
+    end
+  end
+
+  # These are fairly basic tests. The agent will accept all kinds of
+  # semantically nonsensical payloads. The tests here are useful to
+  # ascertain that things like content type is set correctly for each
+  # endpoint.
+  #
+  # Realistically, the only test that can check that the payload being
+  # sent is the correct one is a system test.
+  describe 'send methods' do
+    before(:all) do
+      # These tests require a functional datadog agent running at the
+      # configured (via agent_host & agent_port) location.
+      # CI has "dd-apm-test-agent" running which does not implement
+      # debugger endpoints, and thus is not suitable for these tests.
+      # These tests can be run locally, and test coverage in CI is
+      # accomplished via system tests.
+      unless agent_host && agent_port && ENV['TEST_DATADOG_AGENT'] == '1'
+        skip "Set TEST_DATADOG_AGENT=1, DD_AGENT_HOST and DD_TRACE_AGENT_PORT in environment to run these tests"
+      end
+    end
+
+    let(:port) { agent_port }
+
+    before do
+      expect(agent_settings).to receive(:hostname).and_return(agent_host)
+      expect(agent_settings).to receive(:port).and_return(port)
+      expect(agent_settings).to receive(:timeout_seconds).and_return(1)
+      expect(agent_settings).to receive(:ssl).and_return(false)
+    end
+
+    let(:client) do
+      described_class.new(agent_settings)
+    end
+
+    describe '.send_diagnostics' do
+      let(:payload) do
+        {}
+      end
+
+      it 'does not raise exceptions' do
+        expect do
+          client.send_diagnostics(payload)
+        end.not_to raise_exception
+      end
+    end
+
+    describe '.send_input' do
+      let(:payload) do
+        {}
+      end
+
+      it 'does not raise exceptions' do
+        expect do
+          client.send_input(payload)
+        end.not_to raise_exception
+      end
+    end
+
+    context 'when agent is not listening' do
+      # Use a bogus port
+      let(:port) { 99999 }
+
+      describe '.send_diagnostics' do
+        let(:payload) do
+          {}
+        end
+
+        it 'raises AgentCommunicationError' do
+          expect do
+            client.send_diagnostics(payload)
+          end.to raise_exception(Datadog::DI::Error::AgentCommunicationError, /(?:Connection refused|connect).*99999/)
+        end
+      end
+
+      describe '.send_input' do
+        let(:payload) do
+          {}
+        end
+
+        it 'raises AgentCommunicationError' do
+          expect do
+            client.send_input(payload)
+          end.to raise_exception(Datadog::DI::Error::AgentCommunicationError, /(?:Connection refused|connect).*99999/)
+        end
+      end
+    end
+  end
+end