queue takeover via config

Author: heyoka

apps/faxe/priv/faxe.schema

@@ -508,14 +508,41 @@
     [{default, "q_"}, {datatype, string}]
 }.
 
+%% rabbitmq default prefix for every queue
+%% the amqp_consume node will use this to ensure this prefix is there for every queue
+%% this is mostly to make rabbitmq ha and other policies work automatically
+{mapping, "rabbitmq.queue_type", "faxe.rabbitmq.queue_type",
+    [{default, "classic"} , {datatype, string}]
+    %, {datatype, {enum,["classic", "quorum", ""]} } ]
+}.
+
 %% rabbitmq default prefix for every exchange
 %% the amqp_consume node will use this to ensure this prefix is there for every exchange
 %% this is mostly to make rabbitmq ha and other policies work automatically
 {mapping, "rabbitmq.exchange_prefix", "faxe.rabbitmq.exchange_prefix",
     [{default, "x_"}, {datatype, string}]
 }.
 
+%% @doc ------------------------------------------------------------------------------
+%% Queue takeover
+%% -----------------------------------------------------------------------------------
+%% rabbitmq default prefix for every takeover-queue
+%% the amqp_consume node will use this to ensure this prefix is there for every queue
+%% such a prefix is often used to select policies by (the start of) the queue name
+{mapping, "rabbitmq.takeover_queue_prefix", "faxe.rabbitmq.takeover_queue_prefix",
+    [{default, "q_"}, {datatype, string}]
+}.
+
+%% type of queue for the takeover queues
+{mapping, "rabbitmq.takeover_queue_type", "faxe.rabbitmq.takeover_queue_type",
+    [{default, "classic"}, {datatype, string}]
+    %, {datatype, {enum,["classic", "quorum", ""]} } ]
+}.
 
+%% enable queue takeover
+{mapping, "rabbitmq.takeover", "faxe.rabbitmq.takeover",
+    [{default, off}, {datatype, flag}, {commented, off}]
+}.
 %% @doc -------------------------------------------------------------------------------
 %% CrateDB defaults (postgreSQL connect)
 %% -------------------------------------------------------------------------------

apps/faxe/src/components/esp_amqp_consume.erl

@@ -93,12 +93,13 @@ options() -> [
    {bindings, string_list, undefined},
    {qx_name, string, undefined}, %% not used currently
    {queue, any, undefined},
-   {queue_type, string, <<>>},
-   {takeover, boolean, false},
+   {queue_type, string, {rabbitmq, queue_type}},
+   {takeover, boolean, {rabbitmq, takeover}},
    {takeover_timeout, duration, <<"5m">>},
    {takeover_queue, string, undefined},
-   {takeover_queue_prefix, string, {rabbitmq, queue_prefix}},
-   {takeover_queue_type, string, <<>>},
+   {takeover_queue_prefix, string, {rabbitmq, takeover_queue_prefix}},
+   {takeover_queue_type, string, {rabbitmq, takeover_queue_type}},
+   %% defaults to "vhost"
    {takeover_queue_vhost, string, undefined},
    {queue_prefix, string, {rabbitmq, queue_prefix}},
    {consumer_tag, string, undefined},
@@ -136,7 +137,7 @@ metrics() ->
    ].
 
 init({GraphId, NodeId} = Idx, _Ins,
-   #{ host := Host0, port := Port, user := _User, pass := _Pass, vhost := _VHost, queue := Q0, queue_type := QType0,
+   #{ host := Host0, port := Port, user := _User, pass := _Pass, vhost := VHost, queue := Q0, queue_type := QType0,
       exchange := Ex0, qx_name := _QxName, prefetch := Prefetch, routing_key := RoutingKey0, bindings := Bindings0,
       dt_field := DTField, dt_format := DTFormat, ssl := UseSSL, include_topic := IncludeTopic,
       topic_as := TopicKey, ack_every := AckEvery0, ack_after := AckTimeout0, as := As, consumer_tag := CTag0,
@@ -149,10 +150,10 @@ init({GraphId, NodeId} = Idx, _Ins,
       '_parent_pid' := ParentPid, '_parent_subscriptions' := ParentSubs, passive := Passive
    } = Opts0) ->
 
-%%   lager:warning("init ~p", [Idx]),
-
+%%   lager:warning("opts ~p", [Opts0]),
    Q = eval_name(Q0, Opts0, Idx),
-   QType = binary_to_list(QType0),
+   QName = faxe_util:prefix_binary(Q, QPrefix),
+   QType = faxe_util:to_list(QType0),
    CTag = case CTag0 of undefined -> <<"c_", GraphId/binary, "_", NodeId/binary>>; _ -> CTag0 end,
    State0 = #state{
       include_topic = IncludeTopic, topic_key = TopicKey, as = As, dedup_queue = memory_queue:new(DedupSize),
