Add a pre_sampling option to metrics methods

This allows sampling decisions to be made outside of metrics emission
but still have the sampling rate sent to the collector. If pre_sampling
is false/not set the existing behavior of sampling in the library is
used. When pre_sampling is true metrics are always sent regardless of
the sample rate. In this mode it is up to the caller to handle sampling.

lib/datadog/statsd.rb

@@ -151,6 +151,7 @@ def self.open(*args)
     # @param [String] stat stat name
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @option opts [Numeric] :by increment value, default 1
     # @see #count
@@ -165,6 +166,7 @@ def increment(stat, opts = EMPTY_OPTIONS)
     # @param [String] stat stat name
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @option opts [Numeric] :by decrement value, default 1
     # @see #count
@@ -180,6 +182,7 @@ def decrement(stat, opts = EMPTY_OPTIONS)
     # @param [Integer] count count
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     def count(stat, count, opts = EMPTY_OPTIONS)
       opts = { sample_rate: opts } if opts.is_a?(Numeric)
@@ -196,6 +199,7 @@ def count(stat, count, opts = EMPTY_OPTIONS)
     # @param [Numeric] value gauge value.
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @example Report the current user count:
     #   $statsd.gauge('user.count', User.count)
@@ -210,6 +214,7 @@ def gauge(stat, value, opts = EMPTY_OPTIONS)
     # @param [Numeric] value histogram value.
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @example Report the current user count:
     #   $statsd.histogram('user.count', User.count)
@@ -223,6 +228,7 @@ def histogram(stat, value, opts = EMPTY_OPTIONS)
     # @param [Numeric] value distribution value.
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @example Report the current user count:
     #   $statsd.distribution('user.count', User.count)
@@ -239,6 +245,7 @@ def distribution(stat, value, opts = EMPTY_OPTIONS)
     # @param [Integer] ms timing in milliseconds
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     def timing(stat, ms, opts = EMPTY_OPTIONS)
       opts = { sample_rate: opts } if opts.is_a?(Numeric)
@@ -253,6 +260,7 @@ def timing(stat, ms, opts = EMPTY_OPTIONS)
     # @param [String] stat stat name
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @yield The operation to be timed
     # @see #timing
@@ -272,6 +280,7 @@ def time(stat, opts = EMPTY_OPTIONS)
     # @param [Numeric] value set value.
     # @param [Hash] opts the options to create the metric with
     # @option opts [Numeric] :sample_rate sample rate, 1 for always
+    # @option opts [Boolean] :pre_sampled If true, the client assumes the caller has already sampled metrics at :sample_rate, and doesn't perform sampling.
     # @option opts [Array<String>] :tags An array of tags
     # @example Record a unique visitory by id:
     #   $statsd.set('visitors.uniques', User.id)
@@ -394,7 +403,7 @@ def send_stats(stat, delta, type, opts = EMPTY_OPTIONS)
 
       sample_rate = opts[:sample_rate] || @sample_rate || 1
 
-      if sample_rate == 1 || rand <= sample_rate
+      if sample_rate == 1 || opts[:pre_sampled] || rand <= sample_rate
         full_stat = serializer.to_stat(stat, delta, type, tags: opts[:tags], sample_rate: sample_rate)
 
         forwarder.send_message(full_stat)

spec/statsd_spec.rb

@@ -271,6 +271,19 @@
       end
     end
 
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.increment('foobar', sample_rate: 0.5, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'foobar:1|c|@0.5'
+      end
+    end
+
     context 'with a increment by' do
       it 'increments by the number given' do
         subject.increment('foobar', by: 5)
@@ -326,6 +339,19 @@
       end
     end
 
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.decrement('foobar', sample_rate: 0.5, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'foobar:-1|c|@0.5'
+      end
+    end
+
     context 'with a decrement by' do
       it 'decrements by the number given' do
         subject.decrement('foobar', by: 5)
@@ -367,6 +393,19 @@
         expect(socket.recv[0]).to eq_with_telemetry 'foobar:123|c|@0.1'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.count('foobar', 123, sample_rate: 0.1, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'foobar:123|c|@0.1'
+      end
+    end
   end
 
   describe '#gauge' do
@@ -425,6 +464,19 @@
         expect(socket.recv[0]).to eq_with_telemetry 'begrutten-suffusion:536|g|@0.1'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.gauge('begrutten-suffusion', 536, sample_rate: 0.1, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'begrutten-suffusion:536|g|@0.1'
+      end
+    end
   end
 
   describe '#histogram' do
@@ -470,6 +522,19 @@
         expect(socket.recv[0]).to eq_with_telemetry 'begrutten-suffusion:536|g|@0.1'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.histogram('ohmy', 536, sample_rate: 0.1, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'ohmy:536|h|@0.1'
+      end
+    end
   end
 
   describe '#set' do
@@ -516,6 +581,19 @@
         expect(socket.recv[0]).to eq_with_telemetry 'my.set:536|s|@0.5'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.set('my.set', 536, sample_rate: 0.5, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'my.set:536|s|@0.5'
+      end
+    end
   end
 
   describe '#timing' do
@@ -562,6 +640,19 @@
         expect(socket.recv[0]).to eq_with_telemetry 'foobar:500|ms|@0.5'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.timing('foobar', 500, sample_rate: 0.5, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'foobar:500|ms|@0.5'
+      end
+    end
   end
 
   describe '#time' do
@@ -673,6 +764,23 @@
         expect(socket.recv[0]).to eq_with_telemetry 'foobar:1000|ms|@0.5'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.time('foobar', sample_rate: 0.5, pre_sampled: true) do
+          Timecop.travel(after_date)
+          allow(Process).to receive(:clock_gettime).and_return(1)
+        end
+
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'foobar:1000|ms|@0.5'
+      end
+    end
   end
 
   describe '#distribution' do
@@ -706,6 +814,19 @@
         expect(socket.recv[0]).to eq_with_telemetry 'begrutten-suffusion:536|d|@0.5'
       end
     end
+
+    context 'with pre sampling' do
+      before do
+        allow(subject).to receive(:rand).and_return(1)
+      end
+
+      it 'sends the sample rate without additional sampling' do
+        subject.distribution('begrutten-suffusion', 536, sample_rate: 0.5, pre_sampled: true)
+        subject.flush(sync: true)
+
+        expect(socket.recv[0]).to eq_with_telemetry 'begrutten-suffusion:536|d|@0.5'
+      end
+    end
   end
 
   describe '#event' do