Allow tcp or udp sockets

Author: KronicDeth

lib/logger_logstash_backend.ex

@@ -14,33 +14,69 @@
 # limitations under the License.
 ################################################################################
 defmodule LoggerLogstashBackend do
+  alias LoggerLogstashBackend.Socket
+
   use GenEvent
   use Timex
 
+  # Struct
+
+  defstruct ~w(
+               host
+               level
+               metadata
+               name
+               port
+               protocol
+               socket
+               type
+             )a
+
+  # Functions
+
+  ## GenEvent callbacks
+
   def init({__MODULE__, name}) do
-    {:ok, configure(name, [])}
+    # trap exits, so that cleanup can occur in terminate/2
+    Process.flag(:trap_exit, true)
+
+    configure(name, [])
   end
 
   def handle_call({:configure, opts}, %{name: name}) do
-    {:ok, :ok, configure(name, opts)}
+    case configure(name, opts) do
+      {:ok, state} ->
+        {:ok, :ok, state}
+      reply = {:error, _reason} ->
+        {:remove_handler, reply}
+    end
   end
 
   def handle_event(
     {level, _gl, {Logger, msg, ts, md}}, %{level: min_level} = state
   ) do
     if is_nil(min_level) or Logger.compare_levels(level, min_level) != :lt do
       log_event level, msg, ts, md, state
+    else
+      {:ok, state}
     end
-    {:ok, state}
   end
 
+  @doc """
+  Closes socket when the backend is removed
+  """
+  def terminate(_, state)  do
+    :ok = Socket.close(state)
+
+    state
+  end
+
+  ## Private Functions
+
   defp log_event(
-    level, msg, ts, md, %{
-      host: host,
-      port: port,
+    level, msg, ts, md, state = %{
       type: type,
-      metadata: metadata,
-      socket: socket
+      metadata: metadata
     }
   ) do
     fields = md
@@ -56,10 +92,19 @@ defmodule LoggerLogstashBackend do
       message: to_string(msg),
       fields: fields
     }
-    :gen_udp.send socket, host, port, to_char_list(json)
+
+    with {:error, reason} <- Socket.send(state, [json, "\n"]) do
+      # fallback in case TCP configuration is bad
+      IO.puts :stderr,
+              "Could not log message (#{json}) to socket (#{url(state)}) due to #{inspect reason}.  " <>
+              "Check that state (#{inspect state}) is correct."
+    end
+
+
+    {:ok, state}
   end
 
-  defp configure(name, opts) do
+  defp configure(name, opts) when is_atom(name) and is_list(opts) do
     env = Application.get_env :logger, name, []
     opts = Keyword.merge env, opts
     Application.put_env :logger, name, opts
@@ -69,18 +114,23 @@ defmodule LoggerLogstashBackend do
     type = Keyword.get opts, :type, "elixir"
     host = Keyword.get opts, :host
     port = Keyword.get opts, :port
-    {:ok, socket} = :gen_udp.open 0
-    %{
-      name: name,
-      host: to_char_list(host),
-      port: port,
-      level: level,
-      socket: socket,
-      type: type,
-      metadata: metadata
-    }
+    protocol = Keyword.get opts, :protocol, :udp
+
+    state = %__MODULE__{level: level, metadata: metadata, name: name, protocol: protocol, type: type}
+
+    configure_socket(state, host, port)
   end
 
+  defp configure_socket(state, host, port) when is_nil(host) or is_nil(port), do: {:ok, state}
+  defp configure_socket(state, host, port) do
+    with reply = {:error, reason} <- Socket.open %{state | host: to_char_list(host), port: port} do
+      IO.puts :stderr, "Could not open socket (#{url(state)}) due to reason #{inspect reason}"
+
+      reply
+    end
+  end
+
+
   # inspects the argument only if it is a pid
   defp inspect_pid(pid) when is_pid(pid), do: inspect(pid)
   defp inspect_pid(other), do: other
@@ -91,4 +141,8 @@ defmodule LoggerLogstashBackend do
       {key, inspect_pid(value)}
     end
   end
+
+  defp url(%__MODULE__{host: host, port: port, protocol: protocol}) do
+    "#{protocol}://#{host}:#{port}"
+  end
 end

lib/logger_logstash_backend/socket.ex

