From c5bd4670425cc59d2f5a53f1c2c21d4b7e869455 Mon Sep 17 00:00:00 2001 From: Ilya Vassilevsky Date: Fri, 11 Jan 2019 03:46:11 +0400 Subject: [PATCH] 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: https://github.com/jaegertracing/jaeger-client-node/issues/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 | 5 +- src/exometer_report_influxdb.erl | 83 ++++++++++++++++++-------------- 2 files changed, 49 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index 1516d6c..5f75e1c 100644 --- a/README.md +++ b/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. diff --git a/src/exometer_report_influxdb.erl b/src/exometer_report_influxdb.erl index e8052a5..b9840e7 100644 --- a/src/exometer_report_influxdb.erl +++ b/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 = <>, + 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);