Merge pull request #971 from appsignal/fix-hackney-pool-saturation

Fix Hackney pool saturation

.changesets/fix-check-ins-not-being-sent-after-some-time.md

@@ -0,0 +1,8 @@
+---
+bump: patch
+type: fix
+---
+
+Fix an issue where, after a certain amount of time, check-ins would no longer be sent.
+
+This issue also caused the default Hackney connection pool to be saturated, affecting other code that uses the default Hackney connection pool.

lib/appsignal.ex

@@ -48,7 +48,8 @@ defmodule Appsignal do
       {Appsignal.Tracer, []},
       {Appsignal.Monitor, []},
       {Appsignal.Probes, []},
-      {Appsignal.CheckIn.Scheduler, []}
+      {Appsignal.CheckIn.Scheduler, []},
+      :hackney_pool.child_spec(:appsignal_transmitter, [])
     ]
 
     result = Supervisor.start_link(children, strategy: :one_for_one, name: Appsignal.Supervisor)

lib/appsignal/check_in/scheduler.ex

@@ -98,11 +98,11 @@ defmodule Appsignal.CheckIn.Scheduler do
     config = Appsignal.Config.config()
     endpoint = "#{config[:logging_endpoint]}/check_ins/json"
 
-    case @transmitter.transmit(endpoint, {Enum.reverse(events), :ndjson}, config) do
-      {:ok, status_code, _, _} when status_code in 200..299 ->
+    case @transmitter.transmit_and_close(endpoint, {Enum.reverse(events), :ndjson}, config) do
+      {:ok, status_code, _} when status_code in 200..299 ->
         @integration_logger.trace("Transmitted #{description}")
 
-      {:ok, status_code, _, _} ->
+      {:ok, status_code, _} ->
         @integration_logger.error(
           "Failed to transmit #{description}: status code was #{status_code}"
         )

lib/appsignal/diagnose/report.ex

@@ -13,6 +13,7 @@ defmodule Appsignal.Diagnose.Report do
     case Transmitter.transmit(config[:diagnose_endpoint], {%{diagnose: report}, :json}, config) do
       {:ok, 200, _, reference} ->
         {:ok, body} = :hackney.body(reference)
+        :hackney.close(reference)
 
         case Jason.decode(body) do
           {:ok, response} -> {:ok, response["token"]}
@@ -21,6 +22,7 @@ defmodule Appsignal.Diagnose.Report do
 
       {:ok, status_code, _, reference} ->
         {:ok, body} = :hackney.body(reference)
+        :hackney.close(reference)
         {:error, %{status_code: status_code, body: body}}
 
       {:error, reason} ->

lib/appsignal/transmitter.ex

@@ -13,6 +13,12 @@ defmodule Appsignal.Transmitter do
   def transmit(url, payload_and_format \\ {nil, nil}, config \\ nil)
   def transmit(url, nil, config), do: transmit(url, {nil, nil}, config)
 
+  # This function calls `:hackney.request/5` -- it is the
+  # caller's responsibility to ensure that `:hackney.close/1` is
+  # called afterwards.
+  #
+  # If you're not interested in the body, only in the status code
+  # and headers, use `transmit_and_close/3` instead.
   def transmit(url, {payload, format}, config) do
     config = config || Appsignal.Config.config()
 
@@ -32,6 +38,17 @@ defmodule Appsignal.Transmitter do
     request(:post, url, headers, body)
   end
 
+  def transmit_and_close(url, payload_and_format \\ {nil, nil}, config \\ nil) do
+    case transmit(url, payload_and_format, config) do
+      {:ok, status, headers, reference} ->
+        :hackney.close(reference)
+        {:ok, status, headers}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
   defp encode_body(nil, _), do: ""
   defp encode_body(payload, :json), do: Jason.encode!(payload)
 
@@ -41,6 +58,13 @@ defmodule Appsignal.Transmitter do
   end
 
   defp options do
+    ssl_options() ++
+      [
+        pool: :appsignal_transmitter
+      ]
+  end
+
+  defp ssl_options do
     ca_file_path = Appsignal.Config.ca_file_path()
 
     options =

lib/appsignal/utils/push_api_key_validator.ex

@@ -5,10 +5,10 @@ defmodule Appsignal.Utils.PushApiKeyValidator do
   def validate(config) do
     url = "#{config[:endpoint]}/1/auth"
 
-    case Transmitter.transmit(url, nil, config) do
-      {:ok, 200, _, _} -> :ok
-      {:ok, 401, _, _} -> {:error, :invalid}
-      {:ok, status_code, _, _} -> {:error, status_code}
+    case Transmitter.transmit_and_close(url, nil, config) do
+      {:ok, 200, _} -> :ok
+      {:ok, 401, _} -> {:error, :invalid}
+      {:ok, status_code, _} -> {:error, status_code}
       {:error, reason} -> {:error, reason}
     end
   end

test/appsignal/transmitter_test.exs

@@ -80,9 +80,11 @@ defmodule Appsignal.TransmitterTest do
   end
 
   test "uses the default CA certificate" do
-    [_method, _url, _headers, _body, [ssl_options: ssl_options]] =
+    [_method, _url, _headers, _body, options] =
       Transmitter.request(:get, "https://example.com")
 
+    ssl_options = Keyword.get(options, :ssl_options)
+
     assert ssl_options[:verify] == :verify_peer
     assert ssl_options[:cacertfile] == Config.ca_file_path()
 
@@ -123,9 +125,10 @@ defmodule Appsignal.TransmitterTest do
     path = "priv/cacert.pem"
 
     with_config(%{ca_file_path: path}, fn ->
-      [_method, _url, _headers, _body, [ssl_options: ssl_options]] =
+      [_method, _url, _headers, _body, options] =
         Transmitter.request(:get, "https://example.com")
 
+      ssl_options = Keyword.get(options, :ssl_options)
       assert ssl_options[:cacertfile] == path
     end)
   end
@@ -136,8 +139,10 @@ defmodule Appsignal.TransmitterTest do
     with_config(%{ca_file_path: path}, fn ->
       log =
         capture_log(fn ->
-          assert [_method, _url, _headers, _body, []] =
-                   Transmitter.request(:get, "https://example.com")
+          [_method, _url, _headers, _body, options] =
+            Transmitter.request(:get, "https://example.com")
+
+          refute Keyword.has_key?(options, :ssl_options)
         end)
 
       # credo:disable-for-lines:2 Credo.Check.Readability.MaxLineLength

test/support/appsignal/fake_transmitter.ex

@@ -31,6 +31,16 @@ defmodule Appsignal.FakeTransmitter do
     Agent.get(__MODULE__, & &1[:response]).()
   end
 
+  def transmit_and_close(url, payload, config) do
+    case transmit(url, payload, config) do
+      {:ok, status, headers, _reference} ->
+        {:ok, status, headers}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
   def transmitted do
     Agent.get(__MODULE__, &Enum.reverse(&1[:transmitted]))
   end