src/lwes_journaller.erl

   1 %
   2 % This module implements an lwes journaller.
   3 %
   4 % configuration is as follows
   5 %
   6 % [ { root, "." },                      % journal root
   7 %   { name, "all_events.log" },         % journal name
   8 %   { interval, <rotation_interval> },  % interval for jouirnal file rotation
   9 % ]
  10
  11 -module (lwes_journaller).
  12
  13 -behaviour (gen_server).
  14
  15 %% API
  16 -export ([ start_link/1,
  17            process_event/2,
  18            rotate/1 ]).
  19
  20 %% gen_server callbacks
  21 -export ([ init/1,
  22            handle_call/3,
  23            handle_cast/2,
  24            handle_info/2,
  25            terminate/2,
  26            code_change/3
  27          ]).
  28
  29 -record (state, {
  30                   journal_root,
  31                   journal_file_name,
  32                   journal_file_ext,
  33                   journal_current,
  34                   journal_last_rotate,
  35                   timer
  36                 }).
  37
  38 %%====================================================================
  39 %% API
  40 %%====================================================================
  41 start_link (Config) ->
  42   gen_server:start_link ( ?MODULE, [Config], []).
  43
  44 process_event (Event, Pid) ->
  45   gen_server:cast (Pid, {process, Event}),
  46   Pid.
  47
  48 rotate (Pid) ->
  49   gen_server:cast (Pid, {rotate}).
  50
  51 %%====================================================================
  52 %% gen_server callbacks
  53 %%====================================================================
  54 init ([Config]) ->
  55   % get appication variables
  56   Root = proplists:get_value (root, Config, "."),
  57   Name = proplists:get_value (name, Config, "all_events.log"),
  58   Interval = proplists:get_value (interval, Config, 60),
  59   Ext = "gz",
  60
  61   % I want terminate to be called
  62   process_flag (trap_exit, true),
  63
  64   % open journal file
  65   { ok, File } = open (Root, Name, Ext),
  66
  67   % setup rotation of journal
  68   { ok, TRef } =
  69     timer:apply_interval (Interval * 1000, ?MODULE, rotate, [self()]),
  70
  71   { ok, #state {
  72           journal_root        = Root,
  73           journal_file_name   = Name,
  74           journal_file_ext    = Ext,
  75           journal_current     = File,
  76           journal_last_rotate = seconds_since_epoch (),
  77           timer               = TRef
  78         }
  79   }.
  80
  81 handle_call (Request, From, State) ->
  82   error_logger:warning_msg ("Unrecognized call ~p from ~p~n",[Request, From]),
  83   { reply, ok, State }.
  84
  85 handle_cast ( {process, {udp, _, {V1, V2, V3, V4}, P, B}},
  86               State = #state { journal_current = Journal }) ->
  87   S = byte_size (B),
  88   M = milliseconds_since_epoch (),
  89   I = 1,
  90   ok = file:write ( Journal,
  91                     [ <<S:16/integer-unsigned-big,  % 2
  92                         M:64/integer-unsigned-big,  % 8
  93                         V4:8/integer-unsigned-big,  % 1
  94                         V3:8/integer-unsigned-big,  % 1
  95                         V2:8/integer-unsigned-big,  % 1
  96                         V1:8/integer-unsigned-big,  % 1
  97                         P:16/integer-unsigned-big,  % 2
  98                         I:16/integer-unsigned-big,  % 2
  99                         0:32/integer-signed-big     % 4
 100                       >>,
 101                       B
 102                     ]),
 103   { noreply, State };
 104 handle_cast ( {rotate}, State = #state {
 105                                   journal_root = Root,
 106                                   journal_file_name = Name,
 107                                   journal_file_ext  = Ext,
 108                                   journal_current = File,
 109                                   journal_last_rotate = LastRotate
 110                                 }) ->
 111   file:close (File),
 112   rename (Root, Name, Ext, LastRotate),
 113   {ok, NewFile} = open (Root, Name, Ext),
 114   { noreply, State#state { journal_current = NewFile,
 115                            journal_last_rotate = seconds_since_epoch () }};
 116 handle_cast (Request, State) ->
 117   error_logger:warning_msg ("Unrecognized cast ~p~n",[Request]),
 118   { noreply, State }.
 119
 120 handle_info (Request, State) ->
 121   error_logger:warning_msg ("Unrecognized info ~p~n",[Request]),
 122   {noreply, State}.
 123
 124 terminate (_Reason, #state {
 125                       journal_root = Root,
 126                       journal_file_name = Name,
 127                       journal_file_ext  = Ext,
 128                       journal_current = File,
 129                       journal_last_rotate = LastRotate
 130                     }) ->
 131   file:close (File),
 132   rename (Root, Name, Ext, LastRotate),
 133   ok.
 134
 135 code_change (_OldVsn, State, _Extra) ->
 136   {ok, State}.
 137
 138 %%====================================================================
 139 %% Internal
 140 %%====================================================================
 141 open (Root, Name, Ext) ->
 142   JournalFile = filename:join ([Root, string:join ([Name, Ext],".")]),
 143   file:open (JournalFile, [ write, raw, compressed ]).
 144
 145 rename (Root, Name, Ext, LastRotate) ->
 146   {{Year,Month,Day},{Hour,Minute,Second}} =
 147     calendar:now_to_universal_time(os:timestamp()),
 148   NewFile =
 149     filename:join
 150       ([Root,
 151         io_lib:format
 152           ("~s.~4.10.0B~2.10.0B~2.10.0B~2.10.0B~2.10.0B~2.10.0B.~b.~b.~s",
 153             [ Name, Year, Month, Day, Hour, Minute, Second, LastRotate,
 154               seconds_since_epoch(), Ext])]),
 155   CurrentFile = filename:join ([Root, string:join ([Name, Ext],".")]),
 156   error_logger:info_msg("rename ~p -> ~p",[CurrentFile, NewFile]),
 157   ok = file:rename (CurrentFile, NewFile).
 158
 159 milliseconds_since_epoch () ->
 160   {Meg, Sec, Mic} = os:timestamp(),
 161   trunc (Meg * 1000000000 + Sec * 1000 + Mic / 1000).
 162
 163 seconds_since_epoch () ->
 164   {M, S, _ } = os:timestamp(),
 165   M*1000000+S.
 166
 167 %%====================================================================
 168 %% Test functions
 169 %%====================================================================
 170 -ifdef (TEST).
 171 -include_lib ("eunit/include/eunit.hrl").
 172
 173 -endif.