ha-handler/lib/ha_handler/ha_proxy.ex

defmodule HAHandler.HAProxy do
  @moduledoc """
  Interface to HAProxy's runtime API:
    https://www.haproxy.com/documentation/hapee/latest/api/runtime-api/
  """

  # TODO: can we keep the socket open by keeping the port state in a GenServer
  # and abusing the prompt mode of HAProxy's socket somehow? The later is
  # somewhat clunky, but I'm not sure if there is a way around it since HAProxy
  # seems to return EOF once a command is executed (unless prompt mode is
  # enabled).

  import HAHandler, only: [haproxy_socket: 0]

  alias :procket, as: Socket

  # How long do we wait for an answer - in milliseconds.
  @socket_read_timeout 5_000

  @doc """
  Sends a command to HAProxy's socket - please refer to HAProxy's runtime API
  documentation for available commands (see moduledoc).

  This method will either return an answer with `{:ok, answer}` or error with
  `{:error, err}`.
  """
  def execute(command) when is_binary(command) do
    case open_socket(haproxy_socket()) do
      {:ok, socket} ->
        send(socket, {self(), {:command, command <> "\n"}})
        read_from_socket(socket)
      {:error, err} ->
        {:error, err}
    end
  end

  defp extract_stats(data) do
    extracted = for entry <- data do
      for mapping <- entry do
        case mapping do
          %{
            "id" => id,
            "objType" => type,
            "field" => %{"name" => name},
            "value" => %{"value" => value},
          } ->
            %{:id => id, :type => type, :field => name, :value => value}
          _ ->
            nil
        end
      end
    end

    extracted
    |> List.flatten()
    |> Enum.group_by(
      fn mapping -> {mapping.type, mapping.id} end,
      fn mapping -> %{mapping.field => mapping.value} end
    )
    |> Enum.map(
      fn {{type, id}, grouped_mappings} ->
        grouped_mappings
        |> Enum.reduce(fn l, r -> Map.merge(l,r) end)
        |> Map.put("type", type)
        |> Map.put("id", id)
      end
    )
    |> Enum.group_by(
      fn entry -> Map.get(entry, "type") end
    )
  end

  @doc """
  Executes and parse the output of the `show stats` HAProxy command, returning
  a list of Maps.
  """
  def get_stats() do
    case execute("show stat json") do
      {:ok, raw} ->
        case Poison.decode(raw) do
          {:ok, data} ->
            extract_stats(data)

          {:error, err} ->
            {:error, err}
        end

      {:error, err} ->
        {:error, err}
    end
  end

  # Opens an UNIX socket - there is no built-in support in Elixir/Erlang so we
  # use low-level C bindings provided by :procket
  # (https://github.com/msantos/procket).
  #
  # Heavily inspired by the `NVim.Link` module:
  # https://github.com/kbrw/neovim-elixir/blob/master/lib/link.ex
  defp open_socket(path) do
    family = Socket.family(:unix)
    type = :stream
    # 0 means use the default protocol in the family.
    protocol = 0

    pad = 8 * (Socket.unix_path_max() - byte_size(path))
    sockaddr = Socket.sockaddr_common(family, byte_size(path)) <> path <> <<0::size(pad)>>

    {:ok, socket} = Socket.socket(family, type, protocol)

    case Socket.connect(socket, sockaddr) do
      :ok ->
        # We connect our UNIX socket to an erlang Port:
        #
        # > Ports provide a mechanism to start operating system processes
        # > external to the Erlang VM and communicate with them via message
        # > passing.
        # See https://hexdocs.pm/elixir/Port.html for details.
        #
        # See https://www.erlang.org/doc/man/erlang.html#open_port-2 for more
        # port options.
        #
        # * :binary: all I/O from the port is binary data objects as opposed to
        #   lists of bytes.
        # * Messages are delivered on a per line basis. Each line (delimited by
        #   the OS-dependent newline sequence) is delivered in a single message.
        #   The message data format is {Flag, Line}, where Flag is eol or noeol,
        #   and Line is the data delivered (without the newline sequence).
        {fdin, fdout} = {socket, socket}
        port = Port.open({:fd, fdin, fdout}, [{:line, 10_000}, :binary])

        {:ok, port}

      {:error, err} ->
        {:error, err}
    end
  end

  # Messages may be split due to the `{:line, L}` option specific in
  # `open_socket/1`.
  defp read_from_socket(socket, acc \\ "") do
    receive do
      {_port, {:data, {:noeol, data}}} ->
        read_from_socket(socket, acc <> data)

      {_port, {:data, {:eol, data}}} ->
        {:ok, acc <> data}

      msg ->
        {:error, {:unexpected_message, msg}}
    after
      @socket_read_timeout ->
        {:error, :timeout}
    end
  end
end