perf(gateway): memoise the parsed internal rpc url (#2174)

fluxer_gateway/src/gateway/gateway_http_client.erl

@@ -39,7 +39,9 @@
     max_concurrency => pos_integer(),
     failure_threshold => pos_integer(),
     recovery_timeout_ms => pos_integer(),
-    content_type => binary() | string()
+    content_type => binary() | string(),
+    host_key => binary(),
+    is_https => boolean()
 }.
 -type response() :: {ok, non_neg_integer(), [{binary(), binary()}], binary()} | {error, term()}.
 
@@ -77,7 +79,7 @@ request(Workload, Method, Url, Headers, Body, Opts) when is_map(Opts) ->
     MaxConcurrency = maps:get(max_concurrency, WorkloadOpts),
     FailureThreshold = maps:get(failure_threshold, WorkloadOpts),
     RecoveryTimeoutMs = maps:get(recovery_timeout_ms, WorkloadOpts),
-    Host = gateway_http_client_request:extract_host_key(Url),
+    Host = resolve_host_key(Url, WorkloadOpts),
     CircuitKey = {Workload, Host},
     do_request_with_circuit(
         CircuitKey,
@@ -258,6 +260,13 @@ request_with_acquired_slot(
         gateway_http_client_response:release_inflight_slot(Workload)
     end.
 
+-spec resolve_host_key(iodata(), request_options()) -> binary().
+resolve_host_key(Url, Opts) ->
+    case maps:get(host_key, Opts, undefined) of
+        Host when is_binary(Host) -> Host;
+        _ -> gateway_http_client_request:extract_host_key(Url)
+    end.
+
 -spec ensure_runtime(workload()) -> ok.
 ensure_runtime(Workload) ->
     ok = ensure_started(),

fluxer_gateway/src/gateway/gateway_http_client_request.erl

@@ -6,6 +6,7 @@
 -export([
     safe_do_request/6,
     extract_host_key/1,
+    url_metadata/1,
     ensure_binary/1,
     ensure_list/1
 ]).
@@ -20,7 +21,9 @@
     max_concurrency => pos_integer(),
     failure_threshold => pos_integer(),
     recovery_timeout_ms => pos_integer(),
-    content_type => binary() | string()
+    content_type => binary() | string(),
+    host_key => binary(),
+    is_https => boolean()
 }.
 -type response() :: {ok, non_neg_integer(), [{binary(), binary()}], binary()} | {error, term()}.
 
@@ -57,6 +60,18 @@ extract_host_key(Url) ->
         _:_ -> <<"unknown">>
     end.
 
+-spec url_metadata(iodata()) -> {binary(), boolean()}.
+url_metadata(Url) ->
+    UrlString = ensure_list(Url),
+    try uri_string:parse(UrlString) of
+        Parsed when is_map(Parsed) ->
+            {extract_host_from_parsed(Parsed), parsed_is_https(Parsed)};
+        _ ->
+            {<<"unknown">>, false}
+    catch
+        _:_ -> {<<"unknown">>, false}
+    end.
+
 -spec ensure_binary(term()) -> binary().
 ensure_binary(Value) when is_binary(Value) -> Value;
 ensure_binary(Value) -> iodata_to_binary_or_format(Value).
@@ -99,17 +114,25 @@ build_http_options(UrlString, Opts) ->
         {timeout, RecvTimeout},
         {autoredirect, false}
     ],
-    case is_https_url(UrlString) of
+    case is_https_request(UrlString, Opts) of
         true -> [{ssl, https_ssl_options()} | BaseOptions];
         false -> BaseOptions
     end.
 
+-spec is_https_request(string(), request_options()) -> boolean().
+is_https_request(UrlString, Opts) ->
+    case maps:get(is_https, Opts, undefined) of
+        IsHttps when is_boolean(IsHttps) -> IsHttps;
+        _ -> is_https_url(UrlString)
+    end.
+
 -spec is_https_url(string()) -> boolean().
 is_https_url(UrlString) ->
