Create question pipeline filter

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## main
 
+- Add `erldns_questions` questions filter to the packet pipeline.
 - Update dns_erlang v4.2 and remove `erldns_records:name_type/1`.
 
 ## v7.0.0

erldns.example.config

@@ -4,6 +4,7 @@
             #{name => inet_1, port => 8053}
         ]},
         {packet_pipeline, [
+            erldns_questions,
             erldns_query_throttle,
             erldns_packet_cache,
             erldns_resolver,

src/listeners/erldns_proto_tcp.erl

@@ -104,7 +104,8 @@ handle_decoded(_, _, _, #dns_message{qr = true}, _) ->
     {error, not_a_question};
 handle_decoded(Socket, TimerPid, TS0, DecodedMessage, IpAddr) ->
     forward_dp_to_timer(DecodedMessage, TimerPid),
-    Response = erldns_pipeline:call(DecodedMessage, #{transport => tcp, host => IpAddr}),
+    InitOpts = #{monotonic_time => TS0, transport => tcp, host => IpAddr},
+    Response = erldns_pipeline:call(DecodedMessage, InitOpts),
     EncodedResponse = erldns_encoder:encode_message(Response),
     exit(TimerPid, kill),
     ok = gen_tcp:send(Socket, [<<(byte_size(EncodedResponse)):16>>, EncodedResponse]),

src/listeners/erldns_proto_udp.erl

@@ -102,7 +102,8 @@ handle_decoded(_, _, _, #dns_message{qr = true}, _) ->
     {error, not_a_question};
 handle_decoded(Socket, IpAddr, Port, DecodedMessage0, TS0) ->
     DecodedMessage = normalize_edns_max_payload_size(DecodedMessage0),
-    Response = erldns_pipeline:call(DecodedMessage, #{transport => udp, host => IpAddr}),
+    InitOpts = #{monotonic_time => TS0, transport => udp, host => IpAddr},
+    Response = erldns_pipeline:call(DecodedMessage, InitOpts),
     Result = erldns_encoder:encode_message(Response, #{}),
     EncodedResponse =
         case Result of

src/pipes/erldns_pipeline.erl

@@ -15,6 +15,7 @@ be injected as a new pipe handler in the right order.
 
 The following are enabled by default, see their documentation for details:
 
+- `m:erldns_questions`
 - `m:erldns_query_throttle`
 - `m:erldns_packet_cache`
 - `m:erldns_resolver`
@@ -49,6 +50,7 @@ The API expected by a module pipe is defined as a behaviour by this module.
 ```erlang
 {erldns, [
     {packet_pipeline, [
+        erldns_questions,
         erldns_query_throttle,
         erldns_packet_cache,
         erldns_resolver,
@@ -93,6 +95,9 @@ call(Msg, _Opts) ->
 -type transport() :: tcp | udp.
 -doc "Options that can be passed and accumulated to the pipeline.".
 -type opts() :: #{
+    query_labels := dns:labels(),
+    query_type := dns:type(),
+    monotonic_time := integer(),
     resolved := boolean(),
     transport := transport(),
     host := host(),
@@ -139,6 +144,7 @@ This callback can return
 -endif.
 
 -define(DEFAULT_PACKET_PIPELINE, [
+    erldns_questions,
     erldns_query_throttle,
     erldns_packet_cache,
     erldns_resolver,
@@ -265,4 +271,11 @@ prepare_pipe(Fun, _) when is_function(Fun) ->
     erlang:error({badpipe, {function_pipe_has_wrong_arity, Fun}}).
 
 def_opts() ->
-    #{resolved => false, transport => udp, host => undefined}.
+    #{
+        query_labels => [],
+        query_type => ?DNS_TYPE_A,
+        monotonic_time => 0,
+        resolved => false,
+        transport => udp,
+        host => undefined
+    }.

src/pipes/erldns_questions.erl

@@ -0,0 +1,32 @@
+-module(erldns_questions).
+-moduledoc """
+Remove all redundant questions from a DNS message,
+and parses the first question into a list of labels.
+
+## Telemetry events
+
+- `[erldns, pipeline, questions]` with `#{count => non_neg_integer()}`
+where `count` is the number of questions removed.
+""".
+
+-include_lib("dns_erlang/include/dns.hrl").
+
+-behaviour(erldns_pipeline).
+
+-export([call/2]).
+
+-doc "`c:erldns_pipeline:call/2` callback.".
+-spec call(dns:message(), erldns_pipeline:opts()) -> erldns_pipeline:return().
+call(#dns_message{qc = 0} = Msg, _) ->
+    {stop, Msg#dns_message{qr = true}};
+call(#dns_message{qc = 1, questions = [#dns_query{name = Name, type = Type}]} = Msg, Opts) ->
+    Labels = dns:dname_to_lower_labels(Name),
+    {Msg, Opts#{query_labels := Labels, query_type := Type}};
+call(#dns_message{questions = [Q1 | Rest]} = Msg, #{host := Host} = Opts) ->
+    Labels = dns:dname_to_lower_labels(Q1#dns_query.name),
+    Measurements = #{count => length(Rest)},
+    Metadata = #{host => Host, questions => [Q1 | Rest]},
+    telemetry:execute([erldns, pipeline, questions], Measurements, Metadata),
+    Msg1 = Msg#dns_message{qc = 1, questions = [Q1]},
+    Opts1 = Opts#{query_labels := Labels, query_type := Q1#dns_query.type},
+    {Msg1, Opts1}.

test/questions_SUITE.erl

@@ -0,0 +1,69 @@
+-module(questions_SUITE).
+-compile([export_all, nowarn_export_all]).
+
+-behaviour(ct_suite).
+
+-include_lib("stdlib/include/assert.hrl").
+-include_lib("dns_erlang/include/dns.hrl").
+
+-spec all() -> [ct_suite:ct_test_def()].
+all() ->
+    [{group, all}].
+
+-spec groups() -> [ct_suite:ct_group_def()].
+groups() ->
+    [{all, [parallel], [empty, one_question, many_questions]}].
+
+-spec init_per_suite(ct_suite:ct_config()) -> ct_suite:ct_config().
+init_per_suite(Config) ->
+    application:ensure_all_started([telemetry]),
+    Events = [
+        [erldns, pipeline, questions]
+    ],
+    ok = telemetry:attach_many(?MODULE, Events, fun ?MODULE:telemetry_handler/4, []),
+    Config.
+
+-spec end_per_suite(ct_suite:ct_config()) -> term().
+end_per_suite(_) ->
+    application:stop(telemetry).
+
+%% Tests
+empty(_) ->
+    Msg = #dns_message{},
+    ?assertMatch({stop, #dns_message{}}, erldns_questions:call(Msg, def_opts())),
+    assert_no_telemetry_event().
+
+one_question(_) ->
+    Q = #dns_query{name = ~"example.com", type = ?DNS_TYPE_ANY},
+    Msg = #dns_message{qc = 1, questions = [Q]},
+    ?assertMatch({Msg, _}, erldns_questions:call(Msg, def_opts())),
+    assert_no_telemetry_event().
+
+many_questions(_) ->
+    Q = #dns_query{name = ~"example.com", type = ?DNS_TYPE_ANY},
+    Msg = #dns_message{qc = 2, questions = [Q, Q]},
+    ?assertMatch({#dns_message{qc = 1}, _}, erldns_questions:call(Msg, def_opts())),
+    assert_telemetry_event().
+
+def_opts() ->
+    erldns_pipeline:def_opts().
+
+telemetry_handler(EventName, _, _, _) ->
+    ct:pal("EventName ~p~n", [EventName]),
+    self() ! EventName.
+
+assert_telemetry_event() ->
+    receive
+        [erldns, pipeline, questions] ->
+            ok
+    after 1000 ->
+        ct:fail("Telemetry event not triggered: questions")
+    end.
+
+assert_no_telemetry_event() ->
+    receive
+        [erldns, pipeline, questions] ->
+            ct:fail("Telemetry event not triggered: questions")
+    after 100 ->
+        ok
+    end.