@@ -169,17 +170,27 @@ init({GraphId, NodeId} = Idx, _Ins,
       true ->
          TakeoverQ1 = eval_name(TakeoverQ0, Opts0, Idx),
          TakeoverQ2 = faxe_util:prefix_binary(TakeoverQ1, TakeoverQPrefix),
+
          TakeoverTime = faxe_time:duration_to_ms(TakeoverTimeout),
          NewTOpts = Opts0#{takeover_queue => TakeoverQ2, takeover_time => TakeoverTime},
          TakeoverOpts0 = init_takeover_consumer(self(), Idx, CTag, NewTOpts),
-         %% start takeover consumer
-         State01 = State0#state{takeover_consumer_opts = TakeoverOpts0},
-         TakeoverPid = start_takeover_consumer(State01),
-         TakeoverData = memory_queue:new(DedupSize),
-         {TakeoverOpts0, State01#state{takeover_consumer_pid = TakeoverPid, takeover_data = TakeoverData}}
+         CompQ = #{queue => QName, queue_type => QType, vhost => VHost},
+         case check_unique_q(TakeoverOpts0, CompQ) of
+            true ->
+               %% start takeover consumer
+               State01 = State0#state{takeover_consumer_opts = TakeoverOpts0},
+               TakeoverPid = start_takeover_consumer(State01),
+               TakeoverData = memory_queue:new(DedupSize),
+               {TakeoverOpts0, State01#state{takeover_consumer_pid = TakeoverPid, takeover_data = TakeoverData}};
+            false ->
+               lager:warning(
+                  "cannot start takeover action, because both queues are the same ~p, will continue without takeover",
+                  [CompQ]),
+               {undefined, State0}
+         end
    end,
 
-   TakeoverQType = binary_to_list(TakeoverQType0),
+   TakeoverQType = faxe_util:to_list(TakeoverQType0),
    TakeoverQ = case is_map(TakeoverOpts) andalso is_map_key(queue, TakeoverOpts) of
                   true -> maps:get(queue, TakeoverOpts);
                   false -> undefined
@@ -192,7 +203,7 @@ init({GraphId, NodeId} = Idx, _Ins,
 
    Host = binary_to_list(Host0),
 %%   lager:info("opts before: ~p",[Opts0]),
-   QName = faxe_util:prefix_binary(Q, QPrefix),
+
    Opts = Opts0#{
       host => Host, consumer_tag => CTag,
       exchange => faxe_util:prefix_binary(Ex, XPrefix),
@@ -201,7 +212,7 @@ init({GraphId, NodeId} = Idx, _Ins,
       routing_key => faxe_util:to_rkey(RoutingKey0),
       bindings => faxe_util:to_rkey(Bindings0)
    },
-%%   lager:info("opts: ~p",[Opts]),
+   lager:info("opts: ~p",[Opts]),
 
    State = State1#state{
       opts = Opts, ack_after = AckTimeout, queue_type = QType,
@@ -217,6 +228,12 @@ init({GraphId, NodeId} = Idx, _Ins,
 
    {ok, start_consumer(NewState)}.
 
+check_unique_q(#{queue := Q, queue_type := QType, vhost := VHost}, #{queue := Q, queue_type := QType, vhost := VHost}) ->
+   false;
+check_unique_q(_, _) ->
+   true.
+
+
 init_takeover_consumer(ParentPid, IdxParent, CTag,
     Opts = #{takeover_queue := Q0, takeover_queue_type := QType0, '_name' := Name, takeover_queue_vhost := TVHost,
        vhost := VHost}) ->
@@ -228,7 +245,7 @@ init_takeover_consumer(ParentPid, IdxParent, CTag,
       use_flow_ack => false,
       %% for the takeover-consumer, the takeover options become the "normal" q opts
       queue => Q0,
-      queue_type => QType0,
+      queue_type => faxe_util:to_list(QType0),
       vhost => TakeoverVHost,
       %% cannot use queue prefix
       queue_prefix => <<>>,

apps/faxe/src/faxe.erl

@@ -489,30 +489,6 @@ start_task(TaskId, Permanent) when Permanent == true orelse Permanent == false -
 start_task(TaskId, GraphRunMode, Permanent) ->
    start_task(TaskId, #task_modes{run_mode = GraphRunMode, permanent = Permanent}).
 
-%%do_start_task(T = #task{name = Name, definition = GraphDef},
-%%    #task_modes{concurrency = Concurrency, permanent = Perm} = Mode) ->
-%%   case dataflow:create_graph(Name, GraphDef) of
-%%      {ok, Graph} ->
-%%         try dataflow:start_graph(Graph, Mode) of
-%%            _ ->
-%%               faxe_db:save_task(T#task{pid = Graph, last_start = faxe_time:now_date(), permanent = Perm}),
-%%               Res =
-%%               case Concurrency of
-%%                  1 -> {ok, Graph};
-%%                  Num when Num > 1 ->
-%%                     start_concurrent(T, Mode),
-%%                     {ok, Graph}
-%%               end,
-%%%%               flow_changed({task, Name, start}),
-%%               Res
-%%         catch
-%%            _:_ = E ->
-%%               lager:error("graph_start_error: ~p",[E]),
-%%               {error, {graph_start_error, E}}
-%%         end;
-%%      {error, {already_started, _Pid}} -> {error, already_started}
-%%   end.
-
 start_concurrent(Task = #task{}, #task_modes{concurrency = Con} = Mode) ->
    F = fun(Num) -> start_copy(Task, Mode, Num) end,
    lists:map(F, lists:seq(2, Con)).
@@ -599,16 +575,11 @@ get_group_leader([T=#task{group_leader = true} | _R]) ->
 
 -spec stop_task(integer()|binary()|#task{}) -> ok.
 %% @doc just stop the graph process and its children
-stop_task(_T=#task{pid = Graph}) when is_pid(Graph) ->
-   case is_process_alive(Graph) of
-      true ->
-         df_graph:stop(Graph);
-      false ->
-         {error, not_running}
-   end;
-
+stop_task(T=#task{pid = _Graph}) ->
+   stop_task(T, false);
 stop_task(TaskId) ->
    stop_task(TaskId, false).
+
 -spec stop_task(#task{}|integer()|binary(), true|false) -> ok.
 stop_task(T = #task{}, Permanent) ->
    do_stop_task(T, Permanent);

apps/faxe/src/faxe_dfs.erl

@@ -218,15 +218,15 @@ eval_options([#{name := OptName, type := OptType, default := CKey}|Opts], Acc) w
    eval_options([{OptName, OptType, CKey}|Opts], Acc);
 eval_options([{OptName, OptType, CKey}|Opts], Acc) when is_tuple(CKey) ->
 %%   lager:info("call conf_val with: ~p for: ~p with Acc: ~p", [CKey, {OptName, OptType, CKey}, Acc]),
-   OptVal = conf_val(CKey),
+   OptVal = conf_val(CKey, OptType),
    eval_options(Opts, Acc ++ [{OptName, OptType, OptVal}]);
 eval_options([Opt|Opts], Acc) ->
    eval_options(Opts, Acc ++ [Opt]).
 
-conf_val({ConfigKey, ConfigSubKey}) ->
+conf_val({ConfigKey, ConfigSubKey}, OptionType) ->
    ConfigData = application:get_env(faxe, ConfigKey, []),
-   conv_config_val(proplists:get_value(ConfigSubKey, ConfigData));
-conf_val({ConfigKey, ConfigSubKey, ConfigSubSubKey}) ->
+   conv_config_val(proplists:get_value(ConfigSubKey, ConfigData), OptionType);
+conf_val({ConfigKey, ConfigSubKey, ConfigSubSubKey}, OptionType) ->
    Value =
    case application:get_env(faxe, ConfigKey, []) of
       [] -> [];
@@ -236,16 +236,15 @@ conf_val({ConfigKey, ConfigSubKey, ConfigSubSubKey}) ->
             ConfSubData when is_list(ConfSubData) -> proplists:get_value(ConfigSubSubKey, ConfSubData)
          end
    end,
-   conv_config_val(Value).
+   conv_config_val(Value, OptionType).
 
 %% @doc we know config does not give us any binary values, but in faxe we only use
-%% binaries for strings, so we convert any string(list) val to binary
-%% at the moment this is dangerous, as we do not know exactly whether the list was meant
-%% to be a string
-conv_config_val([]) -> [];
-conv_config_val(Val) when is_list(Val) ->
-   list_to_binary(Val);
-conv_config_val(Val) -> Val.
+%% binaries for strings, so we convert any binary/string option type to binary
+conv_config_val([], binary) -> <<>>;
+conv_config_val([], string) -> <<>>;
+conv_config_val(Val, binary) -> faxe_util:to_bin(Val);
+conv_config_val(Val, string) -> faxe_util:to_bin(Val);
+conv_config_val(Val, _Type) -> Val.
 
 node_id({Name, Id}) when is_binary(Name) andalso is_integer(Id) ->
    <<Name/binary, (integer_to_binary(Id))/binary>>.