Packages

Fault-tolerant dataflow DSL

Current section

Files

Jump to
kflow src framework kflow_sup.erl
Raw

src/framework/kflow_sup.erl

%%%===================================================================
%%% @copyright 2019 Klarna Bank AB (publ)
%%% @private
%%%===================================================================
-module(kflow_sup).
-behaviour(supervisor3).
%% API
-export([start_link/0]).
%% Supervisor3 callbacks
-export([init/1, post_init/1]).
-include("kflow_int.hrl").
-define(SERVER, ?MODULE).
-ifndef(TEST).
-define(COOLDOWN, 10).
-else.
-define(COOLDOWN, 1).
-endif.
%%%===================================================================
%%% API functions
%%%===================================================================
start_link() ->
supervisor3:start_link({local, ?SERVER}, ?MODULE, level1).
%%%===================================================================
%%% Supervisor callbacks
%%%===================================================================
init(level1) ->
SupFlags = { _SupStrategy = one_for_one
, _SupIntensity = 0
, _SupPeriod = 5
},
StartArgs = [{local, kflow_sup2}, ?MODULE, level2],
Level2 = { _Id = kflow_sup2
, _Start = {supervisor3, start_link, StartArgs}
, _Restart = {permanent, ?COOLDOWN}
, _Shutdown = infinity
, _Type = supervisor
, _Module = [kflow_sup]
},
{ok, {SupFlags, [Level2]}};
init(level2) ->
SupFlags = { _SupStrategy = one_for_all
, _SupIntensity = 1
, _SupPeriod = 5
},
BrodClients = brod_client_specs(),
PipeSup = { _Id = kflow_pipe_sup
, _Start = {kflow_pipe_sup, start_link, []}
, _Restart = permanent
, _Shutdown = infinity
, _Type = supervisor
, _Module = [kflow_pipe_sup]
},
{ok, {SupFlags, BrodClients ++ [PipeSup]}}.
post_init(_) ->
ignore.
%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec brod_client_specs() -> [supervisor3:child_spec()].
brod_client_specs() ->
Clients = ?cfg(kafka_clients),
[brod_client_spec(I) || I <- maps:keys(Clients)].
%% @private Create a child spec for a brod client. The idea is that
%% bringing brod clients under kflow supervision tree will make
%% restarts more aggressive and more prone to recover stuck consumers
-spec brod_client_spec(atom()) -> supervisor3:child_spec().
brod_client_spec(Name) ->
{Endpoints, ClientConfig} = kflow:kafka_client_settings(Name),
InitArgs = [Endpoints, Name, ClientConfig],
{ _Id = Name
, _Start = {brod_client, start_link, InitArgs}
, _Restart = {permanent, 10}
, _Shutdown = 5000
, _Type = worker
, _Module = [brod_client]
}.