exometer_influxdb

-module(exometer_report_influxdb).

-behaviour(exometer_report).

%% gen_server callbacks
-export([exometer_init/1,
         exometer_info/2,
         exometer_cast/2,
         exometer_call/3,
         exometer_report/5,
         exometer_subscribe/5,
         exometer_unsubscribe/4,
         exometer_newentry/2,
         exometer_setopts/4,
         exometer_terminate/2]).


-ifdef(TEST).
-export([evaluate_subscription_options/5,
         make_packet/5]).
-endif.

-include_lib("exometer_core/include/exometer.hrl").

-define(DEFAULT_HOST, <<"127.0.0.1">>).
-define(DEFAULT_DB, <<"exometer">>).
-define(DEFAULT_PROTOCOL, http).
-define(DEFAULT_PORT, 8086).
-define(DEFAULT_USERNAME, undefined).
-define(DEFAULT_PASSWORD, undefined).
-define(DEFAULT_PRECISION, u).
-define(DEFAULT_SERIES_NAME, undefined).
-define(DEFAULT_FORMATTING, []).
-define(DEFAULT_TIMESTAMP_OPT, false).
-define(DEFAULT_BATCH_WINDOW_SIZE, 0).
-define(DEFAULT_AUTOSUBSCRIBE, false).
-define(DEFAULT_SUBSCRIPTIONS_MOD, undefined).

-define(VALID_PRECISIONS, [n, u, ms, s, m, h]).

-define(HTTP(Proto), (Proto =:= http orelse Proto =:= https)).

-include("log.hrl").

-type options() :: [{atom(), any()}].
-type value() :: any().
-type callback_result() :: {ok, state()} | any().
-type precision() :: n | u | ms | s | m | h.
-type protocol() :: http | udp.

-record(state, {protocol :: protocol(),
                db :: binary(), % for http
                username :: undefined | binary(), % for http
                password :: undefined | binary(), % for http
                host :: inet:ip_address() | inet:hostname(), % for udp
                port :: inet:port_number(),  % for udp
                timestamping :: boolean(),
                precision :: precision(),
                collected_metrics = #{} :: map(),
                batch_window_size = 0 :: integer(),
                tags :: map(),
                series_name :: atom() | binary(),
                formatting :: list(),
                metrics :: map(),
                autosubscribe :: boolean(),
                subscriptions_module :: module(),
                connection :: gen_udp:socket() | reference()}).
-type state() :: #state{}.


