From eff344234138b399703197022e922d5cfbe89430 Mon Sep 17 00:00:00 2001 From: John Hossler Date: Wed, 17 Apr 2024 13:58:21 -0700 Subject: [PATCH 1/4] chore: emit telemetry events for streams, connections, and sockets so we can introspect connection details from our server --- lib/connection.ex | 19 +++++++++++++++---- lib/socket.ex | 10 +++++++--- lib/stream.ex | 24 ++++++++++++++++++++---- mix.exs | 1 + mix.lock | 1 + 5 files changed, 44 insertions(+), 11 deletions(-) diff --git a/lib/connection.ex b/lib/connection.ex index 31c896c..ea29519 100644 --- a/lib/connection.ex +++ b/lib/connection.ex @@ -8,7 +8,8 @@ defmodule Kadabra.Connection do remote_settings: nil, requested_streams: 0, local_settings: nil, - queue: nil + queue: nil, + start_time: nil use GenServer require Logger @@ -29,7 +30,8 @@ defmodule Kadabra.Connection do config: term, flow_control: term, local_settings: Connection.Settings.t(), - queue: pid + queue: pid, + start_time: integer() } @type sock :: {:sslsocket, any, pid | {any, any}} @@ -64,7 +66,8 @@ defmodule Kadabra.Connection do config: config, queue: queue, local_settings: settings, - flow_control: %FlowControl{} + flow_control: %FlowControl{}, + start_time: System.monotonic_time() } end @@ -169,7 +172,15 @@ defmodule Kadabra.Connection do end end - def terminate(_reason, %{config: config}) do + def terminate(reason, %{config: config, start_time: start_time}) do + duration = System.monotonic_time() - start_time + :telemetry.execute([:kadabra, :connection, :stop], + %{duration: duration}, + %{ + uri: config.uri, + reason: reason, + connection: self() + }) Kernel.send(config.client, {:closed, config.queue}) :ok end diff --git a/lib/socket.ex b/lib/socket.ex index b1bb956..19cc9ab 100644 --- a/lib/socket.ex +++ b/lib/socket.ex @@ -120,6 +120,7 @@ defmodule Kadabra.Socket do def parse_bin(socket, bin, state) do case FrameParser.parse(bin) do {:ok, frame, rest} -> + :telemetry.execute([:kadabra, :socket, :recv_frame], %{}, %{frame: frame, socket: socket}) Kernel.send(state.active_user, {:recv, frame}) parse_bin(socket, rest, state) @@ -131,11 +132,12 @@ defmodule Kadabra.Socket do # Internal socket helpers defp socket_send({:sslsocket, _, _} = pid, bin) do - # IO.puts("Sending #{byte_size(bin)} bytes") + :telemetry.execute([:kadabra, :socket, :send], %{}, %{type: :ssl, bin: bin, socket: pid}) :ssl.send(pid, bin) end defp socket_send(pid, bin) do + :telemetry.execute([:kadabra, :socket, :send], %{}, %{type: :tcp, bin: bin, socket: pid}) :gen_tcp.send(pid, bin) end @@ -179,7 +181,8 @@ defmodule Kadabra.Socket do do_recv_bin(bin, state) end - def handle_info({:tcp_closed, _socket}, state) do + def handle_info({:tcp_closed, socket}, state) do + :telemetry.execute([:kadabra, :socket, :closed], %{}, %{type: :tcp, socket: socket}) Kernel.send(state.active_user, {:closed, self()}) {:noreply, %{state | socket: nil}} end @@ -188,7 +191,8 @@ defmodule Kadabra.Socket do do_recv_bin(bin, state) end - def handle_info({:ssl_closed, _socket}, state) do + def handle_info({:ssl_closed, socket}, state) do + :telemetry.execute([:kadabra, :socket, :closed], %{}, %{type: :ssl, socket: socket}) Kernel.send(state.active_user, {:closed, self()}) {:noreply, %{state | socket: nil}} end diff --git a/lib/stream.ex b/lib/stream.ex index bc5cb92..3c7b820 100644 --- a/lib/stream.ex +++ b/lib/stream.ex @@ -11,7 +11,8 @@ defmodule Kadabra.Stream do flow: nil, uri: nil, headers: [], - on_response: nil + on_response: nil, + start_time: nil require Logger @@ -35,7 +36,8 @@ defmodule Kadabra.Stream do uri: URI.t(), flow: Kadabra.Stream.FlowControl.t(), headers: [...], - body: binary + body: binary, + start_time: integer() } @closed :closed @@ -60,11 +62,13 @@ defmodule Kadabra.Stream do encoder: config.encoder, decoder: config.decoder, connection: self(), - flow: Stream.FlowControl.new(flow_opts) + flow: Stream.FlowControl.new(flow_opts), + start_time: System.monotonic_time() } end def start_link(%Stream{} = stream) do + :telemetry.execute([:kadabra, :stream, :start], %{}, %{stream_id: stream.id, uri: stream.uri, client: stream.client, connection: stream.connection}) :gen_statem.start_link(__MODULE__, stream, []) end @@ -309,7 +313,19 @@ defmodule Kadabra.Stream do def callback_mode, do: [:handle_event_function, :state_enter] - def terminate(_reason, _state, _stream), do: :void + def terminate(reason, _state, stream) do + duration = System.monotonic_time() - stream.start_time + :telemetry.execute([:kadabra, :stream, :stop], + %{duration: duration}, + %{ + stream_id: stream.id, + uri: stream.uri, + client: stream.client, + connection: stream.connection, + reason: reason + }) + :void + end def code_change(_vsn, state, data, _extra), do: {:ok, state, data} end diff --git a/mix.exs b/mix.exs index dbf7cb8..5640154 100644 --- a/mix.exs +++ b/mix.exs @@ -45,6 +45,7 @@ defmodule Kadabra.Mixfile do defp deps do [ {:certifi, "~> 2.5"}, + {:telemetry, "~> 1.2"}, {:credo, "~> 1.0", only: [:dev, :test], runtime: false}, {:dialyxir, "~> 1.0", only: [:dev], runtime: false}, {:ex_doc, ">= 0.0.0", only: :dev, runtime: false}, diff --git a/mix.lock b/mix.lock index 6cc4e03..382d8a5 100644 --- a/mix.lock +++ b/mix.lock @@ -30,5 +30,6 @@ "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm", "fec8660eb7733ee4117b85f55799fd3833eb769a6df71ccf8903e8dc5447cfce"}, "scribe": {:hex, :scribe, "0.4.1", "4bf5395fb882995f705172b817b9a4ccae6ac83f9918fdedb6a2b9dafb57ec5f", [:mix], [{:pane, "~> 0.1", [hex: :pane, repo: "hexpm", optional: false]}], "hexpm"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.6", "cf344f5692c82d2cd7554f5ec8fd961548d4fd09e7d22f5b62482e5aeaebd4b0", [:make, :mix, :rebar3], [], "hexpm", "bdb0d2471f453c88ff3908e7686f86f9be327d065cc1ec16fa4540197ea04680"}, + "telemetry": {:hex, :telemetry, "1.2.1", "68fdfe8d8f05a8428483a97d7aab2f268aaff24b49e0f599faa091f1d4e7f61c", [:rebar3], [], "hexpm", "dad9ce9d8effc621708f99eac538ef1cbe05d6a874dd741de2e689c47feafed5"}, "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"}, } From b0aecb7f09d6829834af0a4d94f64b2e1e5b7034 Mon Sep 17 00:00:00 2001 From: John Hossler Date: Thu, 18 Apr 2024 13:38:00 -0700 Subject: [PATCH 2/4] feat: add ability with to specify ping payload to support RTT calculations --- lib/connection.ex | 12 +++++++++++- lib/connection/egress.ex | 5 +++++ lib/connection/processor.ex | 4 ++-- lib/connection_pool.ex | 6 ++++++ lib/frame/ping.ex | 9 +++++++++ lib/kadabra.ex | 5 +++++ 6 files changed, 38 insertions(+), 3 deletions(-) diff --git a/lib/connection.ex b/lib/connection.ex index ea29519..e0689d2 100644 --- a/lib/connection.ex +++ b/lib/connection.ex @@ -79,6 +79,11 @@ defmodule Kadabra.Connection do GenServer.cast(pid, {:send, :ping}) end + def ping(pid, data) do + GenServer.cast(pid, {:send, {:ping, data}}) + end + + # handle_cast def handle_cast({:send, type}, state) do @@ -109,12 +114,17 @@ defmodule Kadabra.Connection do # sendf - @spec sendf(:goaway | :ping, t) :: {:noreply, t} + @spec sendf(:goaway | :ping | {:ping, <<_::64>>}, t) :: {:noreply, t} def sendf(:ping, %Connection{config: config} = state) do Egress.send_ping(config.socket) {:noreply, state} end + def sendf({:ping, data}, %Connection{config: config} = state) do + Egress.send_ping(config.socket, data) + {:noreply, state} + end + def sendf(_else, state) do {:noreply, state} end diff --git a/lib/connection/egress.ex b/lib/connection/egress.ex index 770188e..6765c10 100644 --- a/lib/connection/egress.ex +++ b/lib/connection/egress.ex @@ -30,6 +30,11 @@ defmodule Kadabra.Connection.Egress do Socket.send(socket, bin) end + def send_ping(socket, data) do + bin = Ping.new(data) |> Encodable.to_bin() + Socket.send(socket, bin) + end + def send_local_settings(socket, settings) do bin = %Frame.Settings{settings: settings} diff --git a/lib/connection/processor.ex b/lib/connection/processor.ex index 64c88ca..ac535bb 100644 --- a/lib/connection/processor.ex +++ b/lib/connection/processor.ex @@ -178,8 +178,8 @@ defmodule Kadabra.Connection.Processor do {:ok, state} end - def process(%Ping{ack: true}, %{config: config} = state) do - Kernel.send(config.client, {:pong, self()}) + def process(%Ping{ack: true, data: data}, %{config: config} = state) do + Kernel.send(config.client, {:pong, self(), data}) {:ok, state} end diff --git a/lib/connection_pool.ex b/lib/connection_pool.ex index 8f5c00b..b089d5f 100644 --- a/lib/connection_pool.ex +++ b/lib/connection_pool.ex @@ -46,6 +46,7 @@ defmodule Kadabra.ConnectionPool do end def ping(pid), do: GenServer.call(pid, :ping) + def ping(pid, data), do: GenServer.call(pid, {:ping, data}) def close(pid), do: GenServer.call(pid, :close) @@ -79,6 +80,11 @@ defmodule Kadabra.ConnectionPool do {:reply, :ok, state} end + def handle_call({:ping, data}, _from, state) do + Connection.ping(state.connection, data) + {:reply, :ok, state} + end + def handle_call({:request, requests}, from, state) do GenServer.reply(from, :ok) diff --git a/lib/frame/ping.ex b/lib/frame/ping.ex index 95f29f2..5a1dea7 100644 --- a/lib/frame/ping.ex +++ b/lib/frame/ping.ex @@ -31,6 +31,15 @@ defmodule Kadabra.Frame.Ping do } end + @spec new(<<_::64>>) :: t + def new(<>) do + %__MODULE__{ + ack: false, + data: <>, + stream_id: 0 + } + end + @doc ~S""" Initializes a new `Frame.Ping` given a `Frame`. diff --git a/lib/kadabra.ex b/lib/kadabra.ex index 10b4558..5dfd950 100644 --- a/lib/kadabra.ex +++ b/lib/kadabra.ex @@ -156,6 +156,11 @@ defmodule Kadabra do Kadabra.ConnectionPool.ping(pid) end + @spec ping(pid, <<_::64>>) :: no_return + def ping(pid, data) when byte_size(data) == 8 do + Kadabra.ConnectionPool.ping(pid, data) + end + @doc ~S""" Makes a request with given headers and optional body. From 7130d1434b981067bf26150584421c1ebeb5739d Mon Sep 17 00:00:00 2001 From: John Hossler Date: Mon, 22 Apr 2024 17:55:07 -0700 Subject: [PATCH 3/4] chore: dry up changes to added ping payload option --- lib/connection.ex | 11 +---------- lib/connection/egress.ex | 5 ----- lib/connection_pool.ex | 6 ------ lib/frame/ping.ex | 19 +++++++++---------- lib/kadabra.ex | 16 ++++++++-------- 5 files changed, 18 insertions(+), 39 deletions(-) diff --git a/lib/connection.ex b/lib/connection.ex index e0689d2..c10b7b9 100644 --- a/lib/connection.ex +++ b/lib/connection.ex @@ -75,10 +75,6 @@ defmodule Kadabra.Connection do GenServer.call(pid, :close) end - def ping(pid) do - GenServer.cast(pid, {:send, :ping}) - end - def ping(pid, data) do GenServer.cast(pid, {:send, {:ping, data}}) end @@ -114,12 +110,7 @@ defmodule Kadabra.Connection do # sendf - @spec sendf(:goaway | :ping | {:ping, <<_::64>>}, t) :: {:noreply, t} - def sendf(:ping, %Connection{config: config} = state) do - Egress.send_ping(config.socket) - {:noreply, state} - end - + @spec sendf(:goaway | {:ping, <<_::64>> | none}, t) :: {:noreply, t} def sendf({:ping, data}, %Connection{config: config} = state) do Egress.send_ping(config.socket, data) {:noreply, state} diff --git a/lib/connection/egress.ex b/lib/connection/egress.ex index 6765c10..08ff3f5 100644 --- a/lib/connection/egress.ex +++ b/lib/connection/egress.ex @@ -25,11 +25,6 @@ defmodule Kadabra.Connection.Egress do Socket.send(socket, bin) end - def send_ping(socket) do - bin = Ping.new() |> Encodable.to_bin() - Socket.send(socket, bin) - end - def send_ping(socket, data) do bin = Ping.new(data) |> Encodable.to_bin() Socket.send(socket, bin) diff --git a/lib/connection_pool.ex b/lib/connection_pool.ex index b089d5f..41bff80 100644 --- a/lib/connection_pool.ex +++ b/lib/connection_pool.ex @@ -45,7 +45,6 @@ defmodule Kadabra.ConnectionPool do GenServer.call(pid, {:request, [request]}) end - def ping(pid), do: GenServer.call(pid, :ping) def ping(pid, data), do: GenServer.call(pid, {:ping, data}) def close(pid), do: GenServer.call(pid, :close) @@ -75,11 +74,6 @@ defmodule Kadabra.ConnectionPool do {:stop, :shutdown, :ok, state} end - def handle_call(:ping, _from, state) do - Connection.ping(state.connection) - {:reply, :ok, state} - end - def handle_call({:ping, data}, _from, state) do Connection.ping(state.connection, data) {:reply, :ok, state} diff --git a/lib/frame/ping.ex b/lib/frame/ping.ex index 5a1dea7..1f78dc2 100644 --- a/lib/frame/ping.ex +++ b/lib/frame/ping.ex @@ -13,25 +13,24 @@ defmodule Kadabra.Frame.Ping do stream_id: integer } + @empty_payload <<0, 0, 0, 0, 0, 0, 0, 0>> + @doc ~S""" - Returns new unacked ping frame. + Returns new unacked ping frame. Optionally takes a payload of 8 bytes, which + can be used to help calculate RTT times of pings that your application sends. ## Examples iex> Kadabra.Frame.Ping.new %Kadabra.Frame.Ping{data: <<0, 0, 0, 0, 0, 0, 0, 0>>, ack: false, stream_id: 0} + iex> Kadabra.Frame.Ping.new(payload: <<1, 2, 3, 4, 5, 6, 7, 8>>) + %Kadabra.Frame.Ping{data: <<1, 2, 3, 4, 5, 6, 7, 8>>, + ack: false, stream_id: 0} """ - @spec new() :: t - def new do - %__MODULE__{ - ack: false, - data: <<0, 0, 0, 0, 0, 0, 0, 0>>, - stream_id: 0 - } - end + @spec new(<<_::64>> | none()) :: t + def new(nil), do: new(@empty_payload) - @spec new(<<_::64>>) :: t def new(<>) do %__MODULE__{ ack: false, diff --git a/lib/kadabra.ex b/lib/kadabra.ex index 5dfd950..fd1c0fa 100644 --- a/lib/kadabra.ex +++ b/lib/kadabra.ex @@ -147,17 +147,17 @@ defmodule Kadabra do iex> {:ok, pid} = Kadabra.open('https://http2.golang.org') iex> Kadabra.ping(pid) iex> receive do - ...> {:pong, _pid} -> "got pong!" + ...> {:pong, _pid, _resp} -> "got pong!" ...> end "got pong!" + iex> Kadabra.ping(pid, <<1::64>>) # Send 8-byte data + iex> receive do + ...> {:pong, _pid, <<1::64>>} -> "got our data back!" + ...> end + "got our data back!" """ - @spec ping(pid) :: no_return - def ping(pid) do - Kadabra.ConnectionPool.ping(pid) - end - - @spec ping(pid, <<_::64>>) :: no_return - def ping(pid, data) when byte_size(data) == 8 do + @spec ping(pid, <<_::64>> | none) :: no_return + def ping(pid, data \\ nil) when is_nil(data) or byte_size(data) == 8 do Kadabra.ConnectionPool.ping(pid, data) end From d6b2815c8c9747e0aa5a9a9affa4a0cce9a2145a Mon Sep 17 00:00:00 2001 From: John Hossler Date: Wed, 24 Apr 2024 22:16:22 -0700 Subject: [PATCH 4/4] fix: combine docs for ping constructor to fix build issue around redefined doc attributes --- lib/frame/ping.ex | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/lib/frame/ping.ex b/lib/frame/ping.ex index 1f78dc2..8ba6f37 100644 --- a/lib/frame/ping.ex +++ b/lib/frame/ping.ex @@ -19,6 +19,8 @@ defmodule Kadabra.Frame.Ping do Returns new unacked ping frame. Optionally takes a payload of 8 bytes, which can be used to help calculate RTT times of pings that your application sends. + Can also be used to initialize a new `Frame.Ping` given a `Frame`. + ## Examples iex> Kadabra.Frame.Ping.new @@ -27,8 +29,13 @@ defmodule Kadabra.Frame.Ping do iex> Kadabra.Frame.Ping.new(payload: <<1, 2, 3, 4, 5, 6, 7, 8>>) %Kadabra.Frame.Ping{data: <<1, 2, 3, 4, 5, 6, 7, 8>>, ack: false, stream_id: 0} + iex> frame = %Kadabra.Frame{payload: <<0, 0, 0, 0, 0, 0, 0, 0>>, + ...> flags: 0x1, type: 0x6, stream_id: 0} + iex> Kadabra.Frame.Ping.new(frame) + %Kadabra.Frame.Ping{data: <<0, 0, 0, 0, 0, 0, 0, 0>>, ack: true, + stream_id: 0} """ - @spec new(<<_::64>> | none()) :: t + @spec new(<<_::64>> | none() | Frame.t()) :: t def new(nil), do: new(@empty_payload) def new(<>) do @@ -39,18 +46,6 @@ defmodule Kadabra.Frame.Ping do } end - @doc ~S""" - Initializes a new `Frame.Ping` given a `Frame`. - - ## Examples - - iex> frame = %Kadabra.Frame{payload: <<0, 0, 0, 0, 0, 0, 0, 0>>, - ...> flags: 0x1, type: 0x6, stream_id: 0} - iex> Kadabra.Frame.Ping.new(frame) - %Kadabra.Frame.Ping{data: <<0, 0, 0, 0, 0, 0, 0, 0>>, ack: true, - stream_id: 0} - """ - @spec new(Frame.t()) :: t def new(%Frame{type: 0x6, payload: <>, flags: flags, stream_id: sid}) do %__MODULE__{ ack: ack?(flags),