Initial commit.

.gitignore

@@ -0,0 +1,8 @@
+*~
+*.beam
+.gitignore
+ebin/*.app
+c_src/*.o
+.eunit/*
+.#*
+deps/*

include/logsplat_log.hrl

@@ -0,0 +1,18 @@
+%%% Author : Geoff Cant <nem@erlang.geek.nz>
+%%% Description : Logging macros
+%%% Created : 13 Jan 2006 by Geoff Cant <nem@erlang.geek.nz>
+
+-ifndef(logging_macros).
+-define(logging_macros, true).
+
+-define(INFO(Format, Args),
+        error_logger:info_msg("(~p ~p:~p) " ++ Format ++ "~n",
+                              [self(), ?MODULE, ?LINE | Args])).
+-define(WARN(Format, Args),
+        error_logger:warning_msg("(~p ~p:~p) " ++ Format ++ "~n",
+                                 [self(), ?MODULE, ?LINE | Args])).
+-define(ERR(Format, Args),
+        error_logger:error_msg("(~p ~p:~p) " ++ Format ++ "~n",
+                               [self(), ?MODULE, ?LINE | Args])).
+
+-endif. %logging

rebar.config

@@ -0,0 +1,4 @@
+{erl_opts, [debug_info]}.
+{deps, [{cowboy, "", {git, "git://github.com/heroku/cowboy.git", "heroku_unstable"}}
+        ,{ex_uri, "", {git, "git://github.com/heroku/ex_uri.git", "master"}}
+       ]}.

src/logsplat.app.src

@@ -0,0 +1,15 @@
+{application, logsplat,
+ [{description, "Logplex protocol injector."}
+  ,{vsn, "0.1.0"}
+  ,{registered, []}
+  ,{applications,
+    [kernel
+     ,stdlib
+     ,ssl
+     ,cowboy
+    ]}
+  ,{mod, { logsplat_app, []}}
+  ,{env,
+    [
+    ]}
+ ]}.

src/logsplat.erl

@@ -0,0 +1,44 @@
+%%%-------------------------------------------------------------------
+%% @copyright Geoff Cant
+%% @author Geoff Cant <nem@erlang.geek.nz>
+%% @version {@vsn}, {@date} {@time}
+%% @doc logsplat public API
+%% @end
+%%%-------------------------------------------------------------------
+-module(logsplat).
+
+-include_lib("ex_uri/include/ex_uri.hrl").
+
+%% API
+-export([ heroku_test/3
+          ,heroku_test_req/3
+        ]).
+
+%%====================================================================
+%% API
+%%====================================================================
+
+heroku_test(Count, Token, Url) when is_list(Url),
+                                    is_integer(Count),
+                                    Count > 0->
+    RawReq = lsp_request:to_iolist(heroku_test_req(Count, Token, Url)),
+    {ok, Uri = #ex_uri{}, _} = ex_uri:decode(Url),
+    {ok, Pid} = lsp_http_client:start_link(Uri),
+    {Pid,
+     lsp_http_client:raw_request(Pid, RawReq, 5000)}.
+
+heroku_test_req(Count, Token, Url) ->
+    {ok,
+     Uri0 = #ex_uri{authority = Auth},
+     _} = ex_uri:decode(Url),
+    NewAuth = Auth#ex_uri_authority{userinfo = "token:" ++ Token},
+    Uri = Uri0#ex_uri{authority = NewAuth},
+    Msgs = [ lsp_msg:heroku(now, Token, "console.1",
+                            io_lib:format("Logsplat test message ~p from ~p.",
+                                          [N, self()]))
+             || N <- lists:seq(1, Count) ],
+    lsp_request:new(Msgs, Uri).
+
+%%====================================================================
+%% Internal functions
+%%====================================================================

src/logsplat_app.erl

@@ -0,0 +1,67 @@
+%%%-------------------------------------------------------------------
+%% @copyright Geoff Cant
+%% @author Geoff Cant <nem@erlang.geek.nz>
+%% @version {@vsn}, {@date} {@time}
+%% @doc logsplat OTP App callback module.
+%% @end
+%%%-------------------------------------------------------------------
+
+-module(logsplat_app).
+
+-behaviour(application).
+
+-define(APP, logsplat).
+
+%% Application callbacks
+-export([start/2, stop/1]).
+
+-export([config/0, config/1, config/2,
+         start/0, a_start/2]).
+
+%%%===================================================================
+%%% Convenience Functions
+%%%===================================================================
+
+start() ->
+    a_start(?APP, permanent).
+
+config(Key, Default) ->
+    case application:get_env(?APP, Key) of
+        undefined -> Default;
+        {ok, Val} -> Val
+    end.
+
+config(Key) ->
+    case application:get_env(?APP, Key) of
+        undefined -> erlang:error({missing_config, Key});
+        {ok, Val} -> Val
+    end.
+
+config() ->
+    application:get_all_env(?APP).
+
+
+%% ===================================================================
+%% Application callbacks
+%% ===================================================================
+
+start(_StartType, _StartArgs) ->
+    logsplat_sup:start_link().
+
+stop(_State) ->
+    ok.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+a_start(App, Type) ->
+    start_ok(App, Type, application:start(App, Type)).
+
+start_ok(_App, _Type, ok) -> ok;
+start_ok(_App, _Type, {error, {already_started, _App}}) -> ok;
+start_ok(App, Type, {error, {not_started, Dep}}) ->
+    ok = a_start(Dep, Type),
+    a_start(App, Type);
+start_ok(App, _Type, {error, Reason}) ->
+    erlang:error({app_start_failed, App, Reason}).

src/logsplat_sup.erl

@@ -0,0 +1,41 @@
+%%%-------------------------------------------------------------------
+%% @copyright Geoff Cant
+%% @author Geoff Cant <nem@erlang.geek.nz>
+%% @version {@vsn}, {@date} {@time}
+%% @doc logsplat top level supervisor
+%% @end
+%%%-------------------------------------------------------------------
+-module(logsplat_sup).
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+%%====================================================================
+%% API functions
+%%====================================================================
+%%--------------------------------------------------------------------
+%% @spec start_link() -> {ok,Pid} | ignore | {error,Error}
+%% @doc: Starts the supervisor
+%% @end
+%%--------------------------------------------------------------------
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+%%====================================================================
+%% Supervisor callbacks
+%%====================================================================
+
+%% Child :: {Id,StartFunc,Restart,Shutdown,Type,Modules}
+init([]) ->
+    {ok, { {one_for_all, 0, 1}, []} }.
+
+%%====================================================================
+%% Internal functions
+%%====================================================================

src/lsp_http_client.erl

@@ -0,0 +1,174 @@
+%%%-------------------------------------------------------------------
+%% @copyright Geoff Cant
+%% @author Geoff Cant <nem@erlang.geek.nz>
+%% @version {@vsn}, {@date} {@time}
+%% @doc
+%% @end
+%%%-------------------------------------------------------------------
+-module(lsp_http_client).
+
+-include("logsplat_log.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("ex_uri/include/ex_uri.hrl").
+
+%% API
+-export([start_link/7
+         ,raw_request/3
+         ,close/1
+         ,start_link/1
+        ]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-record(state, {drain_id, channel_id, dest,
+                client}).
+
+%%====================================================================
+%% API
+%%====================================================================
+
+start_link(Uri = #ex_uri{scheme = Scheme,
+                         authority = #ex_uri_authority{host = Host,
+                                                       port = Port}}) ->
+    start_link(1, 1, ex_uri:encode(ex_uri:hide_userinfo(Uri)),
+               Scheme, Host,
+               find_port(Scheme, Port), 5000).
+
+start_link(DrainId, ChannelId, Dest, Scheme, Host, Port, Timeout) ->
+    gen_server:start_link(?MODULE,
+                          [#state{drain_id=DrainId,
+                                  channel_id=ChannelId,
+                                  dest=Dest},
+                           Scheme, Host, Port],
+                          [{timeout, Timeout}]).
+
+raw_request(Pid, Req, Timeout) ->
+    gen_server:call(Pid, {raw_request, Req}, Timeout).
+
+close(Pid) ->
+    gen_server:cast(Pid, close),
+    timer:kill_after(timer:seconds(1), Pid), % XXX - necessary?
+    ok.
+
+%%====================================================================
+%% gen_server callbacks
+%%====================================================================
+
+%% @private
+init([State = #state{},
+      Scheme, Host, Port]) ->
+    {ok, Client0} = client_init(Scheme),
+    ConnectStart = os:timestamp(),
+    try cowboy_client:connect(scheme_to_transport(Scheme),
+                              Host, Port, Client0) of
+        {ok, Client} ->
+            ConnectEnd = os:timestamp(),
+            ?INFO("drain_id=~p channel_id=~p dest=~s at=try_connect "
+                  "attempt=success connect_time=~p",
+                  log_info(State, [ltcy(ConnectStart, ConnectEnd)])),
+            {ok, State#state{client=Client}};
+        {error, Why} ->
+            ConnectEnd = os:timestamp(),
+            ?WARN("drain_id=~p channel_id=~p dest=~s at=try_connect "
+                  "attempt=fail connect_time=~p tcp_err=~1000p",
+                  log_info(State, [ltcy(ConnectStart, ConnectEnd), Why])),
+            ignore
+    catch
+        Class:Err ->
+            Report = {Class, Err, erlang:get_stacktrace()},
+            ConnectEnd = os:timestamp(),
+            ?WARN("drain_id=~p channel_id=~p dest=~s at=connect "
+                  "attempt=fail err=exception connect_time=~p "
+                  "next_state=disconnected "
+                  "data=~1000p",
+                  log_info(State, [ltcy(ConnectStart, ConnectEnd), Report])),
+            ignore
+    end.
+
+%% @private
+handle_call({raw_request, Req}, _From, State) ->
+    ReqStart = os:timestamp(),
+    case raw_request(Req, State) of
+        {ok, Status, Headers, NewState} ->
+            ReqEnd = os:timestamp(),
+            ?INFO("drain_id=~p channel_id=~p dest=~s at=response "
+                  "result=success status=~p req_time=~p",
+                  log_info(State, [Status, ltcy(ReqStart, ReqEnd)])),
+            {reply, {ok, Status, Headers}, NewState};
+        {error, Why} = Err ->
+            ReqEnd = os:timestamp(),
+            ?WARN("drain_id=~p channel_id=~p dest=~s at=response"
+                  " result=error req_time=~p tcp_err=\"~1000p\"",
+                  log_info(State, [ltcy(ReqStart, ReqEnd), Why])),
+            {stop, normal, Err, State}
+    end;
+
+handle_call(Call, _From, State) ->
+    ?WARN("Unexpected call ~p.", [Call]),
+    {noreply, State}.
+
+%% @private
+handle_cast(close, State = #state{client=Client}) ->
+    cowboy_client:close(Client),
+    {stop, normal, State};
+
+handle_cast(Msg, State) ->
+    ?WARN("Unexpected cast ~p", [Msg]),
+    {noreply, State}.
+
+%% @private
+handle_info(Info, State) ->
+    ?WARN("Unexpected info ~p", [Info]),
+    {noreply, State}.
+
+%% @private
+terminate(_Reason, _State) ->
+    ok.
+
+%% @private
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+
+%%--------------------------------------------------------------------
+%%% Internal functions
+%%--------------------------------------------------------------------
+
+scheme_to_transport("https") -> ranch_ssl;
+scheme_to_transport("http") -> ranch_tcp.
+
+client_init("http") ->
+    cowboy_client:init([]);
+client_init("https") ->
+    cowboy_client:init([{reuse_sessions, false}
+                       ]).
+
+find_port(_, Port) when is_integer(Port) -> Port;
+find_port("http", _) -> 80;
+find_port("https", _) -> 443.
+
+log_info(#state{drain_id=DrainId, channel_id=ChannelId, dest=Dest}, Rest)
+  when is_list(Rest) ->
+    [DrainId, ChannelId, Dest | Rest].
+
+
+ltcy(Start, End) ->
+    timer:now_diff(End, Start).
+
+raw_request(Request, State = #state{client=Client}) ->
+    try
+        {ok, Client2} =
+            cowboy_client:raw_request(Request, Client),
+        case cowboy_client:response(Client2) of
+            {ok, Status, Headers, NewClient} ->
+                {ok, Status, Headers,
+                 State#state{client = NewClient}};
+            {error, _Why} = Err ->
+                Err
+        end
+    catch
+        Class:Ex ->
+            {error, {Class, Ex, erlang:get_stacktrace()}}
+    end.