Add socket protocol support and a telemetry gate

Pools accept a protocol option (or the protocol app env) so http
pools can run on shackle_socket, the socket NIF protocol added in
shackle 0.7.4, instead of gen_tcp; https pools always use
shackle_ssl. Setup and client specs widen to shackle:socket().
shackle_socket needs OTP 27.3+ at runtime, so the eunit coverage is
gated on OTP 27+.

Requests emit telemetry in the caller hot path on every call; the
telemetry app env (read once into a persistent_term at app start)
now disables emission entirely, mirroring shackle.

Author: lpgauth

README.md

@@ -84,6 +84,14 @@ ok
     <td>random</td>
     <td>connection selection strategy</td>
   </tr>
+  <tr>
+    <td>protocol</td>
+    <td>shackle_socket | shackle_tcp</td>
+    <td>shackle_tcp</td>
+    <td>shackle transport for http pools (shackle_socket runs on the
+    socket NIF and needs OTP 27.3+); https pools always use
+    shackle_ssl</td>
+  </tr>
   <tr>
     <td>reconnect</td>
     <td>boolean()</td>
@@ -141,6 +149,9 @@ remains observable via shackle's own telemetry — buoy's events
 surface the buoy-level routing decision without duplicating that
 work.
 
+Set the `telemetry` app env to `false` to skip event emission
+entirely (read once at application start).
+
 ## Tests
 
 ```makefile

include/buoy.hrl

@@ -47,6 +47,7 @@
 -type option()        :: {backlog_size, pos_integer()} |
                          {pool_size, pos_integer()} |
                          {pool_strategy, random | round_robin} |
+                         {protocol, shackle_socket | shackle_tcp} |
                          {reconnect, boolean()} |
                          {reconnect_time_max, pos_integer() | infinity} |
                          {reconnect_time_min, pos_integer()} |

rebar.config

@@ -1,10 +1,10 @@
 {deps, [
   {foil, "0.1.6"},
-  {shackle, "0.7.3"},
+  {shackle, "0.7.4"},
   {telemetry, "1.4.2"}
 ]}.
 
-{dialyzer, [{plt_extra_apps, [public_key]}]}.
+{dialyzer, [{plt_extra_apps, [public_key, ssl]}]}.
 
 {ex_doc, [
   {api_reference, true},

rebar.lock

@@ -2,19 +2,19 @@
 [{<<"foil">>,{pkg,<<"foil">>,<<"0.1.6">>},0},
  {<<"knot">>,{pkg,<<"knot">>,<<"0.1.2">>},1},
  {<<"metal">>,{pkg,<<"metal">>,<<"0.1.2">>},1},
- {<<"shackle">>,{pkg,<<"shackle">>,<<"0.7.3">>},0},
+ {<<"shackle">>,{pkg,<<"shackle">>,<<"0.7.4">>},0},
  {<<"telemetry">>,{pkg,<<"telemetry">>,<<"1.4.2">>},0}]}.
 [
 {pkg_hash,[
  {<<"foil">>, <<"31A9048E8EC6E8AC951ED3C4E4EB8F549465920C70F19E92CD9FC8A6867E7253">>},
  {<<"knot">>, <<"3B6C07CD24831D62752D879F031BCEDCFADB343738679EB1774A029D0A8B7D48">>},
  {<<"metal">>, <<"7501F14B075380F829498C4CE605643D6C7F0C55F60EBFCFBF694C33A558DF98">>},
- {<<"shackle">>, <<"DEDE0777BC297B212DE3DBE4377B391DA991FC011B4D5201960A5C4F11E83B2C">>},
+ {<<"shackle">>, <<"CF7AF47CCA9C881C436E2E36EE91AB43CBF5C7AAF407146E00F4C4F66B5828B5">>},
  {<<"telemetry">>, <<"A0CB522801DFFB1C49FE6E30561BADFFC7B6D0E180DB1300DF759FAA22062855">>}]},
 {pkg_hash_ext,[
  {<<"foil">>, <<"8D34B12D712FFB30AD9E192F57885ABB1A61510CD3D9F6AA44470510A50B979E">>},
  {<<"knot">>, <<"42CB91A8024A8857D70A61B33E83D31370D5EB48681B366AE1203CB7C7CCFB38">>},
  {<<"metal">>, <<"CAC84E5CF2875B56E68874F3F0B3D598AF3EC30397EC2151D5C81C408961E6E4">>},
- {<<"shackle">>, <<"056944FEEB85ED0F3E1B7505FBC7174CDE0788993A19E9ADA85137A23CA56349">>},
+ {<<"shackle">>, <<"43CCDE785082B9670BDFA40BF459D9CF602E7EB425FA31C73A66EBEE0E3CA9B5">>},
  {<<"telemetry">>, <<"928F6495066506077862C0D1646609EED891A4326BEE3126BA54B60AF61FEBB1">>}]}
 ].

src/buoy.erl

@@ -69,16 +69,10 @@ async_request(Method, #buoy_url {
             Request = {request, Method, Path, Headers, Host, Body},
             Pid = buoy_opts(pid, BuoyOpts),
             Timeout = buoy_opts(timeout, BuoyOpts),
-            telemetry:execute([buoy, request, sent],
-                              #{count => 1},
-                              #{method => Method, host => Host,
-                                async => true}),
+            buoy_telemetry:request_sent(Method, Host, true),
             shackle:cast(PoolName, Request, Pid, Timeout);
         {error, Reason} = E ->
-            telemetry:execute([buoy, request, error],
-                              #{count => 1},
-                              #{method => Method, host => Host,
-                                reason => Reason}),
+            buoy_telemetry:request_error(Method, Host, Reason),
             E
     end.
 
@@ -135,16 +129,10 @@ request(Method, #buoy_url {
             Body = buoy_opts(body, BuoyOpts),
             Request = {request, Method, Path, Headers, Host, Body},
             Timeout = buoy_opts(timeout, BuoyOpts),
-            telemetry:execute([buoy, request, sent],
-                              #{count => 1},
-                              #{method => Method, host => Host,
-                                async => false}),
+            buoy_telemetry:request_sent(Method, Host, false),
             shackle:call(PoolName, Request, Timeout);
         {error, Reason} = E ->
-            telemetry:execute([buoy, request, error],
-                              #{count => 1},
-                              #{method => Method, host => Host,
-                                reason => Reason}),
+            buoy_telemetry:request_error(Method, Host, Reason),
             E
     end.
 

src/buoy_app.erl

@@ -30,6 +30,8 @@ stop() ->
     {ok, pid()}.
 
 start(_StartType, _StartArgs) ->
+    persistent_term:put({buoy, telemetry},
+        ?GET_ENV(telemetry, true)),
     buoy_sup:start_link().
 
 -spec stop(term()) ->

src/buoy_client.erl

@@ -33,7 +33,7 @@ init(_Opts) ->
         queue = queue:new()
     }}.
 
--spec setup(inet:socket(), state()) ->
+-spec setup(shackle:socket(), state()) ->
     {ok, state()}.
 
 setup(_Socket, State) ->

src/buoy_pool.erl

@@ -100,7 +100,7 @@ client_options(Protocol, Hostname, Port, Options) ->
 
     [{ip, binary_to_list(Hostname)},
      {port, Port},
-     {protocol, shackle_protocol(Protocol)},
+     {protocol, shackle_protocol(Protocol, Options)},
      {reconnect, Reconnect},
      {reconnect_time_max, ReconnectTimeMax},
      {reconnect_time_min, ReconnectTimeMin},
@@ -120,7 +120,7 @@ pool_options(Options) ->
      {pool_size, PoolSize},
      {pool_strategy, PoolStrategy}].
 
-shackle_protocol(http) ->
-    shackle_tcp;
-shackle_protocol(https) ->
+shackle_protocol(http, Options) ->
+    ?LOOKUP(protocol, Options, ?GET_ENV(protocol, shackle_tcp));
+shackle_protocol(https, _Options) ->
     shackle_ssl.

src/buoy_telemetry.erl

@@ -0,0 +1,39 @@
+-module(buoy_telemetry).
+-include("buoy_internal.hrl").
+
+-compile(inline).
+-compile({inline_size, 512}).
+
+-export([
+    request_error/3,
+    request_sent/3
+]).
+
+%% internal
+-spec request_error(method(), host(), atom()) ->
+    ok.
+
+request_error(Method, Host, Reason) ->
+    case enabled() of
+        true ->
+            telemetry:execute([buoy, request, error], #{count => 1},
+                #{method => Method, host => Host, reason => Reason});
+        false ->
+            ok
+    end.
+
+-spec request_sent(method(), host(), boolean()) ->
+    ok.
+
+request_sent(Method, Host, Async) ->
+    case enabled() of
+        true ->
+            telemetry:execute([buoy, request, sent], #{count => 1},
+                #{method => Method, host => Host, async => Async});
+        false ->
+            ok
+    end.
+
+%% private
+enabled() ->
+    persistent_term:get({buoy, telemetry}, true).

test/buoy_tests.erl

@@ -32,9 +32,27 @@ buoy_test_() ->
         fun put_subtest/0,
         fun head_subtest/0,
         fun telemetry_sent_subtest/0,
-        fun telemetry_error_subtest/0
+        fun telemetry_error_subtest/0,
+        fun telemetry_disabled_subtest/0
     ]}.
 
+buoy_socket_test_() ->
+    case list_to_integer(erlang:system_info(otp_release)) >= 27 of
+        true ->
+            {setup,
+                fun () -> setup([{protocol, shackle_socket}]) end,
+                fun (_) -> cleanup() end,
+            [
+                fun custom_subtest/0,
+                fun get_subtest/0,
+                fun post_subtest/0,
+                fun put_subtest/0,
+                fun head_subtest/0
+            ]};
+        false ->
+            []
+    end.
+
 %% tests
 custom_subtest() ->
     {ok, ReqId} = buoy:async_custom(?VERB, ?URL(?URL_5), #{}),
@@ -118,16 +136,40 @@ telemetry_error_subtest() ->
         telemetry:detach(HandlerId)
     end.
 
+telemetry_disabled_subtest() ->
+    Self = self(),
+    HandlerId = <<"buoy-test-disabled">>,
+    ok = telemetry:attach(HandlerId, [buoy, request, sent],
+        fun (Event, Measurements, Metadata, _) ->
+            Self ! {telemetry, Event, Measurements, Metadata}
+        end, undefined),
+    persistent_term:put({buoy, telemetry}, false),
+    try
+        {ok, ?RESP_1} = buoy:get(?URL(?URL_1), #{}),
+        receive
+            {telemetry, _, _, _} ->
+                erlang:error(unexpected_telemetry_event)
+        after 200 ->
+            ok
+        end
+    after
+        persistent_term:put({buoy, telemetry}, true),
+        telemetry:detach(HandlerId)
+    end.
+
 %% utils
 cleanup() ->
     buoy_pool:stop(?URL(?URL_1)),
     buoy_app:stop(),
     buoy_http_server:stop().
 
 setup() ->
+    setup([]).
+
+setup(PoolOptions) ->
     error_logger:tty(false),
     {ok, _} = buoy_http_server:start(),
     timer:sleep(200),
     buoy_app:start(),
-    buoy_pool:start(?URL(?URL_1)),
+    buoy_pool:start(?URL(?URL_1), PoolOptions),
     timer:sleep(200).