-    case uri_string:parse(UrlString) of
-        #{scheme := Scheme} -> string:lowercase(ensure_list(Scheme)) =:= "https";
-        _ -> false
-    end.
+    parsed_is_https(uri_string:parse(UrlString)).
+
+-spec parsed_is_https(term()) -> boolean().
+parsed_is_https(#{scheme := Scheme}) -> string:lowercase(ensure_list(Scheme)) =:= "https";
+parsed_is_https(_) -> false.
 
 -spec https_ssl_options() -> list().
 https_ssl_options() ->
@@ -302,4 +325,49 @@ build_http_options_omits_ssl_for_http_test() ->
     ?assertEqual({connect_timeout, 5000}, lists:keyfind(connect_timeout, 1, Options)),
     ?assertEqual({autoredirect, false}, lists:keyfind(autoredirect, 1, Options)).
 
+memoisable_urls() ->
+    [
+        <<"http://fluxer-api.fluxer.svc.cluster.local/internal/rpc">>,
+        <<"https://fcm.googleapis.com/v1/projects/fluxer/messages:send">>,
+        <<"http://fluxer-api.fluxer.svc.cluster.local:8080/internal/rpc">>,
+        <<"HTTPS://Fluxer-API.Fluxer.SVC.Cluster.Local:8443/internal/rpc">>
+    ].
+
+url_metadata_matches_derive_path_test_() ->
+    [
+        {binary_to_list(Url), fun() ->
+            Derived = {extract_host_key(Url), is_https_url(ensure_list(Url))},
+            ?assertEqual(Derived, url_metadata(Url))
+        end}
+     || Url <- memoisable_urls()
+    ].
+
+url_metadata_values_test() ->
+    ?assertEqual(
+        [
+            {<<"fluxer-api.fluxer.svc.cluster.local">>, false},
+            {<<"fcm.googleapis.com">>, true},
+            {<<"fluxer-api.fluxer.svc.cluster.local">>, false},
+            {<<"fluxer-api.fluxer.svc.cluster.local">>, true}
+        ],
+        [url_metadata(Url) || Url <- memoisable_urls()]
+    ).
+
+url_metadata_handles_unparsable_url_test() ->
+    ?assertEqual({<<"unknown">>, false}, url_metadata(<<"not-a-url">>)).
+
+build_http_options_uses_memoised_scheme_test() ->
+    case has_ca_store() of
+        true ->
+            Opts = #{connect_timeout => 3000, recv_timeout => 5000, is_https => true},
+            ?assertMatch([{ssl, _} | _], build_http_options("http://127.0.0.1:8088/rpc", Opts));
+        false ->
+            ok
+    end.
+
+build_http_options_memoised_http_skips_ssl_test() ->
+    Opts = #{connect_timeout => 3000, recv_timeout => 5000, is_https => false},
+    Options = build_http_options("https://fcm.googleapis.com/v1", Opts),
+    ?assertEqual(false, lists:keyfind(ssl, 1, Options)).
+
 -endif.

fluxer_gateway/src/gateway/rpc_client.erl

@@ -9,6 +9,7 @@
     handle_http_response/2,
     rpc_headers/1,
     rpc_url/0,
+    rpc_url_meta/0,
     is_retryable/1,
     backoff_delay/2
 ]).
@@ -17,6 +18,7 @@
 
 -define(DEFAULT_RPC_PATH, <<"/internal/rpc">>).
 -define(RPC_AUTH_HEADER, <<"x-fluxer-rpc-auth">>).
+-define(URL_META_TERM_KEY, {?MODULE, rpc_url_meta}).
 
 -type rpc_request() :: map().
 -type rpc_response() :: {ok, map()} | {error, term()}.
@@ -132,14 +134,16 @@ backoff_delay(Attempt, {_MaxAttempts, BaseMs, MaxMs, JitterMs}) ->
 
 -spec do_request(rpc_request()) -> rpc_response().
 do_request(Request) ->