%% ===================================================================
%% Public API
%% ===================================================================
-spec exometer_init(options()) -> callback_result().
exometer_init(Opts) ->
    Host = get_opt(host, Opts, ?DEFAULT_HOST),
    Protocol = get_opt(protocol, Opts, ?DEFAULT_PROTOCOL),
    Port = get_opt(port, Opts, ?DEFAULT_PORT),
    DB = get_opt(db, Opts, ?DEFAULT_DB),
    Username = get_opt(username, Opts, ?DEFAULT_USERNAME),
    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),
    {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),
    Formatting = get_opt(formatting, Opts, ?DEFAULT_FORMATTING),
    Autosubscribe = get_opt(autosubscribe, Opts, ?DEFAULT_AUTOSUBSCRIBE),
    SubscriptionsMod = get_opt(subscriptions_module, Opts, ?DEFAULT_SUBSCRIPTIONS_MOD),
    MergedTags = merge_tags([{<<"host">>, net_adm:localhost()}], Tags),
    State =  #state{protocol = Protocol,
                    db = DB,
                    username = Username,
                    password = Password,
                    host = binary_to_list(Host),
                    port = Port,
                    timestamping = Timestamping,
                    precision = Precision,
                    tags = MergedTags,
                    series_name = SeriesName,
                    formatting = Formatting,
                    batch_window_size = BatchWinSize,
                    autosubscribe = Autosubscribe,
                    subscriptions_module = SubscriptionsMod,
                    metrics = maps:new()},
    code:load_file(hackney_tcp),
    code:load_file(hackney_ssl),
    case connect(Protocol, Host, Port, Username, Password) of
        {ok, Connection} ->
            ?info("InfluxDB reporter connecting success: ~p", [Opts]),
            {ok, State#state{connection = Connection}};
        Error ->
            ?error("InfluxDB reporter connecting error: ~p", [Error]),
            prepare_reconnect(),
            {ok, State}
    end.

-spec exometer_report(exometer_report:metric(),
                      exometer_report:datapoint(),
                      exometer_report:extra(),
                      value(),
                      state()) -> callback_result().
exometer_report(_Metric, _DataPoint, _Extra, _Value,
                #state{connection = undefined} = State) ->
    ?info("InfluxDB reporter isn't connected and will reconnect."),
    {ok, State};
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,
                       maps:from_list([{DataPoint, Value}]), State);
        Error ->
            ?warning("InfluxDB reporter got trouble when looking ~p metric's tag: ~p",
                     [Metric, Error]),
            Error
    end.

-spec exometer_subscribe(exometer_report:metric(),
                         exometer_report:datapoint(),
                         exometer_report:interval(),
                         exometer_report:extra(),
                         state()) -> callback_result().
exometer_subscribe(Metric, _DataPoint, _Interval, SubscribeOpts,
                   #state{metrics=Metrics, tags=DefaultTags,
                          series_name=DefaultSeriesName,
                          formatting=DefaultFormatting} = State) ->
    {MetricName, Tags} = evaluate_subscription_options(Metric, SubscribeOpts, DefaultTags,
                                                       DefaultSeriesName, DefaultFormatting),
    case MetricName of
        [] -> exit({invalid_metric_name, MetricName});
        _  ->
            NewMetrics = maps:put(Metric, {MetricName, Tags}, Metrics),
            {ok, State#state{metrics = NewMetrics}}
    end.

-spec exometer_unsubscribe(exometer_report:metric(),
                           exometer_report:datapoint(),
                           exometer_report:extra(),
                           state()) -> callback_result().
exometer_unsubscribe(Metric, _DataPoint, _Extra,
                     #state{metrics = Metrics} = State) ->
    {ok, State#state{metrics = maps:remove(Metric, Metrics)}}.

-spec exometer_call(any(), pid(), state()) ->
    {reply, any(), state()} | {noreply, state()} | any().
exometer_call(_Unknown, _From, State) ->
    {ok, State}.

-spec exometer_cast(any(), state()) -> {noreply, state()} | any().
exometer_cast(_Unknown, State) ->
    {ok, State}.

-spec exometer_info(any(), state()) -> callback_result().
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 = #{}});
    true -> {ok, State}
    end;
exometer_info(_Unknown, State) ->
    {ok, State}.

-spec exometer_newentry(exometer:entry(), state()) -> callback_result().
exometer_newentry(#exometer_entry{name = Name, type = Type} = _Entry, 
                  #state{autosubscribe = Autosubscribe, 
                         subscriptions_module = Module} = State) ->
    case {Autosubscribe, Module} of
        {true, undefined} ->
            ?warning("InfluxDB reporter has activated autosubscribe option, "
                     "but subscriptions module is undefined.");
        {true, Module} when is_atom(Module) ->
            subscribe(Module:subscribe(Name, Type));
        _ -> []
    end,
    {ok, State}.

-spec exometer_setopts(exometer:entry(), options(),
                       exometer:status(), state()) -> callback_result().
exometer_setopts(_Metric, _Options, _Status, State) ->
    {ok, State}.

-spec exometer_terminate(any(), state()) -> any().
exometer_terminate(Reason, _) ->
    ?info("InfluxDB reporter is terminating with reason: ~p~n", [Reason]),
    ignore.


%% ===================================================================
%% Internal functions
%% ===================================================================
-spec connect(protocol(), string() | binary(), integer(),
              undefined | iodata(), undefined | iodata()) ->
    {ok, pid() | reference()} | {error, term()}.
connect(Proto, Host, Port, Username, Password) when ?HTTP(Proto) ->
    {ok, _} = application:ensure_all_started(hackney),
    Options = case {Username, Password} of
        {undefined, _} -> [];
        {_, undefined} -> [];
        _ -> [{basic_auth, {Username, Password}}]
    end ++ [{pool, false}],
    Transport = case Proto of
        http -> 
            case code:is_loaded(hackney_tcp) of
                false ->  hackney_tcp_transport;
                _ -> hackney_tcp
            end;
        https -> 
            case code:is_loaded(hackney_ssl) of
               false ->  hackney_ssl_transport;
               _ -> hackney_ssl
            end
    end,
    hackney:connect(Transport, Host, Port, Options);
connect(udp, _, _, _, _) -> gen_udp:open(0);
connect(Protocol, _, _, _, _) -> {error, {Protocol, not_supported}}.

-spec reconnect(state()) -> {ok, state()}.
reconnect(#state{protocol = Protocol, host = Host, port = Port,
                 username = Username, password = Password} = State) ->
    case connect(Protocol, Host, Port, Username, Password) of
        {ok, Connection} ->
            ?info("InfluxDB reporter reconnecting success: ~p",
                  [{Protocol, Host, Port, Username, Password}]),
            {ok, State#state{connection = Connection}};
        Error ->
            ?error("InfluxDB reporter reconnecting error: ~p", [Error]),
            prepare_reconnect(),
            {ok, State#state{connection = undefined}}
    end.

prepare_batch_send(Time) ->
    erlang:send_after(Time, self(), {exometer_influxdb, send}).

prepare_reconnect() ->
    erlang:send_after(1000, self(), {exometer_influxdb, reconnect}).

-spec maybe_send(list(), list(), map(), map(), state()) ->
    {ok, state()} | {error, term()}.
maybe_send(OriginMetricName, MetricName, Tags0, Fields, 
           #state{batch_window_size = BatchWinSize, 
                  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).

-spec send(binary() | list(), state()) ->
    {ok, state()} | {error, term()}.
send(Packet, #state{protocol = Proto, connection= Connection,
                    precision = Precision, db = DB,
                    timestamping = Timestamping} = State)
    when ?HTTP(Proto) ->
    QsVals = case Timestamping of
                 false -> [{<<"db">>, DB}];
                 true  -> [{<<"db">>, DB}, {<<"precision">>, Precision}]
             end,
    Url = hackney_url:make_url(<<"/">>, <<"write">>, QsVals),
    Req = {post, Url, [], Packet},
    case hackney:send_request(Connection, Req) of
        {ok, 204, _, Ref} ->
            hackney:body(Ref),
            {ok, State};
        {ok, Status, _Headers, Ref} ->
            {ok, Body} = hackney:body(Ref),
            ?warning("InfluxDB reporter got unexpected response with code ~p"
                     " and body: ~p. Reconnecting ...", [Status, Body]),
            reconnect(State);
        {error, _} = Error ->
            ?error("InfluxDB reporter HTTP sending error: ~p", [Error]),
            reconnect(State)
    end;
send(Packet, #state{protocol = udp, connection = Socket,
                    host = Host, port = Port} = State) ->
    case gen_udp:send(Socket, Host, Port, Packet) of
        ok -> {ok, State};
        Error ->
            ?error("InfluxDB reporter UDP sending error: ~p", [Error]),
            reconnect(State)
    end;
send(_, #state{protocol = Protocol}) -> {error, {Protocol, not_supported}}.

-spec merge_tags(list() | map(), list() | map()) -> map().
merge_tags(Tags, AdditionalTags) when is_list(Tags) ->
    merge_tags(maps:from_list(Tags), AdditionalTags);
merge_tags(Tags, AdditionalTags) when is_list(AdditionalTags) ->
    merge_tags(Tags, maps:from_list(AdditionalTags));
merge_tags(Tags, AdditionalTags) when not is_map(AdditionalTags) -> Tags;
merge_tags(Tags, AdditionalTags) -> maps:merge(Tags, AdditionalTags).

-spec subscribe(list() | {exometer_report:metric(),
                          exometer_report:datapoint(),
                          exometer_report:interval(),
                          exometer_report:extra()}) -> ok.
subscribe(Subscribtions) when is_list(Subscribtions) ->
    [subscribe(Subscribtion) || Subscribtion <- Subscribtions];
subscribe({Name, DataPoint, Interval, Extra, Retry}) when is_boolean(Retry) ->
    exometer_report:subscribe(?MODULE, Name, DataPoint, Interval, Extra, Retry);
subscribe({Name, DataPoint, Interval, Extra}) ->
    exometer_report:subscribe(?MODULE, Name, DataPoint, Interval, Extra);
subscribe(_Name) -> 
    [].

-spec get_opt(atom(), list(), any()) -> any().
get_opt(K, Opts, Default) ->
    exometer_util:get_opt(K, Opts, Default).

%% LINE PROTOCOL
-define(SEP(V), case V of <<>> -> <<>>; [] -> <<>>; _ -> <<$,>> end).

-spec microsecs() -> integer().
-ifdef(TEST).
microsecs() -> 1456993524527361.
-else.
microsecs() ->
    {MegaSecs, Secs, MicroSecs} = os:timestamp(),
    MegaSecs * 1000000 * 1000000 + Secs * 1000000 + MicroSecs.
-endif.

-spec convert_time_unit(integer(), erlang:time_unit() | minutes | hours) ->
    integer().
convert_time_unit(MicroSecs, hours) ->
    round(convert_time_unit(MicroSecs, minutes) / 60);
convert_time_unit(MicroSecs, minutes) ->
    round(convert_time_unit(MicroSecs, seconds) / 60);
convert_time_unit(MicroSecs, seconds) ->
    round(convert_time_unit(MicroSecs, milli_seconds) / 1000);
convert_time_unit(MicroSecs, milli_seconds) ->
    round(MicroSecs / 1000);
convert_time_unit(MicroSecs, nano_seconds) ->
    MicroSecs * 1000.

-spec unix_time(precision() | undefined) -> integer() | undefined.
unix_time(n)  -> convert_time_unit(microsecs(), nano_seconds);
unix_time(u)  -> microsecs();
unix_time(ms) -> convert_time_unit(microsecs(), milli_seconds);
unix_time(s)  -> convert_time_unit(microsecs(), seconds);
unix_time(m)  -> convert_time_unit(microsecs(), minutes);
unix_time(h)  -> convert_time_unit(microsecs(), hours);
unix_time(_)  -> undefined.

-spec metric_to_string(list()) -> string().
metric_to_string([Final]) -> metric_elem_to_list(Final);
metric_to_string([H | T]) ->
    metric_elem_to_list(H) ++ "_" ++ metric_to_string(T).

-spec metric_elem_to_list(atom() | string() | integer()) -> string().
metric_elem_to_list(E) when is_atom(E) -> atom_to_list(E);
metric_elem_to_list(E) when is_binary(E) -> binary_to_list(E);
metric_elem_to_list(E) when is_list(E) -> E;
metric_elem_to_list(E) when is_integer(E) -> integer_to_list(E).

-spec name(exometer_report:metric() | atom() | binary()) -> binary().
name(Metric) when is_atom(Metric) -> atom_to_binary(Metric, utf8);
name(Metric) when is_binary(Metric) -> Metric;
name(Metric) -> iolist_to_binary(metric_to_string(Metric)).

-spec key(integer() | atom() | list() | binary()) -> binary().
key(K) when is_integer(K) -> key(integer_to_binary(K));
key(K) when is_list(K) -> key(list_to_binary(K));
key(K) when is_atom(K) -> key(atom_to_binary(K, utf8));
key(K) ->
    binary:replace(K, [<<" ">>, <<$,>>, <<$=>>], <<$\\>>,
                   [global, {insert_replaced, 1}]).

-spec value(any()) -> binary() | list().
value(V) when is_integer(V) -> [integer_to_binary(V), $i];
value(V) when is_float(V) -> float_to_binary(V);
value(V) when is_atom(V) -> value(atom_to_binary(V, utf8));
value(V) when is_list(V) -> value(list_to_binary(V));
value(V) when is_binary(V) ->
    [$", binary:replace(V, <<$">>, <<$\\, $">>, [global]), $"].

-spec flatten_fields(map()) -> list().
flatten_fields(Fields) ->
    maps:fold(fun(K, V, Acc) ->
                  [Acc, ?SEP(Acc), key(K), $=, value(V)]
              end, <<>>, Fields).

-spec flatten_tags(map() | list()) -> list().
flatten_tags(Tags) when is_map(Tags) -> flatten_tags(maps:to_list(Tags));
flatten_tags(Tags) ->
    lists:foldl(fun({K, V}, Acc) ->
                    [Acc, ?SEP(Acc), key(K), $=, key(V)]
                end, [], lists:keysort(1, Tags)).

-spec make_packet(exometer_report:metric(), map() | list(),
                  map(), boolean() | non_neg_integer(), precision()) -> 
    list().
make_packet(Measurement, Tags, Fields, Timestamping, Precision) ->
    BinaryTags = flatten_tags(Tags),
    BinaryFields = flatten_fields(Fields),
    case Timestamping of
        false ->
            [name(Measurement), ?SEP(BinaryTags), BinaryTags, " ",
            BinaryFields, " "];
        true ->
            [name(Measurement), ?SEP(BinaryTags), BinaryTags, " ",
            BinaryFields, " ", integer_to_binary(unix_time(Precision))];
        Timestamp when is_integer(Timestamp) -> % for batch sending with timestamp
            [name(Measurement), ?SEP(BinaryTags), BinaryTags, " ",
            BinaryFields, " ", integer_to_binary(Timestamp)]
    end.

-spec evaluate_timestamp_opt({boolean(), precision()} | boolean())
                            -> {boolean(), precision()}.
evaluate_timestamp_opt({Term, Precision}) when is_boolean(Term) ->
    case lists:member(Precision, ?VALID_PRECISIONS) of
        true -> {Term, Precision};
        false -> exit(invalid_precision)
    end;
evaluate_timestamp_opt(Term) when is_boolean(Term) ->
    {Term, ?DEFAULT_PRECISION};
evaluate_timestamp_opt(_) ->
    exit(invalid_timestamp_option).

-spec del_indices(list(), [integer()]) -> list().
del_indices(List, Indices) ->
    SortedIndices = lists:reverse(lists:usort(Indices)),
    case length(SortedIndices) == length(Indices) of
        true -> del_indices1(List, SortedIndices);
        false -> exit({invalid_indices, Indices})
    end.

-spec del_indices1(list(), [integer()]) -> list().
del_indices1(List, []) -> List;
del_indices1([], Indices = [ _Index | _Indices1 ]) -> exit({too_many_indices, Indices});
del_indices1(List, [Index | Indices]) when length(List) >= Index, Index > 0 ->
    {L1, [_|L2]} = lists:split(Index-1, List),
    del_indices1(L1 ++ L2, Indices);
del_indices1(_List, Indices) ->
    exit({invalid_indices, Indices}).

-spec evaluate_subscription_options(list(), [{atom(), value()}], map(), atom(), [{atom(), value()}])
                                    -> {list() | atom(), map()}.
evaluate_subscription_options(MetricId, undefined, DefaultTags, DefaultSeriesName, DefaultFormatting) ->
  evaluate_subscription_options(MetricId, [], DefaultTags, DefaultSeriesName, DefaultFormatting);
evaluate_subscription_options(MetricId, Options, DefaultTags, DefaultSeriesName, DefaultFormatting) ->
    TagOpts = proplists:get_value(tags, Options, []),
    TagsResult = evaluate_subscription_tags(MetricId, TagOpts),
    FormattingOpts = proplists:get_value(formatting, Options, DefaultFormatting),
    FormattingResult = evaluate_subscription_formatting(TagsResult, FormattingOpts),
    SeriesName = proplists:get_value(series_name, Options, DefaultSeriesName),
    {FinalMetricId, NewTags} = evaluate_subscription_series_name(FormattingResult, SeriesName),
    TagMap = maps:from_list(NewTags),
    FinalTags = merge_tags(DefaultTags, TagMap),
    {FinalMetricId, FinalTags}.

-spec evaluate_subscription_tags(list(), [{atom(), value()}]) -> 
    {list(), [{atom(), value()}], [integer()]}.
evaluate_subscription_tags(MetricId, TagOpts) ->
    evaluate_subscription_tags(MetricId, TagOpts, [], []).

-spec evaluate_subscription_tags(list(), [{atom(), value()}], [{atom(), value()}], [integer()]) -> 
    {list(), [{atom(), value()}], [integer()]}.
evaluate_subscription_tags(MetricId, [], TagAcc, PosAcc) ->
    {MetricId, TagAcc, PosAcc};
evaluate_subscription_tags(MetricId, [{TagKey, {from_name, Pos}} | TagOpts], TagAcc, PosAcc)
    when is_number(Pos), length(MetricId) >= Pos, Pos > 0 ->
    NewTagAcc = TagAcc ++ [{TagKey, lists:nth(Pos, MetricId)}],
    NewPosAcc = PosAcc ++ [Pos],
    evaluate_subscription_tags(MetricId, TagOpts, NewTagAcc, NewPosAcc);
evaluate_subscription_tags(MetricId, [TagOpt = {TagKey, {from_name, Name}} | TagOpts], TagAcc, PosAcc) ->
    case string:str(MetricId, [Name]) of
        0     -> exit({invalid_tag_option, TagOpt});
        Index ->
            NewTagAcc = TagAcc ++ [{TagKey, Name}],
            NewPosAcc = PosAcc ++ [Index],
            evaluate_subscription_tags(MetricId, TagOpts, NewTagAcc, NewPosAcc)
    end;
evaluate_subscription_tags(MetricId, [Tag = {_Key, _Value} | Tags], TagAcc, PosAcc) ->
    evaluate_subscription_tags(MetricId, Tags, TagAcc ++ [Tag], PosAcc);
evaluate_subscription_tags(_MetricId, [Tag | _] , _TagAcc, _PosAcc) ->
    exit({invalid_tag_option, Tag}).

-spec evaluate_subscription_formatting({list(), [{atom(), value()}], [integer()]}, term())
                                       -> {list(), [{atom(), value()}]}.
evaluate_subscription_formatting({MetricId, Tags, FromNameIndices}, FormattingOpts) ->
    ToPurge = proplists:get_value(purge, FormattingOpts, []),
    KeysToPurge = proplists:get_all_values(tag_keys, ToPurge),
    ValuesToPurge = proplists:get_all_values(tag_values, ToPurge),
    PurgedTags = [{TagKey, TagValue} || {TagKey, TagValue} <- Tags,
                                        lists:member(TagKey, KeysToPurge) == false,
                                        lists:member(TagValue, ValuesToPurge) == false],
    FromNamePurge = proplists:get_value(all_from_name, ToPurge, true),
    PurgedMetricId = case FromNamePurge of
                          true  -> del_indices(MetricId, FromNameIndices);
                          false -> MetricId
                     end,
    {PurgedMetricId, PurgedTags}.

-spec evaluate_subscription_series_name({list(), [{atom(), value()}]}, atom())
                                        -> {list() | atom(), [{atom(), value()}]}.
evaluate_subscription_series_name({MetricId, Tags}, undefined) -> {MetricId, Tags};
evaluate_subscription_series_name({_MetricId, Tags}, SeriesName) -> {SeriesName, Tags}.