Set up flushing to allow for alternate logic.

I always have hated that the default mondemand-server config turned off
flushing in the library in order to replicate and do it's own.  So now
the flush functionality is pluggable, and the server just swaps out it's
own.  This also allows for consistent sending of vmstats (if enabled),
in all cases.

ChangeLog

@@ -1,3 +1,9 @@
+Version 6.7.0 (molinaro)
+  * generalize flush functionality so that the server can plug in a slightly
+  modified version.
+  * cleanup some config, all application config should be read only in
+  mondemand_config
+
 Version 6.6.1 (molinaro)
   * make sure that mondemand vmstats metrics are not emitted without a full
   interval having gone by.  This means a restart of a service can lead to

src/mondemand.app.src

@@ -1,7 +1,7 @@
 { application, mondemand,
   [
    { description, "Erlang Mondemand Bindings." },
-   { vsn, "6.6.1" },
+   { vsn, "6.7.0" },
    { modules, [] },
    { registered, [mondemand,mondemand_sup]},
    { applications, [kernel,stdlib,syntax_tools,lwes,inets]},

src/mondemand.erl

@@ -66,6 +66,7 @@
 
            % other functions
            send_stats/3,
+           flush_one_stat/2,
            reset_stats/0,
            stats/0,
            all/0,
@@ -90,7 +91,8 @@
                   jitter,
                   timer,
                   channels = dict:new(),
-                  http_config
+                  http_config,
+                  flush_config
                 }).
 
 %-=====================================================================-
@@ -302,15 +304,26 @@ send_stats (ProgId, Context, Stats) ->
     ),
   send_event (Event).
 
-flush () ->
+flush ({FlushModule, FlushStatePrepFunction, FlushFunction}) ->
   case mondemand_config:vmstats_prog_id () of
     undefined -> ok;
     ProgId ->
       Context = mondemand_config:vmstats_context (),
       VmStats = mondemand_vmstats:to_mondemand (),
       send_stats (ProgId, Context, VmStats)
   end,
-  mondemand_statdb:flush (1, fun flush_one/1).
+  % allow some initialization of state to be sent to each invocation
+  UserState = case FlushStatePrepFunction of
+                undefined -> undefined;
+                _ ->  FlushModule:FlushStatePrepFunction()
+              end,
+  mondemand_statdb:flush (1,
+                          fun (StatsMsg) ->
+                            erlang:apply (FlushModule,
+                                          FlushFunction,
+                                          [UserState, StatsMsg])
+                          end
+                         ).
 
 reset_stats () ->
   mondemand_statdb:reset_stats().
@@ -329,11 +342,7 @@ current_config () ->
 %-=====================================================================-
 init([]) ->
 
-  IntervalSecs =
-    case application:get_env (mondemand, send_interval) of
-      {ok, I} when is_integer (I) -> I;
-      _ -> ?MD_DEFAULT_SEND_INTERVAL
-    end,
+  IntervalSecs = mondemand_config:send_interval(),
 
   % use milliseconds for interval
   Interval = IntervalSecs * 1000,
@@ -343,6 +352,7 @@ init([]) ->
       {error, _} -> undefined;
       HC -> HC
     end,
+  FlushConfig = mondemand_config:flush_config(),
 
   case mondemand_config:lwes_config () of
     {error, Error} ->
@@ -373,7 +383,8 @@ init([]) ->
               jitter = Jitter,
               interval = Interval,
               channels = ChannelDict,
-              http_config = HttpConfig
+              http_config = HttpConfig,
+              flush_config = FlushConfig
             }
           };
         {error, Error} ->
