From 72ffa3bd9ad9709901ada461310e7f75e98d8402 Mon Sep 17 00:00:00 2001 From: Hampus Date: Sun, 30 Aug 2026 22:48:11 +0200 Subject: [PATCH] perf(gateway): memoise the parsed internal rpc url (#2174) --- .../src/gateway/gateway_http_client.erl | 13 ++- .../gateway/gateway_http_client_request.erl | 80 +++++++++++++++++-- fluxer_gateway/src/gateway/rpc_client.erl | 23 +++++- fluxer_gateway/test/rpc_client_tests.erl | 41 ++++++++++ 4 files changed, 147 insertions(+), 10 deletions(-) diff --git a/fluxer_gateway/src/gateway/gateway_http_client.erl b/fluxer_gateway/src/gateway/gateway_http_client.erl index 6c2612b97..d2534dd6f 100644 --- a/fluxer_gateway/src/gateway/gateway_http_client.erl +++ b/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(), diff --git a/fluxer_gateway/src/gateway/gateway_http_client_request.erl b/fluxer_gateway/src/gateway/gateway_http_client_request.erl index 68d47ba25..f82c32a0d 100644 --- a/fluxer_gateway/src/gateway/gateway_http_client_request.erl +++ b/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. diff --git a/fluxer_gateway/src/gateway/rpc_client.erl b/fluxer_gateway/src/gateway/rpc_client.erl index c4cb355bb..c6a5c31c3 100644 --- a/fluxer_gateway/src/gateway/rpc_client.erl +++ b/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(<<>>) -> <<>>; diff --git a/fluxer_gateway/test/rpc_client_tests.erl b/fluxer_gateway/test/rpc_client_tests.erl index 960c76012..358bf253a 100644 --- a/fluxer_gateway/test/rpc_client_tests.erl +++ b/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)).