Repository: beam-telemetry/telemetry_metrics_statsd
Pull request #54 (2021-03-19 21:31:19)

Handle empty socket pool
If the error is reported for each socket in the pool, the table becomes
empty and no socket is available.

Previously this resulted in `Enum.EmptyError` because we called
`Enum.random` on an empty list. Now we log an error and drop the packets.


Patch:
diff --git a/lib/telemetry_metrics_statsd.ex b/lib/telemetry_metrics_statsd.ex
index d803e19..8d4dc05 100644
--- a/lib/telemetry_metrics_statsd.ex
+++ b/lib/telemetry_metrics_statsd.ex
@@ -376,10 +376,8 @@ defmodule TelemetryMetricsStatsd do
     can be overriden by tags sent via the `:telemetry.execute` call.
   * `:pool_size` - The number of UDP sockets to open to report metrics. Defaults to `1`.
   * `:host_resolution_interval` - When set, makes the reporter resolve the hostname of the StatsD server
-    on a specified interval instead of looking up the hostname on every packet send using the system DNS
-    stack. This can improve the performance of publishing metrics while still allowing for dynamic
-    hostname resolution. Only applies when `:host` is a string and no an IP address. If the provided hostname
-    resolves to multiple IP addresses, the first one is used.
+    on a specified interval instead of looking it up once on start. Only applies when `:host` is a string
+    and no an IP address. If the provided hostname resolves to multiple IP addresses, the first one is used.
 
 
   You can read more about all the options in the `TelemetryMetricsStatsd` module documentation.
@@ -419,13 +417,18 @@ defmodule TelemetryMetricsStatsd do
   end
 
   @doc false
-  @spec get_udp(:ets.tid()) :: UDP.t()
+  @spec get_udp(:ets.tid()) :: {:ok, UDP.t()} | :error
   def get_udp(pool_id) do
-    udps = :ets.lookup(pool_id, :udp)
-
-    udps
-    |> Enum.random()
-    |> elem(1)
+    # The table can be empty if the UDP error is reported for all the sockets.
+    case :ets.lookup(pool_id, :udp) do
+      [] ->
+        Logger.error("Failed to publish metrics over UDP: no open sockets available.")
+        :error
+
+      udps ->
+        {:udp, udp} = Enum.random(udps)
+        {:ok, udp}
+    end
   end
 
   @doc false
@@ -574,7 +577,8 @@ defmodule TelemetryMetricsStatsd do
   end
 
   defp configure_host_resolution(%{host: host, port: port}) do
-    %{host: host, port: port}
+    {:ok, hostent(h_addr_list: [ip | _ips])} = :inet.gethostbyname(host)
+    %{host: ip, port: port}
   end
 
   defp update_pool(pool_id, new_host, new_port) do
diff --git a/lib/telemetry_metrics_statsd/event_handler.ex b/lib/telemetry_metrics_statsd/event_handler.ex
index e6bb18f..e2934c5 100644
--- a/lib/telemetry_metrics_statsd/event_handler.ex
+++ b/lib/telemetry_metrics_statsd/event_handler.ex
@@ -127,18 +127,22 @@ defmodule TelemetryMetricsStatsd.EventHandler do
 
   @spec publish_metrics(pid(), :ets.tid(), [binary()]) :: :ok
   defp publish_metrics(reporter, pool_id, packets) do
-    udp = TelemetryMetricsStatsd.get_udp(pool_id)
-
-    Enum.reduce_while(packets, :cont, fn packet, :cont ->
-      case UDP.send(udp, packet) do
-        :ok ->
-          {:cont, :cont}
-
-        {:error, reason} ->
-          TelemetryMetricsStatsd.udp_error(reporter, udp, reason)
-          {:halt, :halt}
-      end
-    end)
+    case TelemetryMetricsStatsd.get_udp(pool_id) do
+      {:ok, udp} ->
+        Enum.reduce_while(packets, :cont, fn packet, :cont ->
+          case UDP.send(udp, packet) do
+            :ok ->
+              {:cont, :cont}
+
+            {:error, reason} ->
+              TelemetryMetricsStatsd.udp_error(reporter, udp, reason)
+              {:halt, :halt}
+          end
+        end)
+
+      :error ->
+        :ok
+    end
   end
 
   @spec sample(Metrics.t()) :: Metrics.measurement() | nil
diff --git a/lib/telemetry_metrics_statsd/udp.ex b/lib/telemetry_metrics_statsd/udp.ex
index 7a68873..676dbf1 100644
--- a/lib/telemetry_metrics_statsd/udp.ex
+++ b/lib/telemetry_metrics_statsd/udp.ex
@@ -22,7 +22,8 @@ defmodule TelemetryMetricsStatsd.UDP do
   @spec open(config()) ::
           {:ok, t()} | {:error, reason :: term()}
   def open(config) do
-    opts = if config[:socket_path], do: [:local], else: []
+    opts = [active: false]
+    opts = if config[:socket_path], do: [:local | opts], else: opts
 
     case :gen_udp.open(0, opts) do
       {:ok, socket} ->


Test patch:
diff --git a/test/telemetry_metrics_statsd_test.exs b/test/telemetry_metrics_statsd_test.exs
index 5a9e74f..ee028b7 100644
--- a/test/telemetry_metrics_statsd_test.exs
+++ b/test/telemetry_metrics_statsd_test.exs
@@ -328,28 +328,39 @@ defmodule TelemetryMetricsStatsdTest do
     test "reporting a UDP error logs an error" do
       reporter = start_reporter(metrics: [])
       pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
