feat: allow for gen_rpc TLS to be configured (#1507)

Either TCP or SSL server port can be exposed. Not both.

The gen_rpc metrics had to be fixed to account for TLS

Author: edgurgel

README.md

@@ -174,6 +174,11 @@ If you're using the default tenant, the URL is `ws://realtime-dev.localhost:4000
 | OTEL_TRACES_SAMPLER                             | string  | Default to `parentbased_always_on` . More info [here](https://opentelemetry.io/docs/languages/erlang/sampling/#environment-variables)                                                                                                                                                                                           |
 | GEN_RPC_TCP_SERVER_PORT                         | number  | Port served by `gen_rpc`. Must be secured just like the Erlang distribution port. Defaults to 5369                                                                                                                                                                                                                              |
 | GEN_RPC_TCP_CLIENT_PORT                         | number  | `gen_rpc` connects to another node using this port. Most of the time it should be the same as GEN_RPC_TCP_SERVER_PORT. Defaults to 5369                                                                                                                                                                                         |
+| GEN_RPC_SSL_SERVER_PORT                         | number  | Port served by `gen_rpc` secured with TLS. Must also define GEN_RPC_CERTFILE, GEN_RPC_KEYFILE and GEN_RPC_CACERTFILE. If this is defined then only TLS connections will be set-up.                                                                                                                                              |
+| GEN_RPC_SSL_CLIENT_PORT                         | number  | `gen_rpc` connects to another node using this port. Most of the time it should be the same as GEN_RPC_SSL_SERVER_PORT. Defaults to 6369                                                                                                                                                                                         |
+| GEN_RPC_CERTFILE                                | string  | Path to the public key in PEM format. Only needs to be provided if GEN_RPC_SSL_SERVER_PORT is defined                                                                                                                                                                                                                           |
+| GEN_RPC_KEYFILE                                 | string  | Path to the private key in PEM format. Only needs to be provided if GEN_RPC_SSL_SERVER_PORT is defined                                                                                                                                                                                                                          |
+| GEN_RPC_CACERTFILE                              | string  | Path to the certificate authority public key in PEM format. Only needs to be provided if GEN_RPC_SSL_SERVER_PORT is defined                                                                                                                                                                                                     |
 | GEN_RPC_CONNECT_TIMEOUT_IN_MS                   | number  | `gen_rpc` client connect timeout in milliseconds. Defaults to 10000.                                                                                                                                                                                                                                                            |
 | GEN_RPC_SEND_TIMEOUT_IN_MS                      | number  | `gen_rpc` client and server send timeout in milliseconds. Defaults to 10000.                                                                                                                                                                                                                                                    |
 | GEN_RPC_SOCKET_IP                               | string  | Interface which `gen_rpc` will bind to. Defaults to "0.0.0.0" (ipv4) which means that all interfaces are going to expose the `gen_rpc` port.                                                                                                                                                                                    |

config/runtime.exs

@@ -158,11 +158,41 @@ end
 if config_env() != :test do
   gen_rpc_socket_ip = System.get_env("GEN_RPC_SOCKET_IP", "0.0.0.0") |> to_charlist()
 
+  gen_rpc_ssl_server_port = System.get_env("GEN_RPC_SSL_SERVER_PORT")
+
+  gen_rpc_ssl_server_port =
+    if gen_rpc_ssl_server_port do
+      String.to_integer(gen_rpc_ssl_server_port)
+    end
+
+  gen_rpc_default_driver = if gen_rpc_ssl_server_port, do: :ssl, else: :tcp
+
+  if gen_rpc_default_driver == :ssl do
+    gen_rpc_ssl_opts = [
+      certfile: System.fetch_env!("GEN_RPC_CERTFILE"),
+      keyfile: System.fetch_env!("GEN_RPC_KEYFILE"),
+      cacertfile: System.fetch_env!("GEN_RPC_CACERTFILE")
+    ]
+
+    config :gen_rpc,
+      ssl_server_port: gen_rpc_ssl_server_port,
+      ssl_client_port: System.get_env("GEN_RPC_SSL_CLIENT_PORT", "6369") |> String.to_integer(),
+      ssl_client_options: gen_rpc_ssl_opts,
+      ssl_server_options: gen_rpc_ssl_opts,
+      tcp_server_port: false,
+      tcp_client_port: false
+  else
+    config :gen_rpc,
+      ssl_server_port: false,
+      ssl_client_port: false,
+      tcp_server_port: System.get_env("GEN_RPC_TCP_SERVER_PORT", "5369") |> String.to_integer(),
+      tcp_client_port: System.get_env("GEN_RPC_TCP_CLIENT_PORT", "5369") |> String.to_integer()
+  end
+
   case :inet.parse_address(gen_rpc_socket_ip) do
     {:ok, address} ->
       config :gen_rpc,
-        tcp_server_port: System.get_env("GEN_RPC_TCP_SERVER_PORT", "5369") |> String.to_integer(),
-        tcp_client_port: System.get_env("GEN_RPC_TCP_CLIENT_PORT", "5369") |> String.to_integer(),
+        default_client_driver: gen_rpc_default_driver,
         connect_timeout: System.get_env("GEN_RPC_CONNECT_TIMEOUT_IN_MS", "10000") |> String.to_integer(),
         send_timeout: System.get_env("GEN_RPC_SEND_TIMEOUT_IN_MS", "10000") |> String.to_integer(),
         ipv6_only: System.get_env("GEN_RPC_IPV6_ONLY", "false") == "true",

lib/realtime/monitoring/gen_rpc_metrics.ex

@@ -12,79 +12,58 @@ defmodule Realtime.GenRpcMetrics do
       {:ok, nodes_info} = :net_kernel.nodes_info()
       # Ignore "hidden" nodes (remote shell)
       nodes_info = Enum.filter(nodes_info, fn {_k, v} -> v[:type] == :normal end)
+      gen_rpc_server_port = server_port()
+      ip_address_node = ip_address_node(nodes_info)
 
-      # All TCP server sockets are managed by gen_rpc_acceptor_sup supervisor
-      # All TCP client sockets are managed by gen_rpc_client_sup supervisor
-      # For each node gen_rpc might have multiple TCP sockets
-
-      # For client processes we use the remote address (peername)
-      client_port_addresses =
-        port_addresses(:gen_rpc_client_sup)
-        |> Enum.reduce(%{}, fn {address, port}, acc ->
-          update_in(acc, [address], fn value -> [port | value || []] end)
+      {client_ports, server_ports} =
+        :erlang.ports()
+        |> Stream.filter(fn port -> :erlang.port_info(port, :name) == {:name, ~c"tcp_inet"} end)
+        |> Stream.map(&{:inet.peername(&1), :inet.sockname(&1), &1})
+        |> Stream.filter(fn
+          {{:ok, _peername}, {:ok, _sockname}, _port} -> true
+          _ -> false
+        end)
+        |> Stream.map(fn {{:ok, {peername_ipaddress, peername_port}}, {:ok, {_, server_port}}, port} ->
+          {ip_address_node[peername_ipaddress], peername_port, server_port, port}
         end)
+        |> Stream.filter(fn
+          {nil, _, _} ->
+            false
 
-      # For server processes we use the ip address without the tcp port because it's randomly assigned
-      server_port_addresses =
-        port_addresses(:gen_rpc_acceptor_sup)
-        |> Enum.reduce(%{}, fn {{ip_address, _tcp_port}, port}, acc ->
-          update_in(acc, [ip_address], fn value -> [port | value || []] end)
+          {node, peername_port, server_port, _port} ->
+            {_, client_tcp_or_ssl_port} = :gen_rpc_helper.get_client_config_per_node(node)
+            # Only keep Erlang ports that are either serving on the gen_rpc server tcp/ssl port or
+            # connecting to other nodes using the expected client tcp/ssl port for that node
+            peername_port == client_tcp_or_ssl_port or server_port == gen_rpc_server_port
+        end)
+        |> Enum.reduce({%{}, %{}}, fn {node, _peername_port, server_port, port}, {clients, servers} ->
+          if server_port == gen_rpc_server_port do
+            # This Erlang port is serving gen_rpc
+            {clients, update_in(servers, [node], fn value -> [port | value || []] end)}
+          else
+            # This Erlang port is requesting gen_rpc
+            {update_in(clients, [node], fn value -> [port | value || []] end), servers}
+          end
         end)
 
-      Map.new(nodes_info, &info(&1, client_port_addresses, server_port_addresses))
+      Map.new(nodes_info, &info(&1, client_ports, server_ports))
     else
       %{}
     end
   end
 
-  defp port_addresses(supervisor) do
-    Supervisor.which_children(supervisor)
-    |> Stream.flat_map(fn {_, pid, _, _} ->
-      # We then grab the only linked port if available
-      case Process.info(pid, :links) do
-        {:links, links} ->
-          links
-          |> Enum.filter(&is_port/1)
-          |> hd()
-          |> List.wrap()
-
-        _ ->
-          []
-      end
-    end)
-    |> Stream.map(&{:inet.peername(&1), &1})
-    |> Stream.filter(fn
-      {{:ok, _sockname}, _port} -> true
-      _ -> false
-    end)
-    |> Stream.map(fn {{:ok, address}, port} -> {address, port} end)
-  end
-
-  defp info({node, info}, client_port_addresses, server_port_addresses) do
-    case info[:address] do
-      net_address(address: address) when address != :undefined ->
-        {node, info(node, client_port_addresses, server_port_addresses, address)}
-
-      _ ->
-        {node, %{}}
-    end
-  end
-
-  defp info(node, client_port_addresses, server_port_addresses, {ip_address, _}) do
-    {:tcp, client_tcp_port} = :gen_rpc_helper.get_client_config_per_node(node)
-
-    gen_rpc_client_ports = Map.get(client_port_addresses, {ip_address, client_tcp_port}, [])
-    gen_rpc_server_ports = Map.get(server_port_addresses, ip_address, [])
-    gen_rpc_ports = gen_rpc_client_ports ++ gen_rpc_server_ports
+  defp info({node, _}, client_ports, server_ports) do
+    gen_rpc_ports = Map.get(client_ports, node, []) ++ Map.get(server_ports, node, [])
 
     if gen_rpc_ports != [] do
-      %{
-        inet_stats: inet_stats(gen_rpc_ports),
-        queue_size: queue_size(gen_rpc_ports),
-        connections: length(gen_rpc_ports)
-      }
+      {node,
+       %{
+         inet_stats: inet_stats(gen_rpc_ports),
+         queue_size: queue_size(gen_rpc_ports),
+         connections: length(gen_rpc_ports)
+       }}
     else
-      %{}
+      {node, %{}}
     end
   end
 
@@ -103,4 +82,27 @@ defmodule Realtime.GenRpcMetrics do
       acc + queue_size
     end)
   end
+
+  defp server_port() do
+    if Application.fetch_env!(:gen_rpc, :default_client_driver) == :tcp do
+      Application.fetch_env!(:gen_rpc, :tcp_server_port)
+    else
+      Application.fetch_env!(:gen_rpc, :ssl_server_port)
+    end
+  end
+
+  defp ip_address_node(nodes_info) do
+    nodes_info
+    |> Stream.map(fn {node, info} ->
+      case info[:address] do
+        net_address(address: {ip_address, _}) ->
+          {ip_address, node}
+
+        _ ->
+          {nil, node}
+      end
+    end)
+    |> Stream.filter(fn {ip_address, _node} -> ip_address != nil end)
+    |> Map.new()
+  end
 end

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.43.2",
+      version: "2.44.0",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/realtime/monitoring/distributed_metrics_test.exs

@@ -32,5 +32,28 @@ defmodule Realtime.DistributedMetricsTest do
                }
              } = DistributedMetrics.info()
     end
+
+    test "metric matches on both sides", %{node: node} do
+      # We need to generate some data first
+      Realtime.Rpc.call(node, String, :to_integer, ["25"], key: 1)
+      Realtime.Rpc.call(node, String, :to_integer, ["25"], key: 2)
+
+      local_metrics = DistributedMetrics.info()[node][:inet_stats]
+      # Use gen_rpc to not use erl dist and change the result
+      remote_metrics = :gen_rpc.call(node, DistributedMetrics, :info, [])[node()][:inet_stats]
+
+      # It's not going to 100% the same because erl dist sends pings and other things out of our control
+
+      assert local_metrics[:connections] == remote_metrics[:connections]
+
+      assert_in_delta(local_metrics[:send_avg], remote_metrics[:recv_avg], 5)
+      assert_in_delta(local_metrics[:recv_avg], remote_metrics[:send_avg], 5)
+
+      assert_in_delta(local_metrics[:send_oct], remote_metrics[:recv_oct], 5)
+      assert_in_delta(local_metrics[:recv_oct], remote_metrics[:send_oct], 5)
+
+      assert_in_delta(local_metrics[:send_max], remote_metrics[:recv_max], 5)
+      assert_in_delta(local_metrics[:recv_max], remote_metrics[:send_max], 5)
+    end
   end
 end

test/realtime/monitoring/gen_rpc_metrics_test.exs

@@ -40,8 +40,23 @@ defmodule Realtime.GenRpcMetricsTest do
       Realtime.GenRpc.call(node, String, :to_integer, ["25"], key: 1)
       Realtime.GenRpc.call(node, String, :to_integer, ["25"], key: 2)
 
-      local_metrics = GenRpcMetrics.info()[node]
-      remote_metrics = :erpc.call(node, GenRpcMetrics, :info, [])[node()]
+      :erpc.call(node, Realtime.GenRpc, :call, [node(), String, :to_integer, ["25"], [key: 1]])
+
+      local_metrics = GenRpcMetrics.info()[node][:inet_stats]
+      remote_metrics = :erpc.call(node, GenRpcMetrics, :info, [])[node()][:inet_stats]
+
+      assert Map.keys(local_metrics) == [
+               :send_pend,
+               :recv_avg,
+               :recv_cnt,
+               :recv_dvi,
+               :recv_max,
+               :recv_oct,
+               :send_avg,
+               :send_cnt,
+               :send_max,
+               :send_oct
+             ]
 
       assert local_metrics[:connections] == remote_metrics[:connections]