@@ -441,12 +452,15 @@ emit_one (Name, Event, State = #state { channels = ChannelDict }) ->
       State
   end.
 
-handle_info (flush, State = #state {timer = undefined, interval = Interval}) ->
+handle_info (flush, State = #state {timer = undefined,
+                                    interval = Interval,
+                                    flush_config = FlushConfig
+                                   }) ->
   {ok, TRef} = timer:send_interval (Interval, ?MODULE, flush),
-  flush (),
+  flush (FlushConfig),
   {noreply, State#state {timer = TRef}};
-handle_info (flush, State = #state {}) ->
-  flush (),
+handle_info (flush, State = #state {flush_config = FlushConfig}) ->
+  flush (FlushConfig),
   {noreply, State#state {}};
 handle_info (_Info, State) ->
   {noreply, State}.
@@ -466,8 +480,8 @@ send_event (Events) when is_list (Events) ->
 send_event (Event = #lwes_event { name = Name }) ->
   gen_server:cast (?MODULE, {send, Name, lwes_event:to_binary (Event)}).
 
-flush_one (Stats) ->
-  send_event (mondemand_statsmsg:to_lwes (Stats)).
+flush_one_stat (_, StatsMsg) ->
+  send_event (mondemand_statsmsg:to_lwes (StatsMsg)).
 
 open_all (Config) ->
   lists:foldl (fun ({T,C},{ok, D}) ->

src/mondemand_config.erl

@@ -21,6 +21,7 @@
            host/0,
            default_max_sample_size/0,
            default_stats/0,
+           send_interval/0,
            lwes_config/0,
            minutes_to_keep/0,
            max_metrics/0,
@@ -29,15 +30,19 @@
            vmstats_enabled/0,
            vmstats_prog_id/0,
            vmstats_context/0,
-           vmstats_disable_scheduler_wall_time/0
+           vmstats_disable_scheduler_wall_time/0,
+           vmstats_legacy_workaround/0,
+           flush_config/0
          ]).
 
+-define (DEFAULT_SEND_INTERVAL, 60).
 -define (DEFAULT_MAX_SAMPLE_SIZE, 10).
 -define (DEFAULT_STATS, [min, max, sum, count]).
 -define (DEFAULT_MINUTES_TO_KEEP, 10).
+-define (DEFAULT_MAX_METRICS, 512).
+
 -define (MOCHI_SENDER_HOST, mondemand_sender_host_global).
 -define (MOCHI_MAX_METRICS, mondemand_max_metrics_global).
--define (DEFAULT_MAX_METRICS, 512).
 
 % this function is meant to be called before the supervisor and
 % pulls all those configs which are mostly static.
@@ -70,6 +75,12 @@ default_stats () ->
     {ok, L} when is_list (L) -> L
   end.
 
+send_interval () ->
+  case application:get_env (mondemand, send_interval) of
+    {ok, I} when is_integer (I) -> I;
+    _ -> ?DEFAULT_SEND_INTERVAL
+  end.
+
 lwes_config () ->
   % lwes config can be specified in one of two ways
   %   1. the lwes_channel application variable
@@ -297,6 +308,12 @@ vmstats_context () ->
 vmstats_disable_scheduler_wall_time () ->
   vmstats_config (disable_scheduler_wall_time, false).
 
+vmstats_legacy_workaround () ->
+  case application:get_env(mondemand,r15b_workaround) of
+    {ok, true} -> true;
+    _ -> false
+  end.
+
 vmstats_config (K, Default) ->
   case application:get_env (mondemand, vmstats) of
     {ok, L} when is_list (L) ->
@@ -307,6 +324,13 @@ vmstats_config (K, Default) ->
     _ -> Default
   end.
 
+flush_config () ->
+  case application:get_env (mondemand, flush_config) of
+    {ok, {Module, FlushStatePrepFunction, FlushFunction}} ->
+      {Module, FlushStatePrepFunction, FlushFunction};
+    undefined ->
+      {mondemand, undefined, flush_one_stat}
+  end.
 
 %%--------------------------------------------------------------------
 %%% Test functions

src/mondemand_internal.hrl

@@ -4,7 +4,6 @@
 -include("mondemand.hrl").
 
 % global defaults
--define (MD_DEFAULT_SEND_INTERVAL, 60).
 
 % these are common fields in several lwes events
 -define (MD_RECEIPT_TIME,  <<"ReceiptTime">>).

src/mondemand_vmstats.erl

@@ -127,11 +127,7 @@ to_list() ->
 %-=====================================================================-
 init([]) ->
   % work around for the fact that R15B didn't have port_count
-  Legacy =
-    case application:get_env(mondemand,r15b_workaround) of
-      {ok, true} -> true;
-      _ -> false
-    end,
+  Legacy = mondemand_config:vmstats_legacy_workaround(),
 
   % allow scheduler stats to be turned off (should default to true)
   {Former, CollectSchedulerStats} =