Honor UDP packet size (MTU) limit

UDP packets cannot exceed MTU. If they do, the app gets
{error,emsgsize} from gen_udp:send. Similar issue:

jaegertracing/jaeger-client-node#124

This commit adds packet size check. Packets are sent
immediately upon reaching max size, even before the end
of batch window.

Size calculation from CollectedMetrics map would've been
costly. I changed the format to binary. Metrics are encoded
into Line Protocol immediately now, instead of right before
sending.

This also helps avoid data loss. The map did not record
all timestamps. It set the most recent timestamp
for all fields of the same measurement.

It also did not honor differences in tags, but I guess
there aren't any when metrics go through Exometer.

README.md

@@ -39,14 +39,15 @@ This reporter pushes data to [InfluxDB](https://influxdb.com/index.html).
 
 Available options:
 
+* __protocol__ - `http`, `https` or `udp` for operating with InfluxDB. `http` by default. If you use `udp`, check __udp_mtu__ below to avoid `{error,emsgsize}`.
 * __host__ - InfluxDB host. `127.0.0.1` by default.
-* __protocol__ - `http`, `https` or `udp` for operating with InfluxDB. `http` by default.
 * __port__ - InfluxDB port. `8086` by default.
 * __db__ - Database on InfluxDB for writing data. `exometer` by default.
 * __username__ - Username for authorization on InfluxDB.
 * __password__ - Password for authorization on InfluxDB.
 * __timestamping__ - Enable timestamping, `false` by default. To enable `timestamping` with the reporter you can use `true` or `{true, Precision}` where `Precision` is a unit taken from `[n,u,ms,s,m,h]`. The default unit is `u`.
-* __batch_window_size__ - set window size in ms for batch sending. This means the reporter will collect measurements within this interval and send all measurements in one packet. `0` by default.
+* __batch_window_size__ - Set window size in ms for batch sending. This means the reporter will collect measurements within this interval and send all measurements in one packet. `0` by default.
+* __udp_mtu__ - (Used only with __protocol__ == `udp`.) MTU of the network interface through which UDP packets flow to the __host__. `65536` by default (Linux loopback interface MTU). Run `ifconfig` on the machine where this app will run to find it out. Metrics will be sent out if their size (in the Line Protocol format) exceeds this value, even if the current __batch_window_size__ is not over yet. (They will also be sent out at the end of __batch_window_size__ as usual, regardless of their size.)
 
 The following options can be set globally in the reporter config or locally in a specific subscription. The latter case overwrites the first.
 

src/exometer_report_influxdb.erl

@@ -33,11 +33,17 @@
 -define(DEFAULT_FORMATTING, []).
 -define(DEFAULT_TIMESTAMP_OPT, false).
 -define(DEFAULT_BATCH_WINDOW_SIZE, 0).
+-define(DEFAULT_UDP_MTU, 65536).
 -define(DEFAULT_AUTOSUBSCRIBE, false).
 -define(DEFAULT_SUBSCRIPTIONS_MOD, undefined).
 
 -define(VALID_PRECISIONS, [n, u, ms, s, m, h]).
 
+% https://en.wikipedia.org/wiki/User_Datagram_Protocol#Packet_structure
+-define(MAX_UDP_PACKET_SIZE, 65535).
+-define(UDP_HEADER_SIZE, 8).
+-define(IP_HEADER_SIZE, 20).
+
 -define(HTTP(Proto), (Proto =:= http orelse Proto =:= https)).
 
 -include("log.hrl").
@@ -56,8 +62,9 @@
                 port :: inet:port_number(),  % for udp
                 timestamping :: boolean(),
                 precision :: precision(),
-                collected_metrics = #{} :: map(),
-                batch_window_size = 0 :: integer(),
+                collected_metrics = <<>> :: binary(),
+                batch_window_size = ?DEFAULT_BATCH_WINDOW_SIZE :: pos_integer(),
+                max_udp_size = ?DEFAULT_UDP_MTU :: pos_integer(),
                 tags :: map(),
                 series_name :: atom() | binary(),
                 formatting :: list(),
@@ -81,6 +88,7 @@ exometer_init(Opts) ->
     Password = get_opt(password, Opts, ?DEFAULT_PASSWORD),
     TimestampOpt = get_opt(timestamping, Opts, ?DEFAULT_TIMESTAMP_OPT),
     BatchWinSize = get_opt(batch_window_size, Opts, ?DEFAULT_BATCH_WINDOW_SIZE),
+    UDP_MTU = get_opt(udp_mtu, Opts, ?DEFAULT_UDP_MTU),
     {Timestamping, Precision} = evaluate_timestamp_opt(TimestampOpt),
     Tags = [{key(Key), Value} || {Key, Value} <- get_opt(tags, Opts, [])],
     SeriesName = get_opt(series_name, Opts, ?DEFAULT_SERIES_NAME),
@@ -100,6 +108,7 @@ exometer_init(Opts) ->
                     series_name = SeriesName,
                     formatting = Formatting,
                     batch_window_size = BatchWinSize,
+                    max_udp_size = max_udp_size(UDP_MTU),
                     autosubscribe = Autosubscribe,
                     subscriptions_module = SubscriptionsMod,
                     metrics = maps:new()},
