Repository: beam-telemetry/telemetry_metrics_statsd (elixir)

Issue:
Erlang Port Exhaustion
When sending metrics over UDP fails, `TelemetryMetricsStatsd` opens a fresh socket to send the metrics over. I'm assuming this is to handle some transient errors that are tied to that specific socket. However, the old socket is **not being closed** before the new one gets opened, resulting in port leakage and leading to an exhaustion of ports in the VM.

Combined with the Unix domain socket support I've implemented, and given the fact that `:gen_udp.send/4` returns an error if the Unix domain socket is not available, renders the following chart from our production systems last week, where port counts skyrocketed from 8k to 126k in a matter of minutes throughout our infrastructure (we were doing a rollout of the Datadog agent, which made the sockets unavailable for a moment):

![Screenshot_2020-07-20 Metric Explorer Datadog](https://user-images.githubusercontent.com/801356/87973218-772e1480-cac8-11ea-857e-e93098cb31ec.png)

To reproduce the issue:

1. Configure `TelemetryMetricsStatsd` with `socket_path` pointing to something that's _not_ a Unix domain socket.
2. Open `:observer`.
3. Send some metrics over.
4. Observe the port count increasing with each error logged.

The issue remains even when using regular UDP - it's just easier to reproduce with Unix domain sockets.

If I understand the socket re-opening correctly, we can close the old socket as soon as the new one is opened. I'll open a PR with a fix soon.

Pull request:
Close old socket after opening a new one
This makes sure the old, failing socket gets closed after a new socket gets opened. Fixes #39.

Patch:
diff --git a/lib/telemetry_metrics_statsd.ex b/lib/telemetry_metrics_statsd.ex
index 1854d60..00e96ea 100644
--- a/lib/telemetry_metrics_statsd.ex
+++ b/lib/telemetry_metrics_statsd.ex
@@ -441,8 +441,9 @@ defmodule TelemetryMetricsStatsd do
     Logger.error("Failed to publish metrics over UDP: #{inspect(reason)}")
 
     case UDP.open(state.udp_config) do
-      {:ok, udp} ->
-        {:noreply, %{state | udp: udp}}
+      {:ok, new_udp} ->
+        UDP.close(udp)
+        {:noreply, %{state | udp: new_udp}}
 
       {:error, reason} ->
         Logger.error("Failed to reopen UDP socket: #{inspect(reason)}")
diff --git a/lib/telemetry_metrics_statsd/udp.ex b/lib/telemetry_metrics_statsd/udp.ex
index 189f053..80f49f3 100644
--- a/lib/telemetry_metrics_statsd/udp.ex
+++ b/lib/telemetry_metrics_statsd/udp.ex
@@ -47,4 +47,9 @@ defmodule TelemetryMetricsStatsd.UDP do
       :gen_udp.send(socket, host, port, data)
     end
   end
+
+  @spec close(t()) :: :ok
+  def close(%__MODULE__{socket: socket}) do
+    :gen_udp.close(socket)
+  end
 end


Test patch:
diff --git a/test/telemetry_metrics_statsd_test.exs b/test/telemetry_metrics_statsd_test.exs
index 9c50cba..d1d2324 100644
--- a/test/telemetry_metrics_statsd_test.exs
+++ b/test/telemetry_metrics_statsd_test.exs
@@ -284,7 +284,7 @@ defmodule TelemetryMetricsStatsdTest do
   end
 
   describe "UDP error handling" do
-    test "notifying a UDP error logs an error" do
+    test "reporting a UDP error logs an error" do
       reporter = start_reporter(metrics: [])
       udp = TelemetryMetricsStatsd.get_udp(reporter)
 
@@ -296,7 +296,7 @@ defmodule TelemetryMetricsStatsdTest do
              end) =~ ~r/\[error\] Failed to publish metrics over UDP: :closed/
     end
 
-    test "notifying a UDP error for the same socket multiple times generates only one log" do
+    test "reporting a UDP error for the same socket multiple times generates only one log" do
       reporter = start_reporter(metrics: [])
       udp = TelemetryMetricsStatsd.get_udp(reporter)
 
@@ -312,7 +312,7 @@ defmodule TelemetryMetricsStatsdTest do
     end
 
     @tag :capture_log
-    test "notifying a UDP error and fetching a socket returns a new socket" do
+    test "reporting a UDP error and fetching a socket returns a new socket" do
       reporter = start_reporter(metrics: [])
       udp = TelemetryMetricsStatsd.get_udp(reporter)
 
@@ -321,6 +321,17 @@ defmodule TelemetryMetricsStatsdTest do
 
       assert new_udp != udp
     end
+
+    @tag :capture_log
+    test "reporting a UDP error and opening a new socket closes the old socket" do
+      reporter = start_reporter(metrics: [])
+      udp = TelemetryMetricsStatsd.get_udp(reporter)
+
+      TelemetryMetricsStatsd.udp_error(reporter, udp, :closed)
+      Process.sleep(100)
+
+      assert :gen_udp.recv(udp.socket, 0) == {:error, :closed}
+    end
   end
 
   describe "Unix domain socket support" do