-      udp = TelemetryMetricsStatsd.get_udp(pool_id)
+      {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
 
       assert capture_log(fn ->
                TelemetryMetricsStatsd.udp_error(reporter, udp, :closed)
                # errors.
-               eventually(fn -> TelemetryMetricsStatsd.get_udp(pool_id) != udp end)
+               eventually(fn ->
+                 {:ok, new_udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+                 new_udp != udp
+               end)
              end) =~ ~r/\[error\] Failed to publish metrics over UDP: :closed/
     end
 
     test "reporting a UDP error for the same socket multiple times generates only one log" do
       reporter = start_reporter(metrics: [])
       pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
-      udp = TelemetryMetricsStatsd.get_udp(pool_id)
+      {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
 
       assert capture_log(fn ->
                TelemetryMetricsStatsd.udp_error(reporter, udp, :closed)
-               eventually(fn -> TelemetryMetricsStatsd.get_udp(pool_id) != udp end)
+
+               eventually(fn ->
+                 {:ok, new_udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+                 new_udp != udp
+               end)
              end) =~ ~r/\[error\] Failed to publish metrics over UDP: :closed/
 
       assert capture_log(fn ->
                TelemetryMetricsStatsd.udp_error(reporter, udp, :closed)
-               eventually(fn -> TelemetryMetricsStatsd.get_udp(pool_id) != udp end)
+
+               eventually(fn ->
+                 {:ok, new_udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+                 new_udp != udp
+               end)
              end) == ""
     end
 
@@ -357,20 +368,28 @@ defmodule TelemetryMetricsStatsdTest do
     test "reporting a UDP error and fetching a socket returns a new socket" do
       reporter = start_reporter(metrics: [], pool_size: 1)
       pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
-      udp = TelemetryMetricsStatsd.get_udp(pool_id)
+      {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
 
       TelemetryMetricsStatsd.udp_error(reporter, udp, :closed)
-      assert eventually(fn -> TelemetryMetricsStatsd.get_udp(pool_id) != udp end)
+
+      assert eventually(fn ->
+               {:ok, new_udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+               new_udp != udp
+             end)
     end
 
     @tag :capture_log
     test "reporting a UDP error and opening a new socket closes the old socket" do
       reporter = start_reporter(metrics: [])
       pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
-      udp = TelemetryMetricsStatsd.get_udp(pool_id)
+      {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
 
       TelemetryMetricsStatsd.udp_error(reporter, udp, :closed)
-      eventually(fn -> TelemetryMetricsStatsd.get_udp(pool_id) != udp end)
+
+      eventually(fn ->
+        {:ok, new_udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+        new_udp != udp
+      end)
 
       assert :gen_udp.recv(udp.socket, 0) == {:error, :closed}
     end
@@ -568,7 +587,8 @@ defmodule TelemetryMetricsStatsdTest do
 
     udps =
       for _ <- 1..50, uniq: true do
-        TelemetryMetricsStatsd.get_udp(pool_id)
+        {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+        udp
       end
 
     assert length(udps) == 4
@@ -587,49 +607,70 @@ defmodule TelemetryMetricsStatsdTest do
     assert_reported(socket, "http.requests:1|c")
   end
 
-  describe "periodic hostname resolution" do
-    test "is performed when configured" do
+  describe "hostname resolution" do
+    test "is performed on start by default" do
       counter = given_counter("http.request.count")
 
       reporter =
         start_reporter(
           host: "localhost",
-          metrics: [counter],
-          name: :my_statsd,
-          host_resolution_interval: 5000
+          metrics: [counter]
         )
 
       pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
 
+      {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
+      assert udp.host == {127, 0, 0, 1}
+    end
+
+    test "is not periodically repeated by default" do
+      counter = given_counter("http.request.count")
+
+      reporter =
+        start_reporter(
+          host: "localhost",
+          metrics: [counter]
+        )
+
+      pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
       udp = TelemetryMetricsStatsd.get_udp(pool_id)
       assert udp.host == {127, 0, 0, 1}
+
+      with_mock :inet, [:passthrough, :unstick],
+        gethostbyname: fn _ -> {:ok, {:hostent, 'localhost', [], :inet, 4, [{10, 0, 0, 0}]}} end do
+        assert_raise Liveness, fn ->
+          eventually(fn ->
+            udp = TelemetryMetricsStatsd.get_udp(pool_id)
+            udp.host == {10, 0, 0, 0}
+          end)
+        end
+      end
     end
 
-    test "is periodically repeated" do
+    test "is periodically repeated if configured" do
       counter = given_counter("http.request.count")
 
       reporter =
         start_reporter(
           host: "localhost",
           metrics: [counter],
-          name: :my_statsd,
           host_resolution_interval: 100
         )
 
       pool_id = TelemetryMetricsStatsd.get_pool_id(reporter)
-      udp = TelemetryMetricsStatsd.get_udp(pool_id)
+      {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
       assert udp.host == {127, 0, 0, 1}
 
       with_mock :inet, [:passthrough, :unstick],
         gethostbyname: fn _ -> {:ok, {:hostent, 'localhost', [], :inet, 4, [{10, 0, 0, 0}]}} end do
         eventually(fn ->
-          udp = TelemetryMetricsStatsd.get_udp(pool_id)
+          {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
           assert udp.host == {10, 0, 0, 0}
         end)
       end
 
       eventually(fn ->
-        udp = TelemetryMetricsStatsd.get_udp(pool_id)
+        {:ok, udp} = TelemetryMetricsStatsd.get_udp(pool_id)
         assert udp.host == {127, 0, 0, 1}
       end)
     end