@@ -128,7 +137,7 @@ exometer_report(Metric, DataPoint, _Extra, Value,
                 #state{metrics = Metrics} = State) ->
     case maps:get(Metric, Metrics, not_found) of
         {MetricName, Tags} ->
-            maybe_send(Metric, MetricName, Tags,
+            maybe_send(MetricName, Tags,
                        maps:from_list([{DataPoint, Value}]), State);
         Error ->
             ?warning("InfluxDB reporter got trouble when looking ~p metric's tag: ~p",
@@ -175,13 +184,9 @@ exometer_cast(_Unknown, State) ->
 exometer_info({exometer_influxdb, reconnect}, State) ->
     reconnect(State);
 exometer_info({exometer_influxdb, send}, 
-              #state{precision = Precision,
-                     collected_metrics = CollectedMetrics} = State) ->
-    if CollectedMetrics /= #{} ->
-        Packets = [make_packet(MetricName, Tags, Fileds, Timestamping, Precision) ++ "\n"
-                   || {_, {MetricName, Tags, Fileds, Timestamping}} 
-                      <- maps:to_list(CollectedMetrics)],
-        send(Packets, State#state{collected_metrics = #{}});
+              #state{collected_metrics = CollectedMetrics} = State) ->
+    if size(CollectedMetrics) > 0 ->
+        send(CollectedMetrics, State#state{collected_metrics = <<>>});
     true -> {ok, State}
     end;
 exometer_info(_Unknown, State) ->
@@ -272,36 +277,35 @@ prepare_batch_send(Time) ->
 prepare_reconnect() ->
     erlang:send_after(1000, self(), {exometer_influxdb, reconnect}).
 
--spec maybe_send(list(), list(), map(), map(), state()) ->
+-spec maybe_send(list(), map(), map(), state()) ->
     {ok, state()} | {error, term()}.
-maybe_send(OriginMetricName, MetricName, Tags0, Fields, 
-           #state{batch_window_size = BatchWinSize, 
+maybe_send(MetricName, Tags, Fields,
+    #state{batch_window_size = 0,
+           precision = Precision,
+           timestamping = Timestamping} = State) ->
+    Packet = make_packet(MetricName, Tags, Fields, Timestamping andalso unix_time(Precision), Precision),
+    send(Packet, State);
+maybe_send(MetricName, Tags, Fields,
+           #state{protocol = Protocol,
+                  batch_window_size = BatchWindowSize,
+                  max_udp_size = MaxUDPSize,
                   precision = Precision,
                   timestamping = Timestamping,
-                  collected_metrics = CollectedMetrics} = State)
-  when BatchWinSize > 0 ->
-    NewCollectedMetrics = case maps:get(OriginMetricName, CollectedMetrics, not_found) of
-        {MetricName, Tags, Fields1} ->
-            NewFields = maps:merge(Fields, Fields1),
-            maps:put(OriginMetricName, 
-                     {MetricName, Tags, NewFields, Timestamping andalso unix_time(Precision)}, 
-                     CollectedMetrics);
-        {MetricName, Tags, Fields1, _OrigTimestamp} ->
-            NewFields = maps:merge(Fields, Fields1),
-            maps:put(OriginMetricName,
-                     {MetricName, Tags, NewFields, Timestamping andalso unix_time(Precision)},
-                     CollectedMetrics);
-        not_found -> 
-            maps:put(OriginMetricName, 
-                     {MetricName, Tags0, Fields, Timestamping andalso unix_time(Precision)}, 
-                     CollectedMetrics)
-    end,
-    maps:size(CollectedMetrics) == 0 andalso prepare_batch_send(BatchWinSize),
-    {ok, State#state{collected_metrics = NewCollectedMetrics}};
-maybe_send(_, MetricName, Tags, Fields,
-           #state{timestamping = Timestamping, precision = Precision} = State) ->
-    Packet = make_packet(MetricName, Tags, Fields, Timestamping, Precision),
-    send(Packet, State).
+                  collected_metrics = CollectedMetrics} = State) ->
+    maybe_start_new_window(BatchWindowSize, CollectedMetrics),
+    Packet = make_packet(MetricName, Tags, Fields, Timestamping andalso unix_time(Precision), Precision),
+    BinaryPacket = list_to_binary(Packet),
+    NewCollectedMetrics = <<CollectedMetrics/binary, BinaryPacket/binary, "~n">>,
+    if
+        Protocol == udp andalso size(CollectedMetrics) > 0 andalso size(NewCollectedMetrics) > MaxUDPSize ->
+            send(CollectedMetrics, State#state{collected_metrics = BinaryPacket});
+        true ->
+            {ok, State#state{collected_metrics = NewCollectedMetrics}}
+    end.
+
+maybe_start_new_window(Window, Metrics) when size(Metrics) == 0 ->
+    prepare_batch_send(Window);
+maybe_start_new_window(_, _) -> ok.
 
 -spec send(binary() | list(), state()) ->
     {ok, state()} | {error, term()}.
@@ -338,6 +342,11 @@ send(Packet, #state{protocol = udp, connection = Socket,
     end;
 send(_, #state{protocol = Protocol}) -> {error, {Protocol, not_supported}}.
 
+max_udp_size(MTU) when MTU > ?MAX_UDP_PACKET_SIZE ->
+    max_udp_size(?MAX_UDP_PACKET_SIZE);
+max_udp_size(MaxPacketSize) ->
+    MaxPacketSize - ?UDP_HEADER_SIZE - ?IP_HEADER_SIZE.
+
 -spec merge_tags(list() | map(), list() | map()) -> map().
 merge_tags(Tags, AdditionalTags) when is_list(Tags) ->
     merge_tags(maps:from_list(Tags), AdditionalTags);