@@ -0,0 +1,50 @@
+defmodule LoggerLogstashBackend.Socket do
+  @moduledoc """
+  Hides differences between `:udp` and `:tcp` sockets, so that protocol can be chosen in configuration
+  """
+
+  # Types
+
+  @type protocol :: :tcp | :udp
+
+  # Functions
+
+  @doc """
+  Closes the `protocol` `socket`
+  """
+  @spec close(map) :: :ok
+
+  def close(%{socket: socket}), do: :inet.close(socket)
+
+  @doc """
+  Opens a `protocol` socket using [`:gen_tcp.connect/3`](http://erlang.org/doc/man/gen_tcp.html#connect-3) or
+  [`:gen_udp.open/1`](http://erlang.org/doc/man/gen_udp.html#open-1).
+  """
+  @spec open(map) :: {:ok, map} | {:error, :inet.posix}
+
+  def open(map = %{host: host, port: port, protocol: :tcp}) do
+    host
+    |> :gen_tcp.connect(port, [{:active, true}, :binary, {:keepalive, true}])
+    |> put_socket(map)
+  end
+
+  def open(map = %{protocol: :udp}) do
+    0
+    |> :gen_udp.open
+    |> put_socket(map)
+  end
+
+  @doc """
+  Sends a `protocol` message over `socket`.
+  """
+  @spec send(map, iodata) :: :ok | {:error, :closed | :not_owner | :inet.posix}
+  def send(%{protocol: :tcp, socket: socket}, packet), do: :gen_tcp.send(socket, packet)
+  def send(%{host: host, port: port, protocol: :udp, socket: socket}, packet) do
+    :gen_udp.send(socket, host, port, packet)
+  end
+
+  ## Private Functions
+
+  defp put_socket({:ok, socket}, map), do: {:ok, Map.put(map, :socket, socket)}
+  defp put_socket(other, _), do: other
+end

test/logger_logstash_backend/tcp_test.exs

@@ -0,0 +1,118 @@
+################################################################################
+# Copyright 2015 Marcelo Gornstein <marcelog@gmail.com>
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+defmodule LoggerLogstashBackend.TCPTest do
+  use ExUnit.Case, async: false
+  require Logger
+  use Timex
+
+  @backend {LoggerLogstashBackend, :tcp_test}
+
+  setup context = %{line: line} do
+    # have to open socket before configure_backend, so that it is listening when connect happens
+    {:ok, listen_socket} = :gen_tcp.listen 0, [:binary, {:active, true}, {:ip, {127, 0, 0, 1}}, {:reuseaddr, true}]
+    {:ok, port} = :inet.port(listen_socket)
+
+    backend = {LoggerLogstashBackend, String.to_atom("#{inspect __MODULE__}#{line}")}
+
+    Logger.add_backend backend
+    Logger.configure_backend backend, [
+      host: "127.0.0.1",
+      port: port,
+      level: :info,
+      type: "some_app",
+      metadata: [
+        some_metadata: "go here"
+      ],
+      protocol: :tcp
+    ]
+
+    {:ok, accept_socket} = :gen_tcp.accept(listen_socket, 1_000)
+
+    on_exit fn ->
+      Logger.remove_backend backend
+      :ok = :gen_tcp.close accept_socket
+      :ok = :gen_tcp.close listen_socket
+    end
+
+    {:ok, context}
+  end
+
+  test "can log" do
+    Logger.info "hello world", [key1: "field1"]
+
+    assert {:ok, data} = JSX.decode get_log!
+    assert data["type"] === "some_app"
+    assert data["message"] === "hello world"
+
+    fields = data["fields"]
+
+    assert fields["function"] == "test can log/1"
+    assert fields["key1"] == "field1"
+    assert fields["level"] == "info"
+    assert fields["line"] == 54
+    assert fields["module"] == to_string(__MODULE__)
+    assert fields["pid"] == inspect(self)
+    assert fields["some_metadata"] == "go here"
+
+    {:ok, ts} = Timex.parse data["@timestamp"], "%FT%T%z", :strftime
+    ts = Timex.to_unix ts
+
+    now = Timex.to_unix Timex.DateTime.local
+    assert (now - ts) < 1000
+  end
+
+  test "can log pids" do
+    Logger.info "pid", [pid_key: self]
+
+    {:ok, data} = JSX.decode get_log!
+    assert data["type"] === "some_app"
+    assert data["message"] === "pid"
+
+    fields = data["fields"]
+
+    assert fields["function"] == "test can log pids/1"
+    assert fields["pid_key"] == inspect(self)
+    assert fields["level"] == "info"
+    assert fields["line"] == 78
+    assert fields["module"] == to_string(__MODULE__)
+    assert fields["pid"] == inspect(self)
+    assert fields["some_metadata"] == "go here"
+
+    {:ok, ts} = Timex.parse data["@timestamp"], "%FT%T%z", :strftime
+    ts = Timex.to_unix ts
+
+    now = Timex.to_unix Timex.DateTime.local
+    assert (now - ts) < 1000
+  end
+
+  test "cant log when minor levels" do
+    Logger.debug "hello world", [key1: "field1"]
+    {:error, :nothing_received} = get_log
+  end
+
+  defp get_log do
+    receive do
+      {:tcp, _socket, json} -> {:ok, json}
+    after 500 -> {:error, :nothing_received}
+    end
+  end
+
+  defp get_log! do
+    {:ok, log} = get_log
+
+    log
+  end
+end