%%% %%% Copyright (c) 2016-2017 Klarna AB %%% %%% Licensed under the Apache License, Version 2.0 (the "License"); %%% you may not use this file except in compliance with the License. %%% You may obtain a copy of the License at %%% %%% http://www.apache.org/licenses/LICENSE-2.0 %%% %%% Unless required by applicable law or agreed to in writing, software %%% distributed under the License is distributed on an "AS IS" BASIS, %%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %%% See the License for the specific language governing permissions and %%% limitations under the License. %%% %% A brucke config file is a YAML file. %% Cluster names and client names must comply to erlang atom syntax. %% %% kafka_clusters: %% kafka_cluster_1: %% - localhost:9092 %% kafka_cluster_2: %% - kafka-1:9092 %% - kafka-2:9092 %% brod_clients: %% - client: brod_client_1 %% cluster: kafka_cluster_1 %% config: %% ssl: %% # start with "priv/" or provide full path %% cacertfile: priv/ssl/ca.crt %% certfile: priv/ssl/client.crt %% keyfile: priv/ssl/client.key %% routes: %% - upstream_client: brod_client_1 %% downstream_client: brod_client_1 %% upstream_topics: %% - "topic_1" %% downstream_topic: "topic_2" %% repartitioning_strategy: strict_p2p %% default_begin_offset: earliest # optional %% compression: no_compression # optional %% -module(brucke_config). -export([ init/0 , is_configured_client_id/1 , get_cluster_name/1 , get_consumer_group_id/1 , all_clients/0 , all_routes/0 ]). -include("brucke_int.hrl"). -define(CONFIG_FILE_ENV_VAR_NAME, "BRUCKE_CONFIG_FILE"). -define(ETS, ?MODULE). -type config_tag() :: atom() | string() | binary(). -type config_value() :: atom() | string() | integer(). -type config_entry() :: {config_tag(), config_value() | config()}. -type config() :: [config_entry()]. -type client_id() :: brod:client_id(). %%%_* APIs ===================================================================== -spec init() -> ok | no_return(). init() -> File = get_file_path_from_config(), yamerl_app:set_param(node_mods, [yamerl_node_erlang_atom]), try [Configs] = yamerl_constr:file(File, [{erlang_atom_autodetection, true}]), do_init(Configs) catch C : E -> lager:emergency("failed to load brucke config file ~s: ~p:~p\n~p", [File, C, E, erlang:get_stacktrace()]), exit({bad_brucke_config, File}) end. -spec is_configured_client_id(brod_client_id()) -> boolean(). is_configured_client_id(ClientId) when is_atom(ClientId) -> case lookup(ClientId) of false -> false; {ClientId, _, _} -> true end. -spec get_cluster_name(brod_client_id()) -> cluster_name(). get_cluster_name(ClientId) when is_atom(ClientId) -> {ClientId, ClusterName, _Config} = lookup(ClientId), ClusterName. -spec get_consumer_group_id(brod_client_id()) -> consumer_group_id(). get_consumer_group_id(ClientId) when is_atom(ClientId) -> iolist_to_binary([get_cluster_name(ClientId), "-brucke-cg"]). -spec all_clients() -> [client()]. all_clients() -> [{ClientId, begin {ClusterName, Endpoints} = lookup(ClusterName), Endpoints end, ClientConfig } || {ClientId, ClusterName, ClientConfig} <- ets:tab2list(?ETS)]. -spec all_routes() -> [route()]. all_routes() -> brucke_routes:all(). %%%_* Internal functions ======================================================= -spec get_file_path_from_config() -> filename() | no_return(). get_file_path_from_config() -> case os:getenv("BRUCKE_CONFIG_FILE") of false -> case application:get_env(brucke, config_file) of {ok, Path0} -> Path = assert_file(Path0), lager:info("Using brucke config file from application environment " "'config_file': ~p", [Path]), Path; ?undef -> lager:emergency("Brucke config file not found! " "It can either be specified by " "environment variable ~s, " "or in ~p application environment (sys.config)", [?CONFIG_FILE_ENV_VAR_NAME, ?APPLICATION]), exit(brucke_config_not_found) end; Path -> lager:info("Using brucke config file from OS env ~s: ~s", [?CONFIG_FILE_ENV_VAR_NAME, Path]), assert_file(Path) end. -spec assert_file(filename() | {priv, filename()}) -> filename() | no_return(). assert_file({priv, Path}) -> assert_file(filename:join(code:priv_dir(?APPLICATION), Path)); assert_file(Path) -> case filelib:is_regular(Path) of true -> Path; false -> lager:emergency("~s is not a regular file", [Path]), exit({bad_brucke_config_file, Path}) end. -spec do_init([config()]) -> ok | no_return(). do_init(Configs) -> Kf = fun(K) -> case lists:keyfind(K, 1, Configs) of {K, V} -> V; false -> lager:emergency("kafka_cluster is not found in config"), exit({mandatory_config_entry_not_found, K}) end end, Clusters = Kf(kafka_clusters), Clients = Kf(brod_clients), Routes = Kf(routes), case ets:info(?ETS) of ?undef -> ok; _ -> lager:emergency("config already loaded"), exit({?ETS, already_created}) end, ?ETS = ets:new(?ETS, [named_table, protected, set]), try init(Clusters, Clients, Routes) catch exit : Reason -> ok = destroy(), erlang:exit(Reason); error : Reason -> ok = destroy(), erlang:exit({error, Reason, erlang:get_stacktrace()}) end. -spec destroy() -> ok. destroy() -> try ets:delete(?ETS), ok catch error : badarg -> ok end. lookup(Key) -> case ets:lookup(?ETS, Key) of [] -> false; [R] -> R end. -spec init(config(), config(), config()) -> ok | no_return(). init(Clusters, _, _) when not is_list(Clusters) orelse Clusters == [] -> lager:emergency("Expecting list of kafka clusters " "Got ~P\n", [Clusters, 9]), exit(bad_cluster_list); init(_, Clients, _) when not is_list(Clients) orelse Clients == [] -> lager:emergency("Expecting list of brod clients " "Got ~P\n", [Clients, 9]), exit(bad_client_list); init(_, _, Routes) when not is_list(Routes) orelse Routes == [] -> lager:emergency("Expecting list of brucke routes " "Got ~P\n", [Routes, 9]), exit(bad_route_list); init(Clusters, Clients, Routes) -> lists:foreach( fun(Cluster) -> {ClusterName, Endpoints} = validate_cluster(Cluster), case lookup(ClusterName) of false -> ok; {ClusterName, _} -> lager:emergency("Duplicated cluster name ~p", [ClusterName]), exit({duplicated_cluster_name, ClusterName}) end, ets:insert(?ETS, {ClusterName, Endpoints}) end, Clusters), lists:foreach( fun(Client) -> {ClientId, ClusterName, ClientConfig} = validate_client(Client), case lookup(ClientId) of false -> ok; _ -> lager:emergency("Duplicated brod client id ~p", [ClientId]), exit({duplicated_brod_client_id, ClientId}) end, case lookup(ClusterName) of false -> lager:emergency("Cluster name ~s for client ~p is not found", [ClusterName, ClientId]), exit({cluster_not_found_for_client, ClusterName, ClientId}); _ -> ok end, ets:insert(?ETS, {ClientId, ClusterName, ClientConfig}) end, Clients), ok = brucke_routes:init(Routes). validate_cluster({ClusterId, [_|_] = Endpoints}) -> {ensure_binary(ClusterId), [validate_endpoint(Endpoint) || Endpoint <- Endpoints]}; validate_cluster(Other) -> lager:emergency("Expecing cluster config with cluster id " "and a list of hostname:port endpoints"), exit({bad_cluster_config, Other}). validate_client(Client) -> try {_, ClientId} = lists:keyfind(client, 1, Client), {_, ClusterName} = lists:keyfind(cluster, 1, Client), Config0 = proplists:get_value(config, Client, []), Config = validate_client_config(ClientId, Config0), {ensure_atom(ClientId), ensure_binary(ClusterName), Config} catch error:Reason -> lager:emergency("Bad brod client config: ~P.\nreason=~p\nstack=~p", [Client, 9, Reason, erlang:get_stacktrace()]), exit(bad_client_config) end. ensure_atom(A) when is_atom(A) -> A. ensure_binary(A) when is_atom(A) -> ensure_binary(atom_to_list(A)); ensure_binary(L) when is_list(L) -> list_to_binary(L); ensure_binary(B) when is_binary(B) -> B. validate_endpoint(HostPort) when is_list(HostPort) -> case string:tokens(HostPort, ":") of [Host, Port] -> try {Host, list_to_integer(Port)} catch _ : _ -> exit_on_bad_endpoint(HostPort) end; _Other -> exit_on_bad_endpoint(HostPort) end; validate_endpoint(Other) -> exit_on_bad_endpoint(Other). exit_on_bad_endpoint(Bad) -> lager:emergency("Expecting endpoints string of patern Host:Port\n" "Got ~P", [Bad, 9]), exit(bad_endpoint). validate_client_config(ClientId, Config) when is_list(Config) -> lists:map(fun(ConfigEntry) -> do_validate_client_config(ClientId, ConfigEntry) end, Config); validate_client_config(ClientId, Config) -> lager:emergency("Expecing client config to be a list for client ~p.\nGot:~p", [ClientId, Config]), exit(bad_client_config). do_validate_client_config(ClientId, {ssl, Options}) -> {ssl, validate_ssl_files(ClientId, Options)}; do_validate_client_config(_ClientId, {_, _} = ConfigEntry) -> ConfigEntry; do_validate_client_config(ClientId, Other) -> lager:emergency("Unknown client config entry for client ~p," "expecting kv-pair\nGot:~p", [ClientId, Other]), exit(bad_client_config_entry). -spec validate_ssl_files(client_id(), list()) -> list() | none(). validate_ssl_files(ClientId, SslOptions) -> lists:foldl( fun(OptName, OptIn) -> validate_ssl_files(ClientId, OptIn, OptName) end, SslOptions, [ {mandatory, cacertfile} , {optional, certfile} , {optional, keyfile} ]). -spec validate_ssl_files(client_id(), list(), {mandatory | optional, cacertfile | certfile | keyfile}) -> list() | none(). validate_ssl_files(ClientId, SslOptions, {MandatoryOr, OptName}) -> case lists:keyfind(OptName, 1, SslOptions) of {_, Filename0} -> Filename = validate_ssl_file(ClientId, Filename0), lists:keyreplace(OptName, 1, SslOptions, {OptName, Filename}); false when MandatoryOr =:= mandatory -> lager:emergency("ssl option '~p' is not found for client ~p", [OptName, ClientId]), exit(missing_ssl_option); false when MandatoryOr =:= optional -> SslOptions end. -spec validate_ssl_file(client_id(), filename()) -> filename() | none(). validate_ssl_file(ClientId, Filename) -> Path = case filename:split(Filename) of ["priv" | PrivPath] -> filename:join([code:priv_dir(?APPLICATION) | PrivPath]); _ -> Filename end, case filelib:is_regular(Path) of true -> Path; false -> lager:emergency("ssl file ~p not found for client ~p", [Path, ClientId]), exit(bad_ssl_file) end. %%%_* Tests ==================================================================== -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). validate_ssl_files_test() -> ?assertException(exit, missing_ssl_option, validate_ssl_files(client_id, [])), %% cacertfile is mandatory, and bad file should trigger exception ?assertException(exit, bad_ssl_file, validate_ssl_files(client_id, [{cacertfile, "no-such-file"}])), %% certfile is optional but providing a bad file should still %% raise an exception ?assertException(exit, bad_ssl_file, validate_ssl_files(client_id, [{cacertfile, "priv/ssl/ca.crt"}, {certfile, "no-such-file"}])), %% OK case ?assertMatch([{cacertfile, _}], validate_ssl_files(client_id, [{cacertfile, "priv/ssl/ca.crt"}])), ?assertMatch([{cacertfile, _}, {keyfile, _}, {certfile, _}], validate_ssl_files(client_id, [{cacertfile, "priv/ssl/ca.crt"}, {keyfile, "priv/ssl/client.key"}, {certfile, "priv/ssl/client.crt"} ])). -endif. %%%_* Emacs ==================================================================== %%% Local Variables: %%% allout-layout: t %%% erlang-indent-level: 2 %%% End: