Add stream filtering feature flag

acogoluegnes · acogoluegnes · commit 7bc26154b5d2 · 2023-07-10T15:21:53.000+02:00
diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl
@@ -113,3 +113,10 @@
       stability     => stable,
       depends_on    => [stream_single_active_consumer]
      }}).
+
+-rabbit_feature_flag(
+   {stream_filtering,
+    #{desc          => "Support for stream filtering.",
+      stability     => stable,
+      depends_on    => [stream_queue]
+     }}).
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -95,7 +95,8 @@
          outstanding_requests :: #{integer() => #request{}},
          deliver_version :: rabbit_stream_core:command_version(),
          request_timeout :: pos_integer(),
-         outstanding_requests_timer :: undefined | erlang:reference()}).
+         outstanding_requests_timer :: undefined | erlang:reference(),
+         filtering_supported :: boolean()}).
 -record(configuration,
         {initial_credits :: integer(),
          credits_required_for_unblocking :: integer(),
@@ -257,7 +258,8 @@ init([KeepaliveSup,
                                    correlation_id_sequence = 0,
                                    outstanding_requests = #{},
                                    request_timeout = RequestTimeout,
-                                   deliver_version = DeliverVersion},
+                                   deliver_version = DeliverVersion,
+                                   filtering_supported = rabbit_stream_utils:filtering_supported()},
             State =
                 #stream_connection_state{consumers = #{},
                                          blocked = false,
@@ -269,6 +271,13 @@ init([KeepaliveSup,
                 application:get_env(rabbitmq_stream,
                                     connection_negotiation_step_timeout,
                                     10_000),
+                Config = #configuration{
+                            initial_credits = InitialCredits,
+                            credits_required_for_unblocking = CreditsRequiredBeforeUnblocking,
+                            frame_max = FrameMax,
+                            heartbeat = Heartbeat,
+                            connection_negotiation_step_timeout = ConnectionNegotiationStepTimeout
+                           },
             % gen_statem process has its start_link call not return until the init function returns.
             % This is problematic, because we won't be able to call ranch:handshake/2
             % from the init callback as this would cause a deadlock to happen.
@@ -280,20 +289,7 @@ init([KeepaliveSup,
                                   #statem_data{transport = Transport,
                                                connection = Connection,
                                                connection_state = State,
-                                               config =
-                                                   #configuration{initial_credits
-                                                                      =
-                                                                      InitialCredits,
-                                                                  credits_required_for_unblocking
-                                                                      =
-                                                                      CreditsRequiredBeforeUnblocking,
-                                                                  frame_max =
-                                                                      FrameMax,
-                                                                  heartbeat =
-                                                                      Heartbeat,
-                                                                  connection_negotiation_step_timeout
-                                                                      =
-                                                                      ConnectionNegotiationStepTimeout}});
+                                               config = Config});
         {Error, Reason} ->
             rabbit_net:fast_close(RealSocket),
             rabbit_log_connection:warning("Closing connection because of ~tp ~tp",
@@ -1753,6 +1749,31 @@ handle_frame_post_auth(Transport,
                        {publish, PublisherId, MessageCount, Messages}) ->
     handle_frame_post_auth(Transport, Connection, State,
                            {publish, ?VERSION_1, PublisherId, MessageCount, Messages});
+handle_frame_post_auth(Transport,
+                       #stream_connection{filtering_supported = false,
+                                          publishers = Publishers,
+                                          socket = S} = Connection,
+                       State,
+                       {publish_v2, PublisherId, MessageCount, Messages}) ->
+    case Publishers of
+        #{PublisherId := #publisher{message_counters = Counters}} ->
+            increase_messages_received(Counters, MessageCount),
+            increase_messages_errored(Counters, MessageCount),
+            ok;
+        _ ->
+            ok
+    end,
+    rabbit_global_counters:increase_protocol_counter(stream,
+                                                     ?PRECONDITION_FAILED,
+                                                     1),
+    PublishingIds = publishing_ids_from_messages(?VERSION_2, Messages),
+    Command = {publish_error,
+               PublisherId,
+               ?RESPONSE_CODE_PRECONDITION_FAILED,
+               PublishingIds},
+    Frame = rabbit_stream_core:frame(Command),
+    send(Transport, S, Frame),
+    {Connection, State};
 handle_frame_post_auth(Transport,
                        Connection,
                        State,
@@ -1904,15 +1925,42 @@ handle_frame_post_auth(Transport,
             {Connection0, State}
     end;
 handle_frame_post_auth(Transport,
-                       #stream_connection{name = ConnName,
-                                          socket = Socket,
-                                          stream_subscriptions =
-                                              StreamSubscriptions,
-                                          virtual_host = VirtualHost,
-                                          user = User,
-                                          send_file_oct = SendFileOct,
-                                          transport = ConnTransport} =
-                           Connection,
+                       #stream_connection{filtering_supported = false} = Connection,
+                       State,
+                       {request, CorrelationId,
+                        {subscribe,
+                         SubscriptionId, _, _, _, Properties}} = Request) ->
+    case rabbit_stream_utils:filter_defined(Properties) of
+        true ->
+            rabbit_log:warning("Cannot create subcription ~tp, it defines a filter "
+                               "and filtering is not active",
+                               [SubscriptionId]),
+            response(Transport,
+                     Connection,
+                     subscribe,
+                     CorrelationId,
+                     ?RESPONSE_CODE_PRECONDITION_FAILED),
+            rabbit_global_counters:increase_protocol_counter(stream,
+                                                             ?PRECONDITION_FAILED,
+                                                             1),
+            {Connection, State};
+        false ->
+            handle_frame_post_auth(Transport, {ok, Connection}, State, Request)
+    end;
+handle_frame_post_auth(Transport, #stream_connection{} = Connection, State,
+                       {request, _,
+                        {subscribe,
+                         _, _, _, _, _}} = Request) ->
+    handle_frame_post_auth(Transport, {ok, Connection}, State, Request);
+handle_frame_post_auth(Transport,
+                       {ok, #stream_connection{
+                               name = ConnName,
+                               socket = Socket,
+                               stream_subscriptions = StreamSubscriptions,
+                               virtual_host = VirtualHost,
+                               user = User,
+                               send_file_oct = SendFileOct,
+                               transport = ConnTransport} = Connection},
                        #stream_connection_state{consumers = Consumers} = State,
                        {request, CorrelationId,
                         {subscribe,
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_utils.erl b/deps/rabbitmq_stream/src/rabbit_stream_utils.erl
@@ -29,8 +29,10 @@
          sort_partitions/1,
          strip_cr_lf/1,
          consumer_activity_status/2,
+         filter_defined/1,
          filter_spec/1,
-         command_versions/0]).
+         command_versions/0,
+         filtering_supported/0]).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 
@@ -302,6 +304,15 @@ consumer_activity_status(Active, Properties) ->
             waiting
     end.
 
