notify emitting only once

lib/datadog/di/probe.rb

@@ -71,6 +71,8 @@ def initialize(id:, type:,
 
         @rate_limit = rate_limit || (@capture_snapshot ? 1 : 5000)
         @rate_limiter = Datadog::Core::TokenBucket.new(@rate_limit)
+
+        @emitting_notified = false
       end
 
       attr_reader :id
@@ -157,6 +159,11 @@ def file_matches?(path)
       # Line trace point for line probes. Normally this would be a targeted
       # trace point.
       attr_accessor :instrumentation_trace_point
+
+      attr_writer :emitting_notified
+      def emitting_notified?
+        !!@emitting_notified
+      end
     end
   end
 end

lib/datadog/di/probe_manager.rb

@@ -136,9 +136,12 @@ def install_pending_line_probes(file)
       end
 
       def probe_executed_callback(probe:, **opts)
-        # TODO notify emitting once only
-        payload = probe_notification_builder.build_emitting(probe)
-        probe_notifier_worker.add_status(payload)
+        unless probe.emitting_notified?
+          payload = probe_notification_builder.build_emitting(probe)
+          probe_notifier_worker.add_status(payload)
+          probe.emitting_notified = true
+        end
+
         payload = probe_notification_builder.build_executed(probe, **opts)
         probe_notifier_worker.add_snapshot(payload)
       end

spec/datadog/di/integration/instrumentation_spec.rb

@@ -126,6 +126,79 @@ def run_test
           end
         end
       end
+
+      context 'when target is invoked' do
+        let(:probe) do
+          Datadog::DI::Probe.new(id: "1234", type: :log,
+            type_name: 'InstrumentationSpecTestClass', method_name: 'test_method')
+        end
+
+        let(:expected_installed_payload) do
+          {ddsource: 'dd_debugger',
+            debugger: {
+              diagnostics: {
+                parentId: nil,
+                probeId: String,
+                probeVersion: 0,
+                runtimeId: String,
+                status: 'INSTALLED',
+              }},
+            message: String,
+            service: 'rspec',
+            timestamp: Integer,
+          }
+        end
+
+        let(:expected_emitting_payload) do
+          {ddsource: 'dd_debugger',
+            debugger: {
+              diagnostics: {
+                parentId: nil,
+                probeId: String,
+                probeVersion: 0,
+                runtimeId: String,
+                status: 'EMITTING',
+              }},
+            message: String,
+            service: 'rspec',
+            timestamp: Integer,
+          }
+        end
+
+        it 'notifies agent that probe is emitting' do
+          expect(component.probe_notifier_worker).to receive(:add_status) do |status|
+            expect(status).to match(expected_installed_payload)
+          end
+          probe_manager.add_probe(probe)
+          expect(component.probe_notifier_worker).to receive(:add_status) do |status|
+            expect(status).to match(expected_emitting_payload)
+          end
+          allow(component.probe_notifier_worker).to receive(:add_snapshot)
+          expect(InstrumentationSpecTestClass.new.test_method).to eq(42)
+          component.probe_notifier_worker.flush
+        end
+
+        context 'when target is invoked multiple times' do
+          it 'notifies that probe is emitting only once at first invocation' do
+            expect(component.probe_notifier_worker).to receive(:add_status) do |status|
+              expect(status).to match(expected_installed_payload)
+            end
+            probe_manager.add_probe(probe)
+
+            expect(component.probe_notifier_worker).to receive(:add_status) do |status|
+              expect(status).to match(expected_emitting_payload)
+            end
+            expect(component.probe_notifier_worker).to receive(:add_snapshot)
+            expect(InstrumentationSpecTestClass.new.test_method).to eq(42)
+            component.probe_notifier_worker.flush
+
+            expect(component.probe_notifier_worker).not_to receive(:add_status)
+            expect(component.probe_notifier_worker).to receive(:add_snapshot)
+            expect(InstrumentationSpecTestClass.new.test_method).to eq(42)
+            component.probe_notifier_worker.flush
+          end
+        end
+      end
     end
 
     context 'line probe' do