Repository: beam-telemetry/telemetry_metrics_statsd
Pull request #28 (2020-04-08 15:18:40)

Add support for sampling rate
telemetry_metrics recently added `:reporter_options` ([telemetry_metrics#52](https://github.com/beam-telemetry/telemetry_metrics/pull/52)) initiated from a request to add [extra metrics options](https://github.com/beam-telemetry/telemetry_metrics/issues/33).

This PR allows developers to pass `[sample_rate: <rate>]` (where "<rate>" is a value between 0.0 and 1.0) to `:reporter_options` to limit how many measurements to capture.

samullen: I've updated both the standard and datadog formatters to include sampling rate. Not terribly happy with the duplicate use of `Keyword.get(reporter_options, :sampling_rate, 1.0)`, but ¯\_(ツ)_/¯
arkgil: Ah, @samullen, it is failing on OTP 19, because `:rand.uniform_real` is not available in that release. Could you please switch to `:rand.uniform` (unless there is a reason why we can't use it)?

Patch:
diff --git a/lib/telemetry_metrics_statsd.ex b/lib/telemetry_metrics_statsd.ex
index 5ec8ed3..ceea889 100644
--- a/lib/telemetry_metrics_statsd.ex
+++ b/lib/telemetry_metrics_statsd.ex
@@ -222,6 +222,27 @@ defmodule TelemetryMetricsStatsd do
   exceed the Maximum Transmission Unit, or MTU, of the link, so that no data is lost on the way.
   By default the reporter will break up the datagrams at 512 bytes, but this is configurable via
   the `:mtu` option.
+
+  ## Sampling data
+
+  It's not always convenient to capture every piece of data, such as in the
+  case of high-traffic applications. In those cases, you may want to capture a
+  "sample" of the data. You can do this by passing `[sampling_rate: <rate>]` as
+  an option to `:reporter_options`, where "rate" is a value between 0.0 and
+  1.0. The default `:sampling_rate` is 1.0.
+
+  ### Example
+
+      TelemetryMetricsStatsd.start_link(
+        metrics: [
+          counter("http.request.count"),
+          summary("http.request.duration", reporter_options: [sampling_rate: 0.1]),
+          distribution("http.request.duration", buckets: [25, 50, 100, 250], reporter_options: [sampling_rate: 0.1])
+        ]
+      )
+
+  In this example, we are capturing 100% of the measurements for the counter,
+  but only 10% for both summary and distribution.
   """
 
   use GenServer
@@ -231,7 +252,7 @@ defmodule TelemetryMetricsStatsd do
   alias Telemetry.Metrics
   alias TelemetryMetricsStatsd.{EventHandler, UDP}
 
-  @type prefix :: String.t() | nil
+  @type prefix :: String.t() | atom() | nil
   @type host :: String.t() | :inet.ip_address()
   @type option ::
           {:port, :inet.port_number()}
diff --git a/lib/telemetry_metrics_statsd/event_handler.ex b/lib/telemetry_metrics_statsd/event_handler.ex
index 0a2a72f..a7685e6 100644
--- a/lib/telemetry_metrics_statsd/event_handler.ex
+++ b/lib/telemetry_metrics_statsd/event_handler.ex
@@ -86,14 +86,20 @@ defmodule TelemetryMetricsStatsd.EventHandler do
 
   @spec fetch_measurement(Metrics.t(), :telemetry.event_measurements()) ::
           {:ok, number()} | :error
-  defp fetch_measurement(%Metrics.Counter{}, _measurements) do
+  defp fetch_measurement(%Metrics.Counter{} = metric, _measurements) do
     # For counter, we can ignore the measurements and just use 0.
-    {:ok, 0}
+    case sample(metric) do
+      nil -> :error
+      _ -> {:ok, 0}
+    end
   end
 
   defp fetch_measurement(metric, measurements) do
     value =
-      case metric.measurement do
+      case sample(metric) do
+        nil ->
+          nil
+
         fun when is_function(fun, 1) ->
           fun.(measurements)
 
@@ -123,4 +129,14 @@ defmodule TelemetryMetricsStatsd.EventHandler do
       end
     end)
   end
+
+  @spec sample(Metrics.t()) :: Metrics.measurement() | nil
+  defp sample(metric) do
+    rate = Keyword.get(metric.reporter_options, :sampling_rate, 1.0)
+    sample(metric, rate, :rand.uniform())
+  end
+
+  defp sample(metric, 1.0, _random), do: metric.measurement
+  defp sample(metric, rate, random) when rate >= random, do: metric.measurement
+  defp sample(_metric, _rate, _random_real), do: nil
 end
diff --git a/lib/telemetry_metrics_statsd/formatter.ex b/lib/telemetry_metrics_statsd/formatter.ex
index d72ee9c..c48df01 100644
--- a/lib/telemetry_metrics_statsd/formatter.ex
+++ b/lib/telemetry_metrics_statsd/formatter.ex
@@ -26,7 +26,7 @@ defmodule TelemetryMetricsStatsd.Formatter do
   end
 
   def format(formatter, metric, prefix, measurement, tags) do
-    [prefix, ?., formatter.format(metric, measurement, tags)]
+    [to_string(prefix), ?., formatter.format(metric, measurement, tags)]
     |> :erlang.iolist_to_binary()
   end
 end
diff --git a/lib/telemetry_metrics_statsd/formatter/datadog.ex b/lib/telemetry_metrics_statsd/formatter/datadog.ex
index 80d5920..58411d7 100644
--- a/lib/telemetry_metrics_statsd/formatter/datadog.ex
+++ b/lib/telemetry_metrics_statsd/formatter/datadog.ex
@@ -11,6 +11,7 @@ defmodule TelemetryMetricsStatsd.Formatter.Datadog do
       format_metric_name(metric.name),
       ?:,
       format_metric_value(metric, value),
+      format_sampling_rate(metric.reporter_options),
       format_metric_tags(tags)
     ]
   end
@@ -67,4 +68,11 @@ defmodule TelemetryMetricsStatsd.Formatter.Datadog do
   defp format_tag(k, v) do
     [:erlang.atom_to_binary(k, :utf8), ?:, to_string(v)]
   end
+
+  defp format_sampling_rate(reporter_options) do
+    case Keyword.get(reporter_options, :sampling_rate, 1.0) do
+      rate when rate > 0.0 and rate < 1.0 -> "|@#{rate}"
+      _ -> ""
+    end
+  end
 end
diff --git a/lib/telemetry_metrics_statsd/formatter/standard.ex b/lib/telemetry_metrics_statsd/formatter/standard.ex
index 2a307cc..1141451 100644
--- a/lib/telemetry_metrics_statsd/formatter/standard.ex
+++ b/lib/telemetry_metrics_statsd/formatter/standard.ex
@@ -11,7 +11,8 @@ defmodule TelemetryMetricsStatsd.Formatter.Standard do
       format_metric_name(metric.name),
       format_metric_tags(tags),
       ?:,
-      format_metric_value(metric, value)
+      format_metric_value(metric, value),
+      format_sampling_rate(metric.reporter_options)
     ]
   end
 
@@ -51,4 +52,11 @@ defmodule TelemetryMetricsStatsd.Formatter.Standard do
 
   defp format_metric_value(%Metrics.Sum{}, value),
     do: [value |> round() |> :erlang.integer_to_binary(), "|g"]
+
+  defp format_sampling_rate(reporter_options) do
+    case Keyword.get(reporter_options, :sampling_rate, 1.0) do
+      rate when rate > 0.0 and rate < 1.0 -> "|@#{rate}"
+      _ -> ""
+    end
+  end
 end


Test patch:
diff --git a/test/telemetry_metrics_statsd/formatter/datadog_test.exs b/test/telemetry_metrics_statsd/formatter/datadog_test.exs
index 6088850..0328771 100644
--- a/test/telemetry_metrics_statsd/formatter/datadog_test.exs
+++ b/test/telemetry_metrics_statsd/formatter/datadog_test.exs
@@ -72,6 +72,18 @@ defmodule TelemetryMetricsStatsd.Formatter.DatadogTest do
              "my.awesome.metric:131.5|g"
   end
 
+  test "sampling rate is added to third field" do
+    m = given_last_value("my.awesome.metric", reporter_options: [sampling_rate: 0.2])
+
+    assert format(m, 131.4, []) == "my.awesome.metric:131.4|g|@0.2"
+  end
+
+  test "sampling rate is ignored if == 1.0" do
+    m = given_last_value("my.awesome.metric", reporter_options: [sampling_rate: 1.0])
+
+    assert format(m, 131.4, []) == "my.awesome.metric:131.4|g"
+  end
+
   defp format(metric, value, tags) do
     Datadog.format(metric, value, tags)
     |> :erlang.iolist_to_binary()
diff --git a/test/telemetry_metrics_statsd/formatter/standard_test.exs b/test/telemetry_metrics_statsd/formatter/standard_test.exs
index cca7d44..5bce494 100644
--- a/test/telemetry_metrics_statsd/formatter/standard_test.exs
+++ b/test/telemetry_metrics_statsd/formatter/standard_test.exs
@@ -72,6 +72,18 @@ defmodule TelemetryMetricsStatsd.Formatter.StandardTest do
              "my.awesome.metric:132|g"
   end
 
+  test "sampling rate is added to third field" do
+    m = given_last_value("my.awesome.metric", reporter_options: [sampling_rate: 0.2])
+
+    assert format(m, 131.4, []) == "my.awesome.metric:131|g|@0.2"
+  end
+
+  test "sampling rate is ignored if == 1.0" do
+    m = given_last_value("my.awesome.metric", reporter_options: [sampling_rate: 1.0])
+
+    assert format(m, 131.4, []) == "my.awesome.metric:131|g"
+  end
+
   defp format(metric, value, tags) do
     Standard.format(metric, value, tags)
     |> :erlang.iolist_to_binary()
diff --git a/test/telemetry_metrics_statsd/formatter_test.exs b/test/telemetry_metrics_statsd/formatter_test.exs
index fbc611b..267ad4e 100644
--- a/test/telemetry_metrics_statsd/formatter_test.exs
+++ b/test/telemetry_metrics_statsd/formatter_test.exs
@@ -14,6 +14,9 @@ defmodule TelemetryMetricsStatsd.FormatterTest do
 
     assert Formatter.format(Datadog, metric, "api", 1, resource: "users", method: "GET") ==
              "api.http.duration:1|c|#resource:users,method:GET"
+
+    assert Formatter.format(Datadog, metric, :api, 1, resource: "users", method: "GET") ==
+             "api.http.duration:1|c|#resource:users,method:GET"
   end
 
   test "formats the measurement using provided formatter implementation without prefix" do
diff --git a/test/telemetry_metrics_statsd_test.exs b/test/telemetry_metrics_statsd_test.exs
index 5e35821..c09760e 100644
--- a/test/telemetry_metrics_statsd_test.exs
+++ b/test/telemetry_metrics_statsd_test.exs
@@ -421,6 +421,72 @@ defmodule TelemetryMetricsStatsdTest do
     refute_reported(socket)
   end
 
+  test "doesn't report data for Counter metric when outside sample rate" do
+    {socket, port} = given_udp_port_opened()
+
+    counter =
+      given_counter("http.requests",
+        event_name: "http.request",
+        reporter_options: [sampling_rate: 0.1]
+      )
+
+    # :rand.uniform_real will return 0.3280001173553174
+    :rand.seed(:exs1024, {1, 2, 2})
+
+    start_reporter(metrics: [counter], port: port)
+
+    :telemetry.execute([:http, :request], %{sample: 42})
+
+    refute_reported(socket)
+  end
+
+  test "doesn't report data when non-Counter metric outside sample rate" do
+    {socket, port} = given_udp_port_opened()
+    sum = given_sum("http.request.sample", reporter_options: [sampling_rate: 0.1])
+
+    # :rand.uniform_real will return 0.3280001173553174
+    :rand.seed(:exs1024, {1, 2, 2})
+
+    start_reporter(metrics: [sum], port: port)
+
+    :telemetry.execute([:http, :request], %{sample: 42})
+
+    refute_reported(socket)
+  end
+
+  test "report data for Counter metric when inside sample rate" do
+    {socket, port} = given_udp_port_opened()
+
+    counter =
+      given_counter("http.requests",
+        event_name: "http.request",
+        reporter_options: [sampling_rate: 0.1]
+      )
+
+    # :rand.uniform_real will return 0.06907625299228148
+    :rand.seed(:exs1024, {1, 2, 3})
+
+    start_reporter(metrics: [counter], port: port)
+
+    :telemetry.execute([:http, :request], %{sample: 42})
+
+    assert_reported(socket, "http.requests:1|c|@0.1")
+  end
+
+  test "report data when non-Counter metric inside sample rate" do
+    {socket, port} = given_udp_port_opened()
+    sum = given_sum("http.request.sample", reporter_options: [sampling_rate: 0.1])
+
+    # :rand.uniform_real will return 0.06907625299228148
+    :rand.seed(:exs1024, {1, 2, 3})
+
+    start_reporter(metrics: [sum], port: port)
+
+    :telemetry.execute([:http, :request], %{sample: 42})
+
+    assert_reported(socket, "http.request.sample:+42|g|@0.1")
+  end
+
   defp given_udp_port_opened() do
     {:ok, socket} = :gen_udp.open(0, [:binary, active: false])
     {:ok, port} = :inet.port(socket)