+filter_defined(SubscriptionProperties) when is_map(SubscriptionProperties) ->
+    lists:any(fun(<<"filter.",_/binary>>) ->
+                      true;
+                 (_) ->
+                      false
+              end, maps:keys(SubscriptionProperties));
+filter_defined(_) ->
+    false.
+
 filter_spec(Properties) ->
     Filters = maps:fold(fun(<<"filter.",_/binary>>, V, Acc) ->
                                 [V] ++ Acc;
@@ -323,8 +334,14 @@ filter_spec(Properties) ->
     end.
 
 command_versions() ->
+    PublishMaxVersion = case filtering_supported() of
+                            false ->
+                                ?VERSION_1;
+                            true ->
+                                ?VERSION_2
+                        end,
     [{declare_publisher, ?VERSION_1, ?VERSION_1},
-     {publish, ?VERSION_1, ?VERSION_2},
+     {publish, ?VERSION_1, PublishMaxVersion},
      {query_publisher_sequence, ?VERSION_1, ?VERSION_1},
      {delete_publisher, ?VERSION_1, ?VERSION_1},
      {subscribe, ?VERSION_1, ?VERSION_1},
@@ -340,3 +357,6 @@ command_versions() ->
      {route, ?VERSION_1, ?VERSION_1},
      {partitions, ?VERSION_1, ?VERSION_1},
      {stream_stats, ?VERSION_1, ?VERSION_1}].
+
+filtering_supported() ->
+    rabbit_feature_flags:is_enabled(stream_filtering).
diff --git a/deps/rabbitmq_stream/test/commands_SUITE.erl b/deps/rabbitmq_stream/test/commands_SUITE.erl
@@ -698,6 +698,7 @@ subscribe(S, SubId, Stream, SubProperties, C) ->
                                        SubId,
                                        Stream,
                                        SubProperties,
+                                       ?RESPONSE_CODE_OK,
                                        C).
 
 subscribe(S, SubId, Stream, C) ->
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
@@ -33,7 +33,8 @@ all() ->
 
 groups() ->
     [{single_node, [],
-      [test_stream,
+      [filtering_ff, %% must stay at the top, feature flag disabled for this one
+       test_stream,
        test_stream_tls,
        test_publish_v2,
        test_gc_consumers,
@@ -72,6 +73,22 @@ init_per_group(Group, Config)
                          {rabbitmq_ct_tls_verify, verify_none},
                          {rabbitmq_stream, verify_none}
                         ]),
+    %% filtering feature flag disabled for the first test,
+    %% then enabled in the end_per_testcase function
+    ExtraSetupSteps =
+        case Group of
+            single_node ->
+                [fun(StepConfig) ->
+                    rabbit_ct_helpers:merge_app_env(StepConfig,
+                                                    {rabbit,
+                                                     [{forced_feature_flags_on_init,
+                                                       [stream_queue,
+                                                        stream_sac_coordinator_unblock_group,
+                                                        stream_single_active_consumer]}]})
+                 end];
+            _ ->
+                []
+        end,
     rabbit_ct_helpers:run_setup_steps(
       Config1,
       [fun(StepConfig) ->
@@ -86,6 +103,7 @@ init_per_group(Group, Config)
                                                 [{connection_negotiation_step_timeout,
                                                   500}]})
        end]
+      ++ ExtraSetupSteps
       ++ rabbit_ct_broker_helpers:setup_steps());
 init_per_group(cluster = Group, Config) ->
     Config1 = rabbit_ct_helpers:set_config(
@@ -123,6 +141,13 @@ init_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config
 init_per_testcase(TestCase, Config) ->
     rabbit_ct_helpers:testcase_started(Config, TestCase).
 
+end_per_testcase(filtering_ff = TestCase, Config) ->
+    _ = rabbit_ct_broker_helpers:rpc(Config,
+                                     0,
+                                     rabbit_feature_flags,
+                                     enable,
+                                     [stream_filtering]),
+    rabbit_ct_helpers:testcase_finished(Config, TestCase);
 end_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config,
                                       0,
@@ -133,6 +158,32 @@ end_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config)
 end_per_testcase(TestCase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config, TestCase).
 