-    Url = rpc_url(),
+    {Url, HostKey, IsHttps} = rpc_url_meta(),
     Timeout = request_timeout_ms(),
     Payload = iolist_to_binary(json:encode(Request)),
     Headers = rpc_headers(Request),
     RequestOpts = #{
         connect_timeout => min(Timeout, 5000),
         recv_timeout => Timeout,
-        content_type => <<"application/json">>
+        content_type => <<"application/json">>,
+        host_key => HostKey,
+        is_https => IsHttps
     },
     case gateway_http_client:request(rpc, post, Url, Headers, Payload, RequestOpts) of
         {ok, StatusCode, _ResponseHeaders, ResponseBody} ->
@@ -175,6 +179,21 @@ rpc_url() ->
             trim_trailing_slash(ConfiguredEndpoint)
     end.
 
+-spec rpc_url_meta() -> {binary(), binary(), boolean()}.
+rpc_url_meta() ->
+    Url = rpc_url(),
+    case persistent_term:get(?URL_META_TERM_KEY, undefined) of
+        {Url, _HostKey, _IsHttps} = Meta -> Meta;
+        _ -> store_rpc_url_meta(Url)
+    end.
+
+-spec store_rpc_url_meta(binary()) -> {binary(), binary(), boolean()}.
+store_rpc_url_meta(Url) ->
+    {HostKey, IsHttps} = gateway_http_client_request:url_metadata(Url),
+    Meta = {Url, HostKey, IsHttps},
+    persistent_term:put(?URL_META_TERM_KEY, Meta),
+    Meta.
+
 -spec trim_trailing_slash(binary()) -> binary().
 trim_trailing_slash(<<>>) ->
     <<>>;

fluxer_gateway/test/rpc_client_tests.erl

@@ -61,6 +61,47 @@ rpc_url_uses_configured_endpoint_without_trailing_slash_test() ->
         _ = fluxer_gateway_env:update(fun(_) -> OldConfig end)
     end.
 
+rpc_url_meta_matches_derive_path_test() ->
+    OldConfig = fluxer_gateway_env:get_map(),
+    Endpoints = [
+        <<"http://fluxer-api.fluxer.svc.cluster.local/internal/rpc">>,
+        <<"https://api.internal/internal/rpc">>,
+        <<"http://fluxer-api.fluxer.svc.cluster.local:8080/internal/rpc">>,
+        <<"HTTPS://Fluxer-API.Fluxer.SVC.Cluster.Local:8443/internal/rpc">>
+    ],
+    try
+        lists:foreach(fun assert_rpc_url_meta_for_endpoint/1, Endpoints)
+    after
+        _ = fluxer_gateway_env:update(fun(_) -> OldConfig end)
+    end.
+
+assert_rpc_url_meta_for_endpoint(Endpoint) ->
+    _ = fluxer_gateway_env:patch(#{api_rpc_endpoint => Endpoint}),
+    Url = rpc_client:rpc_url(),
+    {HostKey, IsHttps} = gateway_http_client_request:url_metadata(Url),
+    ?assertEqual(HostKey, gateway_http_client_request:extract_host_key(Url)),
+    ?assertEqual({Url, HostKey, IsHttps}, rpc_client:rpc_url_meta()),
+    ?assertEqual({Url, HostKey, IsHttps}, rpc_client:rpc_url_meta()).
+
+rpc_url_meta_expected_values_test() ->
+    OldConfig = fluxer_gateway_env:get_map(),
+    try
+        _ = fluxer_gateway_env:patch(#{
+            api_rpc_endpoint =>
+                <<"https://Fluxer-API.Fluxer.SVC.Cluster.Local:8443/internal/rpc">>
+        }),
+        ?assertEqual(
+            {
+                <<"https://Fluxer-API.Fluxer.SVC.Cluster.Local:8443/internal/rpc">>,
+                <<"fluxer-api.fluxer.svc.cluster.local">>,
+                true
+            },
+            rpc_client:rpc_url_meta()
+        )
+    after
+        _ = fluxer_gateway_env:update(fun(_) -> OldConfig end)
+    end.
+
 is_retryable_timeout_test() ->
     ?assert(rpc_client:is_retryable(timeout)).