+filtering_ff(Config) ->
+    Stream = atom_to_binary(?FUNCTION_NAME, utf8),
+    Transport = gen_tcp,
+    Port = get_stream_port(Config),
+    Opts = [{active, false}, {mode, binary}],
+    {ok, S} = Transport:connect("localhost", Port, Opts),
+    C0 = rabbit_stream_core:init(0),
+    C1 = test_peer_properties(Transport, S, C0),
+    C2 = test_authenticate(Transport, S, C1),
+    C3 = test_create_stream(Transport, S, Stream, C2),
+    PublisherId = 42,
+    C4 = test_declare_publisher(Transport, S, PublisherId, Stream, C3),
+    Body = <<"hello">>,
+    C5 = test_publish_confirm(Transport, S, publish_v2, PublisherId, Body,
+                               publish_error, C4),
+    SubscriptionId = 42,
+    C6 = test_subscribe(Transport, S, SubscriptionId, Stream,
+                        #{<<"filter.0">> => <<"foo">>},
+                        ?RESPONSE_CODE_PRECONDITION_FAILED,
+                        C5),
+
+    C7 = test_delete_stream(Transport, S, Stream, C6),
+    _C8 = test_close(Transport, S, C7),
+    closed = wait_for_socket_close(Transport, S, 10),
+    ok.
+
 test_global_counters(Config) ->
     Stream = atom_to_binary(?FUNCTION_NAME, utf8),
     test_server(gen_tcp, Stream, Config),
@@ -188,11 +239,14 @@ test_publish_v2(Config) ->
     PublisherId = 42,
     C4 = test_declare_publisher(Transport, S, PublisherId, Stream, C3),
     Body = <<"hello">>,
-    C5 = test_publish_confirm(Transport, S, publish_v2, PublisherId, Body, C4),
-    C6 = test_publish_confirm(Transport, S, publish_v2, PublisherId, Body, C5),
+    C5 = test_publish_confirm(Transport, S, publish_v2, PublisherId, Body,
+                              publish_confirm, C4),
+    C6 = test_publish_confirm(Transport, S, publish_v2, PublisherId, Body,
+                              publish_confirm, C5),
     SubscriptionId = 42,
     C7 = test_subscribe(Transport, S, SubscriptionId, Stream,
                         #{<<"filter.0">> => <<"foo">>},
+                        ?RESPONSE_CODE_OK,
                         C6),
     C8 = test_deliver(Transport, S, SubscriptionId, 0, Body, C7),
     C8b = test_deliver(Transport, S, SubscriptionId, 1, Body, C8),
@@ -419,6 +473,7 @@ close_connection_on_consumer_update_timeout(Config) ->
     C4 = test_subscribe(Transport, S, SubId, Stream,
                         #{<<"single-active-consumer">> => <<"true">>,
                           <<"name">> => <<"foo">>},
+                        ?RESPONSE_CODE_OK,
                         C3),
     {Cmd, _C5} = receive_commands(Transport, S, C4),
     ?assertMatch({request, _, {consumer_update, SubId, true}}, Cmd),
@@ -681,18 +736,21 @@ test_declare_publisher(Transport, S, PublisherId, Stream, C0) ->
     C.
 
 test_publish_confirm(Transport, S, PublisherId, Body, C0) ->
-    test_publish_confirm(Transport, S, publish, PublisherId, Body, C0).
+    test_publish_confirm(Transport, S, publish, PublisherId, Body,
+                         publish_confirm, C0).
 
-test_publish_confirm(Transport, S, publish = PublishCmd, PublisherId, Body, C0) ->
+test_publish_confirm(Transport, S, publish = PublishCmd, PublisherId, Body,
+                     ExpectedConfirmCommand,C0) ->
     BodySize = byte_size(Body),
     Messages = [<<1:64, 0:1, BodySize:31, Body:BodySize/binary>>],
     PublishFrame =
         rabbit_stream_core:frame({PublishCmd, PublisherId, 1, Messages}),
     ok = Transport:send(S, PublishFrame),
     {Cmd, C} = receive_commands(Transport, S, C0),
-    ?assertMatch({publish_confirm, PublisherId, [1]}, Cmd),
+    ?assertMatch({ExpectedConfirmCommand, PublisherId, [1]}, Cmd),
     C;
-test_publish_confirm(Transport, S, publish_v2 = PublishCmd, PublisherId, Body, C0) ->
+test_publish_confirm(Transport, S, publish_v2 = PublishCmd, PublisherId, Body,
+                     ExpectedConfirmCommand, C0) ->
     BodySize = byte_size(Body),
     FilterValue = <<"foo">>,
     FilterValueSize = byte_size(FilterValue),
@@ -702,7 +760,12 @@ test_publish_confirm(Transport, S, publish_v2 = PublishCmd, PublisherId, Body, C
         rabbit_stream_core:frame({PublishCmd, PublisherId, 1, Messages}),
     ok = Transport:send(S, PublishFrame),
     {Cmd, C} = receive_commands(Transport, S, C0),
-    ?assertMatch({publish_confirm, PublisherId, [1]}, Cmd),
+    case ExpectedConfirmCommand of
+        publish_confirm ->
+            ?assertMatch({ExpectedConfirmCommand, PublisherId, [1]}, Cmd);
+        publish_error ->
+            ?assertMatch({ExpectedConfirmCommand, PublisherId, _, [1]}, Cmd)
+    end,
     C.
 
 test_subscribe(Transport, S, SubscriptionId, Stream, C0) ->
@@ -711,21 +774,23 @@ test_subscribe(Transport, S, SubscriptionId, Stream, C0) ->
                    SubscriptionId,
                    Stream,
                    #{<<"random">> => <<"thing">>},
+                   ?RESPONSE_CODE_OK,
                    C0).
 
 test_subscribe(Transport,
                S,
                SubscriptionId,
                Stream,
                SubscriptionProperties,
+               ExpectedResponseCode,
                C0) ->
     SubCmd =
         {request, 1,
          {subscribe, SubscriptionId, Stream, 0, 10, SubscriptionProperties}},
     SubscribeFrame = rabbit_stream_core:frame(SubCmd),
     ok = Transport:send(S, SubscribeFrame),
     {Cmd, C} = receive_commands(Transport, S, C0),
-    ?assertMatch({response, 1, {subscribe, ?RESPONSE_CODE_OK}}, Cmd),
+    ?assertMatch({response, 1, {subscribe, ExpectedResponseCode}}, Cmd),
     C.
 
 test_unsubscribe(Transport, Socket, SubscriptionId, C0) ->
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_utils_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_utils_SUITE.erl
