diff --git a/api/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto b/api/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto index af69d33a63401..402937fff28fd 100644 --- a/api/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto +++ b/api/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto @@ -23,7 +23,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE; // Redis Proxy :ref:`configuration overview `. // [#extension: envoy.filters.network.redis_proxy] -// [#next-free-field: 8] +// [#next-free-field: 9] message RedisProxy { option (udpa.annotations.versioning).previous_message_type = "envoy.config.filter.network.redis_proxy.v2.RedisProxy"; @@ -183,6 +183,31 @@ message RedisProxy { Route catch_all_route = 4; } + // RedisFault defines faults used for fault injection. + message RedisFault { + enum RedisFaultType { + // Delays requests. This is the base fault; other faults can have delays added. + DELAY = 0; + + // Returns errors on requests. + ERROR = 1; + } + + // Fault type. + RedisFaultType fault_type = 1 [(validate.rules).enum = {defined_only: true}]; + + // Percentage of requests fault applies to. + config.core.v3.RuntimeFractionalPercent fault_enabled = 2 + [(validate.rules).message = {required: true}]; + + // Delay for all faults. If not set, defaults to zero + google.protobuf.Duration delay = 3; + + // Commands fault is restricted to, if any. If not set, fault applies to all commands + // other than auth and ping (due to special handling of those commands in Envoy). + repeated string commands = 4; + } + reserved 2; reserved "cluster"; @@ -236,6 +261,35 @@ message RedisProxy { // AUTH, but no password is set" error will be returned. config.core.v3.DataSource downstream_auth_password = 6 [(udpa.annotations.sensitive) = true]; + // List of faults to inject. Faults currently come in two flavors: + // - Delay, which delays a request. + // - Error, which responds to a request with an error. Errors can also have delays attached. + // + // Example: + // + // .. code-block:: yaml + // + // faults: + // - fault_type: ERROR + // fault_enabled: + // default_value: + // numerator: 10 + // denominator: HUNDRED + // runtime_key: "bogus_key" + // commands: + // - GET + // - fault_type: DELAY + // fault_enabled: + // default_value: + // numerator: 10 + // denominator: HUNDRED + // runtime_key: "bogus_key" + // delay: 2s + // + // See the :ref:`fault injection section + // ` for more information on how to configure this. + repeated RedisFault faults = 8; + // If a username is provided an ACL style AUTH command will be required with a username and password. // Authenticate Redis client connections locally by forcing downstream clients to issue a `Redis // AUTH command `_ with this username and the *downstream_auth_password* diff --git a/docs/root/configuration/listeners/network_filters/redis_proxy_filter.rst b/docs/root/configuration/listeners/network_filters/redis_proxy_filter.rst index 3c3fb77f3861f..6adf7c8ffb271 100644 --- a/docs/root/configuration/listeners/network_filters/redis_proxy_filter.rst +++ b/docs/root/configuration/listeners/network_filters/redis_proxy_filter.rst @@ -58,7 +58,9 @@ changed to microseconds by setting the configuration parameter :ref:`latency_in_ total, Counter, Number of commands success, Counter, Number of commands that were successful error, Counter, Number of commands that returned a partial or complete error response - latency, Histogram, Command execution time in milliseconds + latency, Histogram, Command execution time in milliseconds (including delay faults) + error_fault, Counter, Number of commands that had an error fault injected + delay_fault, Counter, Number of commands that had a delay fault injected .. _config_network_filters_redis_proxy_per_command_stats: @@ -70,3 +72,51 @@ The Redis proxy filter supports the following runtime settings: redis.drain_close_enabled % of connections that will be drain closed if the server is draining and would otherwise attempt a drain close. Defaults to 100. + +.. _config_network_filters_redis_proxy_fault_injection: + +Fault Injection +--------------- + +The Redis filter can perform fault injection. Currently, Delay and Error faults are supported. +Delay faults delay a request, and Error faults respond with an error. Moreover, errors can be delayed. + +Note that the Redis filter does not check for correctness in your configuration - it is the user's +responsibility to make sure both the default and runtime percentages are correct! This is because +percentages can be changed during runtime, and validating correctness at request time is expensive. +If multiple faults are specified, the fault injection percentage should not exceed 100% for a given +fault and Redis command combination. For example, if two faults are specified; one applying to GET at 60 +%, and one applying to all commands at 50%, that is a bad configuration as GET now has 110% chance of +applying a fault. This means that every request will have a fault. + +If a delay is injected, the delay is additive - if the request took 400ms and a delay of 100ms +is injected, then the total request latency is 500ms. Also, due to implementation of the redis protocol, +a delayed request will delay everything that comes in after it, due to the proxy's need to respect the +order of commands it receives. + +Note that faults must have a `fault_enabled` field, and are not enabled by default (if no default value +or runtime key are set). + +Example configuration: + +.. code-block:: yaml + + faults: + - fault_type: ERROR + fault_enabled: + default_value: + numerator: 10 + denominator: HUNDRED + runtime_key: "bogus_key" + commands: + - GET + - fault_type: DELAY + fault_enabled: + default_value: + numerator: 10 + denominator: HUNDRED + runtime_key: "bogus_key" + delay: 2s + +This creates two faults- an error, applying only to GET commands at 10%, and a delay, applying to all +commands at 10%. This means that 20% of GET commands will have a fault applied, as discussed earlier. diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst index eb3465f99eebe..36f1716c05713 100644 --- a/docs/root/version_history/current.rst +++ b/docs/root/version_history/current.rst @@ -37,6 +37,7 @@ New Features * ext_authz filter: added support for emitting dynamic metadata for both :ref:`HTTP ` and :ref:`network ` filters. * grpc-json: support specifying `response_body` field in for `google.api.HttpBody` message. * load balancer: added a :ref:`configuration` option to specify the active request bias used by the least request load balancer. +* redis: added fault injection support :ref:`fault injection for redis proxy `, described further in :ref:`configuration documentation `. * tap: added :ref:`generic body matcher` to scan http requests and responses for text or hex patterns. Deprecated diff --git a/generated_api_shadow/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto b/generated_api_shadow/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto index 8f996c30f9ae4..0bc52493bb294 100644 --- a/generated_api_shadow/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto +++ b/generated_api_shadow/envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.proto @@ -23,7 +23,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE; // Redis Proxy :ref:`configuration overview `. // [#extension: envoy.filters.network.redis_proxy] -// [#next-free-field: 8] +// [#next-free-field: 9] message RedisProxy { option (udpa.annotations.versioning).previous_message_type = "envoy.config.filter.network.redis_proxy.v2.RedisProxy"; @@ -182,6 +182,31 @@ message RedisProxy { [deprecated = true, (envoy.annotations.disallowed_by_default) = true]; } + // RedisFault defines faults used for fault injection. + message RedisFault { + enum RedisFaultType { + // Delays requests. This is the base fault; other faults can have delays added. + DELAY = 0; + + // Returns errors on requests. + ERROR = 1; + } + + // Fault type. + RedisFaultType fault_type = 1 [(validate.rules).enum = {defined_only: true}]; + + // Percentage of requests fault applies to. + config.core.v3.RuntimeFractionalPercent fault_enabled = 2 + [(validate.rules).message = {required: true}]; + + // Delay for all faults. If not set, defaults to zero + google.protobuf.Duration delay = 3; + + // Commands fault is restricted to, if any. If not set, fault applies to all commands + // other than auth and ping (due to special handling of those commands in Envoy). + repeated string commands = 4; + } + // The prefix to use when emitting :ref:`statistics `. string stat_prefix = 1 [(validate.rules).string = {min_bytes: 1}]; @@ -231,6 +256,35 @@ message RedisProxy { // AUTH, but no password is set" error will be returned. config.core.v3.DataSource downstream_auth_password = 6 [(udpa.annotations.sensitive) = true]; + // List of faults to inject. Faults currently come in two flavors: + // - Delay, which delays a request. + // - Error, which responds to a request with an error. Errors can also have delays attached. + // + // Example: + // + // .. code-block:: yaml + // + // faults: + // - fault_type: ERROR + // fault_enabled: + // default_value: + // numerator: 10 + // denominator: HUNDRED + // runtime_key: "bogus_key" + // commands: + // - GET + // - fault_type: DELAY + // fault_enabled: + // default_value: + // numerator: 10 + // denominator: HUNDRED + // runtime_key: "bogus_key" + // delay: 2s + // + // See the :ref:`fault injection section + // ` for more information on how to configure this. + repeated RedisFault faults = 8; + // If a username is provided an ACL style AUTH command will be required with a username and password. // Authenticate Redis client connections locally by forcing downstream clients to issue a `Redis // AUTH command `_ with this username and the *downstream_auth_password* diff --git a/source/extensions/filters/network/common/redis/BUILD b/source/extensions/filters/network/common/redis/BUILD index 3b4dcedbb01e4..8c9c9e32d3ac7 100644 --- a/source/extensions/filters/network/common/redis/BUILD +++ b/source/extensions/filters/network/common/redis/BUILD @@ -100,3 +100,25 @@ envoy_cc_library( "//source/common/stats:utility_lib", ], ) + +envoy_cc_library( + name = "fault_interface", + hdrs = ["fault.h"], + deps = [ + "@envoy_api//envoy/type/v3:pkg_cc_proto", + ], +) + +envoy_cc_library( + name = "fault_lib", + srcs = ["fault_impl.cc"], + hdrs = ["fault_impl.h"], + deps = [ + ":codec_lib", + ":fault_interface", + "//include/envoy/common:random_generator_interface", + "//include/envoy/upstream:upstream_interface", + "//source/common/protobuf:utility_lib", + "@envoy_api//envoy/extensions/filters/network/redis_proxy/v3:pkg_cc_proto", + ], +) diff --git a/source/extensions/filters/network/common/redis/fault.h b/source/extensions/filters/network/common/redis/fault.h new file mode 100644 index 0000000000000..158969455c92b --- /dev/null +++ b/source/extensions/filters/network/common/redis/fault.h @@ -0,0 +1,52 @@ +#pragma once + +#include +#include + +#include "envoy/common/pure.h" +#include "envoy/type/v3/percent.pb.h" + +#include "absl/types/optional.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Common { +namespace Redis { + +/** + * Fault Type. + */ +enum class FaultType { Delay, Error }; + +class Fault { +public: + virtual ~Fault() = default; + + virtual FaultType faultType() const PURE; + virtual std::chrono::milliseconds delayMs() const PURE; + virtual const std::vector commands() const PURE; + virtual envoy::type::v3::FractionalPercent defaultValue() const PURE; + virtual absl::optional runtimeKey() const PURE; +}; + +using FaultSharedPtr = std::shared_ptr; + +class FaultManager { +public: + virtual ~FaultManager() = default; + + /** + * Get fault type and delay given a Redis command. + * @param command supplies the Redis command string. + */ + virtual const Fault* getFaultForCommand(const std::string& command) const PURE; +}; + +using FaultManagerPtr = std::unique_ptr; + +} // namespace Redis +} // namespace Common +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy \ No newline at end of file diff --git a/source/extensions/filters/network/common/redis/fault_impl.cc b/source/extensions/filters/network/common/redis/fault_impl.cc new file mode 100644 index 0000000000000..4b813ad9d799a --- /dev/null +++ b/source/extensions/filters/network/common/redis/fault_impl.cc @@ -0,0 +1,148 @@ +#include "extensions/filters/network/common/redis/fault_impl.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Common { +namespace Redis { + +struct FaultManagerKeyNamesValues { + // The rbac filter rejected the request + const std::string AllKey = "ALL_KEY"; +}; +using FaultManagerKeyNames = ConstSingleton; + +FaultManagerImpl::FaultImpl::FaultImpl( + envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault base_fault) + : commands_(buildCommands(base_fault)) { + delay_ms_ = std::chrono::milliseconds(PROTOBUF_GET_MS_OR_DEFAULT(base_fault, delay, 0)); + + switch (base_fault.fault_type()) { + case envoy::extensions::filters::network::redis_proxy::v3::RedisProxy::RedisFault::DELAY: + fault_type_ = FaultType::Delay; + break; + case envoy::extensions::filters::network::redis_proxy::v3::RedisProxy::RedisFault::ERROR: + fault_type_ = FaultType::Error; + break; + default: + NOT_REACHED_GCOVR_EXCL_LINE; + break; + } + + default_value_ = base_fault.fault_enabled().default_value(); + runtime_key_ = base_fault.fault_enabled().runtime_key(); +}; + +std::vector FaultManagerImpl::FaultImpl::buildCommands( + envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault base_fault) { + std::vector commands; + for (const std::string& command : base_fault.commands()) { + commands.emplace_back(absl::AsciiStrToLower(command)); + } + return commands; +} + +FaultManagerImpl::FaultManagerImpl( + Random::RandomGenerator& random, Runtime::Loader& runtime, + const Protobuf::RepeatedPtrField< + ::envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault> + faults) + : fault_map_(buildFaultMap(faults)), random_(random), runtime_(runtime) {} + +FaultMap FaultManagerImpl::buildFaultMap( + const Protobuf::RepeatedPtrField< + ::envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault> + faults) { + // Next, create the fault map that maps commands to pointers to Fault objects. + // Group faults by command + FaultMap fault_map; + for (auto base_fault : faults) { + auto fault_ptr = std::make_shared(base_fault); + if (!fault_ptr->commands().empty()) { + for (const std::string& command : fault_ptr->commands()) { + fault_map[command].emplace_back(fault_ptr); + } + } else { + // Generic "ALL" entry in map for faults that map to all keys; also add to each command + fault_map[FaultManagerKeyNames::get().AllKey].emplace_back(fault_ptr); + } + } + + // Add the ALL keys faults to each command too so that we can just query faults by command. + // Get all ALL_KEY faults. + FaultMap::iterator it_outer = fault_map.find(FaultManagerKeyNames::get().AllKey); + if (it_outer != fault_map.end()) { + for (const FaultSharedPtr& fault_ptr : it_outer->second) { + FaultMap::iterator it_inner; + for (it_inner = fault_map.begin(); it_inner != fault_map.end(); it_inner++) { + std::string command = it_inner->first; + if (command != FaultManagerKeyNames::get().AllKey) { + fault_map[command].push_back(fault_ptr); + } + } + } + } + return fault_map; +} + +uint64_t FaultManagerImpl::getIntegerNumeratorOfFractionalPercent( + absl::string_view key, const envoy::type::v3::FractionalPercent& default_value) const { + uint64_t numerator; + if (default_value.denominator() == envoy::type::v3::FractionalPercent::HUNDRED) { + numerator = default_value.numerator(); + } else { + int denominator = + ProtobufPercentHelper::fractionalPercentDenominatorToInt(default_value.denominator()); + numerator = (default_value.numerator() * 100) / denominator; + } + return runtime_.snapshot().getInteger(key, numerator); +} + +// Fault checking algorithm: +// +// For example, if we have an ERROR fault at 5% for all commands, and a DELAY fault at 10% for GET, +// if we receive a GET, we want 5% of GETs to get DELAY, and 10% to get ERROR. Thus, we need to +// amortize the percentages. +// +// 0. Get random number. +// 1. Get faults for given command. +// 2. For each fault, calculate the amortized fault injection percentage. +// +// Note that we do not check to make sure the probabilities of faults are <= 100%! +const Fault* FaultManagerImpl::getFaultForCommandInternal(const std::string& command) const { + FaultMap::const_iterator it_outer = fault_map_.find(command); + if (it_outer != fault_map_.end()) { + auto random_number = random_.random() % 100; + int amortized_fault = 0; + + for (const FaultSharedPtr& fault_ptr : it_outer->second) { + uint64_t fault_injection_percentage = getIntegerNumeratorOfFractionalPercent( + fault_ptr->runtimeKey().value(), fault_ptr->defaultValue()); + if (random_number < (fault_injection_percentage + amortized_fault)) { + return fault_ptr.get(); + } else { + amortized_fault += fault_injection_percentage; + } + } + } + + return nullptr; +} + +const Fault* FaultManagerImpl::getFaultForCommand(const std::string& command) const { + if (!fault_map_.empty()) { + if (fault_map_.count(command) > 0) { + return getFaultForCommandInternal(command); + } else { + return getFaultForCommandInternal(FaultManagerKeyNames::get().AllKey); + } + } + + return nullptr; +} + +} // namespace Redis +} // namespace Common +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy \ No newline at end of file diff --git a/source/extensions/filters/network/common/redis/fault_impl.h b/source/extensions/filters/network/common/redis/fault_impl.h new file mode 100644 index 0000000000000..3850a8a4b4c9c --- /dev/null +++ b/source/extensions/filters/network/common/redis/fault_impl.h @@ -0,0 +1,108 @@ +#pragma once + +#include +#include + +#include "envoy/api/api.h" +#include "envoy/common/random_generator.h" +#include "envoy/extensions/filters/network/redis_proxy/v3/redis_proxy.pb.h" +#include "envoy/upstream/upstream.h" + +#include "common/protobuf/utility.h" +#include "common/singleton/const_singleton.h" + +#include "extensions/filters/network/common/redis/fault.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Common { +namespace Redis { + +using FaultMap = absl::flat_hash_map>; + +/** + * Message returned for particular types of faults. + */ +struct FaultMessagesValues { + const std::string Error = "Fault Injected: Error"; +}; +using FaultMessages = ConstSingleton; + +/** + * Fault management- creation, storage and retrieval. Faults are queried for by command, + * so they are stored in an unordered map using the command as key. For faults that apply to + * all commands, we use a special ALL_KEYS entry in the map. + */ +class FaultManagerImpl : public FaultManager { +public: + FaultManagerImpl( + Random::RandomGenerator& random, Runtime::Loader& runtime, + const Protobuf::RepeatedPtrField< + ::envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault> + base_faults); + + const Fault* getFaultForCommand(const std::string& command) const override; + + static FaultSharedPtr makeFaultForTest(Common::Redis::FaultType fault_type, + std::chrono::milliseconds delay_ms) { + envoy::type::v3::FractionalPercent default_value; + default_value.set_numerator(100); + default_value.set_denominator(envoy::type::v3::FractionalPercent::HUNDRED); + FaultImpl fault = + FaultImpl(fault_type, delay_ms, std::vector(), default_value, "foo"); + return std::make_shared(fault); + } + + // Allow the unit test to have access to private members. + friend class FaultTest; + +private: + class FaultImpl : public Fault { + public: + FaultImpl( + envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault base_fault); + FaultImpl(FaultType fault_type, std::chrono::milliseconds delay_ms, + const std::vector commands, + envoy::type::v3::FractionalPercent default_value, + absl::optional runtime_key) + : fault_type_(fault_type), delay_ms_(delay_ms), commands_(commands), + default_value_(default_value), runtime_key_(runtime_key) {} // For testing only + + FaultType faultType() const override { return fault_type_; }; + std::chrono::milliseconds delayMs() const override { return delay_ms_; }; + const std::vector commands() const override { return commands_; }; + envoy::type::v3::FractionalPercent defaultValue() const override { return default_value_; }; + absl::optional runtimeKey() const override { return runtime_key_; }; + + private: + static std::vector buildCommands( + envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault base_fault); + + FaultType fault_type_; + std::chrono::milliseconds delay_ms_; + const std::vector commands_; + envoy::type::v3::FractionalPercent default_value_; + absl::optional runtime_key_; + }; + + static FaultMap + buildFaultMap(const Protobuf::RepeatedPtrField< + ::envoy::extensions::filters::network::redis_proxy::v3::RedisProxy_RedisFault> + faults); + + uint64_t getIntegerNumeratorOfFractionalPercent( + absl::string_view key, const envoy::type::v3::FractionalPercent& default_value) const; + const Fault* getFaultForCommandInternal(const std::string& command) const; + const FaultMap fault_map_; + +protected: + Random::RandomGenerator& random_; + Runtime::Loader& runtime_; +}; + +} // namespace Redis +} // namespace Common +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy \ No newline at end of file diff --git a/source/extensions/filters/network/redis_proxy/BUILD b/source/extensions/filters/network/redis_proxy/BUILD index 4d452f0cad3c1..b7503144a5f8b 100644 --- a/source/extensions/filters/network/redis_proxy/BUILD +++ b/source/extensions/filters/network/redis_proxy/BUILD @@ -17,6 +17,7 @@ envoy_cc_library( name = "command_splitter_interface", hdrs = ["command_splitter.h"], deps = [ + "//include/envoy/event:dispatcher_interface", "//source/extensions/filters/network/common/redis:codec_interface", ], ) @@ -64,6 +65,7 @@ envoy_cc_library( "//source/common/common:utility_lib", "//source/common/stats:timespan_lib", "//source/extensions/filters/network/common/redis:client_lib", + "//source/extensions/filters/network/common/redis:fault_lib", "//source/extensions/filters/network/common/redis:supported_commands_lib", "//source/extensions/filters/network/common/redis:utility_lib", ], @@ -125,6 +127,7 @@ envoy_cc_extension( "//source/extensions/filters/network:well_known_names", "//source/extensions/filters/network/common:factory_base_lib", "//source/extensions/filters/network/common/redis:codec_lib", + "//source/extensions/filters/network/common/redis:fault_lib", "//source/extensions/filters/network/common/redis:redis_command_stats_lib", "//source/extensions/filters/network/redis_proxy:command_splitter_lib", "//source/extensions/filters/network/redis_proxy:conn_pool_lib", diff --git a/source/extensions/filters/network/redis_proxy/command_splitter.h b/source/extensions/filters/network/redis_proxy/command_splitter.h index e03d0a92e1377..d5408b11ab0cd 100644 --- a/source/extensions/filters/network/redis_proxy/command_splitter.h +++ b/source/extensions/filters/network/redis_proxy/command_splitter.h @@ -3,6 +3,7 @@ #include #include "envoy/common/pure.h" +#include "envoy/event/dispatcher.h" #include "extensions/filters/network/common/redis/codec.h" @@ -80,6 +81,27 @@ class Instance { SplitCallbacks& callbacks) PURE; }; +using CommandSplitterPtr = std::unique_ptr; + +/** + * A command splitter factory that allows creation of the command splitter when + * we have access to the dispatcher parameter. This supports fault injection, + * specifically delay faults, which rely on the dispatcher for creating delay timers. + */ +class CommandSplitterFactory { +public: + virtual ~CommandSplitterFactory() = default; + + /** + * Create a command splitter. + * @param dispatcher supplies the dispatcher . + * @return CommandSplitterPtr a handle to a newly created command splitter. + */ + virtual CommandSplitterPtr create(Event::Dispatcher& dispatcher) PURE; +}; + +using CommandSplitterFactorySharedPtr = std::shared_ptr; + } // namespace CommandSplitter } // namespace RedisProxy } // namespace NetworkFilters diff --git a/source/extensions/filters/network/redis_proxy/command_splitter_impl.cc b/source/extensions/filters/network/redis_proxy/command_splitter_impl.cc index a5bd89588f51a..624cead01a9ca 100644 --- a/source/extensions/filters/network/redis_proxy/command_splitter_impl.cc +++ b/source/extensions/filters/network/redis_proxy/command_splitter_impl.cc @@ -79,7 +79,9 @@ void SplitRequestBase::updateStats(const bool success) { } else { command_stats_.error_.inc(); } - command_latency_->complete(); + if (command_latency_ != nullptr) { + command_latency_->complete(); + } } SingleServerRequest::~SingleServerRequest() { ASSERT(!handle_); } @@ -90,10 +92,12 @@ void SingleServerRequest::onResponse(Common::Redis::RespValuePtr&& response) { callbacks_.onResponse(std::move(response)); } -void SingleServerRequest::onFailure() { +void SingleServerRequest::onFailure() { onFailure(Response::get().UpstreamFailure); } + +void SingleServerRequest::onFailure(std::string error_msg) { handle_ = nullptr; updateStats(false); - callbacks_.onResponse(Common::Redis::Utility::makeError(Response::get().UpstreamFailure)); + callbacks_.onResponse(Common::Redis::Utility::makeError(error_msg)); } void SingleServerRequest::cancel() { @@ -101,13 +105,44 @@ void SingleServerRequest::cancel() { handle_ = nullptr; } +SplitRequestPtr ErrorFaultRequest::create(SplitCallbacks& callbacks, CommandStats& command_stats, + TimeSource& time_source, bool delay_command_latency) { + std::unique_ptr request_ptr{ + new ErrorFaultRequest(callbacks, command_stats, time_source, delay_command_latency)}; + + request_ptr->onFailure(Common::Redis::FaultMessages::get().Error); + command_stats.error_fault_.inc(); + return nullptr; +} + +std::unique_ptr DelayFaultRequest::create(SplitCallbacks& callbacks, + CommandStats& command_stats, + TimeSource& time_source, + Event::Dispatcher& dispatcher, + std::chrono::milliseconds delay) { + return std::make_unique(callbacks, command_stats, time_source, dispatcher, + delay); +} + +void DelayFaultRequest::onResponse(Common::Redis::RespValuePtr&& response) { + response_ = std::move(response); + delay_timer_->enableTimer(delay_); +} + +void DelayFaultRequest::onDelayResponse() { + command_stats_.delay_fault_.inc(); + command_latency_->complete(); // Complete latency of the command stats of the wrapped request + callbacks_.onResponse(std::move(response_)); +} + +void DelayFaultRequest::cancel() { delay_timer_->disableTimer(); } + SplitRequestPtr SimpleRequest::create(Router& router, Common::Redis::RespValuePtr&& incoming_request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source) { + TimeSource& time_source, bool delay_command_latency) { std::unique_ptr request_ptr{ - new SimpleRequest(callbacks, command_stats, time_source)}; - + new SimpleRequest(callbacks, command_stats, time_source, delay_command_latency)}; const auto route = router.upstreamPool(incoming_request->asArray()[1].asString()); if (route) { Common::Redis::RespValueSharedPtr base_request = std::move(incoming_request); @@ -126,7 +161,7 @@ SplitRequestPtr SimpleRequest::create(Router& router, SplitRequestPtr EvalRequest::create(Router& router, Common::Redis::RespValuePtr&& incoming_request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source) { + TimeSource& time_source, bool delay_command_latency) { // EVAL looks like: EVAL script numkeys key [key ...] arg [arg ...] // Ensure there are at least three args to the command or it cannot be hashed. if (incoming_request->asArray().size() < 4) { @@ -135,7 +170,8 @@ SplitRequestPtr EvalRequest::create(Router& router, Common::Redis::RespValuePtr& return nullptr; } - std::unique_ptr request_ptr{new EvalRequest(callbacks, command_stats, time_source)}; + std::unique_ptr request_ptr{ + new EvalRequest(callbacks, command_stats, time_source, delay_command_latency)}; const auto route = router.upstreamPool(incoming_request->asArray()[3].asString()); if (route) { @@ -177,8 +213,9 @@ void FragmentedRequest::onChildFailure(uint32_t index) { SplitRequestPtr MGETRequest::create(Router& router, Common::Redis::RespValuePtr&& incoming_request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source) { - std::unique_ptr request_ptr{new MGETRequest(callbacks, command_stats, time_source)}; + TimeSource& time_source, bool delay_command_latency) { + std::unique_ptr request_ptr{ + new MGETRequest(callbacks, command_stats, time_source, delay_command_latency)}; request_ptr->num_pending_responses_ = incoming_request->asArray().size() - 1; request_ptr->pending_requests_.reserve(request_ptr->num_pending_responses_); @@ -250,13 +287,14 @@ void MGETRequest::onChildResponse(Common::Redis::RespValuePtr&& value, uint32_t SplitRequestPtr MSETRequest::create(Router& router, Common::Redis::RespValuePtr&& incoming_request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source) { + TimeSource& time_source, bool delay_command_latency) { if ((incoming_request->asArray().size() - 1) % 2 != 0) { onWrongNumberOfArguments(callbacks, *incoming_request); command_stats.error_.inc(); return nullptr; } - std::unique_ptr request_ptr{new MSETRequest(callbacks, command_stats, time_source)}; + std::unique_ptr request_ptr{ + new MSETRequest(callbacks, command_stats, time_source, delay_command_latency)}; request_ptr->num_pending_responses_ = (incoming_request->asArray().size() - 1) / 2; request_ptr->pending_requests_.reserve(request_ptr->num_pending_responses_); @@ -321,13 +359,12 @@ void MSETRequest::onChildResponse(Common::Redis::RespValuePtr&& value, uint32_t } } -SplitRequestPtr SplitKeysSumResultRequest::create(Router& router, - Common::Redis::RespValuePtr&& incoming_request, - SplitCallbacks& callbacks, - CommandStats& command_stats, - TimeSource& time_source) { +SplitRequestPtr +SplitKeysSumResultRequest::create(Router& router, Common::Redis::RespValuePtr&& incoming_request, + SplitCallbacks& callbacks, CommandStats& command_stats, + TimeSource& time_source, bool delay_command_latency) { std::unique_ptr request_ptr{ - new SplitKeysSumResultRequest(callbacks, command_stats, time_source)}; + new SplitKeysSumResultRequest(callbacks, command_stats, time_source, delay_command_latency)}; request_ptr->num_pending_responses_ = incoming_request->asArray().size() - 1; request_ptr->pending_requests_.reserve(request_ptr->num_pending_responses_); @@ -391,13 +428,15 @@ void SplitKeysSumResultRequest::onChildResponse(Common::Redis::RespValuePtr&& va } } -InstanceImpl::InstanceImpl(RouterPtr&& router, Stats::Scope& scope, const std::string& stat_prefix, - TimeSource& time_source, bool latency_in_micros) - : router_(std::move(router)), simple_command_handler_(*router_), - eval_command_handler_(*router_), mget_handler_(*router_), mset_handler_(*router_), - split_keys_sum_result_handler_(*router_), - stats_{ALL_COMMAND_SPLITTER_STATS(POOL_COUNTER_PREFIX(scope, stat_prefix + "splitter."))}, - time_source_(time_source) { +InstanceImpl::InstanceImpl(Router& router, Stats::Scope& scope, const std::string& stat_prefix, + TimeSource& time_source, bool latency_in_micros, + Common::Redis::FaultManager& fault_manager, + Event::Dispatcher& dispatcher) + : simple_command_handler_(router), eval_command_handler_(router), mget_handler_(router), + mset_handler_(router), + split_keys_sum_result_handler_(router), stats_{ALL_COMMAND_SPLITTER_STATS(POOL_COUNTER_PREFIX( + scope, stat_prefix + "splitter."))}, + time_source_(time_source), fault_manager_(fault_manager), dispatcher_(dispatcher) { for (const std::string& command : Common::Redis::SupportedCommands::simpleCommands()) { addHandler(scope, stat_prefix, command, latency_in_micros, simple_command_handler_); } @@ -468,6 +507,7 @@ SplitRequestPtr InstanceImpl::makeRequest(Common::Redis::RespValuePtr&& request, return nullptr; } + // Get the handler for the downstream request auto handler = handler_lookup_table_.find(to_lower_string.c_str()); if (handler == nullptr) { stats_.unsupported_command_.inc(); @@ -475,11 +515,46 @@ SplitRequestPtr InstanceImpl::makeRequest(Common::Redis::RespValuePtr&& request, fmt::format("unsupported command '{}'", request->asArray()[0].asString()))); return nullptr; } + + // Fault Injection Check + const Common::Redis::Fault* fault_ptr = fault_manager_.getFaultForCommand(to_lower_string); + + // Check if delay, which determines which callbacks to use. If a delay fault is enabled, + // the delay fault itself wraps the request (or other fault) and the delay fault itself + // implements the callbacks functions, and in turn calls the real callbacks after injecting + // delay on the result of the wrapped request or fault. + const bool has_delay_fault = + fault_ptr != nullptr && fault_ptr->delayMs() > std::chrono::milliseconds(0); + std::unique_ptr delay_fault_ptr; + if (has_delay_fault) { + delay_fault_ptr = DelayFaultRequest::create(callbacks, handler->command_stats_, time_source_, + dispatcher_, fault_ptr->delayMs()); + } + + // Note that the command_stats_ object of the original request is used for faults, so that our + // downstream metrics reflect any faults added (with special fault metrics) or extra latency from + // a delay. 2) we use a ternary operator for the callback parameter- we want to use the + // delay_fault as callback if there is a delay per the earlier comment. ENVOY_LOG(debug, "redis: splitting '{}'", request->toString()); handler->command_stats_.total_.inc(); - SplitRequestPtr request_ptr = handler->handler_.get().startRequest( - std::move(request), callbacks, handler->command_stats_, time_source_); - return request_ptr; + + SplitRequestPtr request_ptr; + if (fault_ptr != nullptr && fault_ptr->faultType() == Common::Redis::FaultType::Error) { + request_ptr = ErrorFaultRequest::create(has_delay_fault ? *delay_fault_ptr : callbacks, + handler->command_stats_, time_source_, has_delay_fault); + } else { + request_ptr = handler->handler_.get().startRequest( + std::move(request), has_delay_fault ? *delay_fault_ptr : callbacks, handler->command_stats_, + time_source_, has_delay_fault); + } + + // Complete delay, if any. The delay fault takes ownership of the wrapped request. + if (has_delay_fault) { + delay_fault_ptr->wrapped_request_ptr_ = std::move(request_ptr); + return delay_fault_ptr; + } else { + return request_ptr; + } } void InstanceImpl::onInvalidRequest(SplitCallbacks& callbacks) { @@ -505,6 +580,12 @@ void InstanceImpl::addHandler(Stats::Scope& scope, const std::string& stat_prefi handler})); } +CommandSplitterPtr CommandSplitterFactoryImpl::create(Event::Dispatcher& dispatcher) { + return std::make_unique(*router_, scope_, stat_prefix_, + time_source_, latency_in_micros_, + *fault_manager_, dispatcher); +} + } // namespace CommandSplitter } // namespace RedisProxy } // namespace NetworkFilters diff --git a/source/extensions/filters/network/redis_proxy/command_splitter_impl.h b/source/extensions/filters/network/redis_proxy/command_splitter_impl.h index 8057f9a91b2ca..813597dc0d2a7 100644 --- a/source/extensions/filters/network/redis_proxy/command_splitter_impl.h +++ b/source/extensions/filters/network/redis_proxy/command_splitter_impl.h @@ -11,10 +11,10 @@ #include "common/common/logger.h" #include "common/common/utility.h" -#include "common/singleton/const_singleton.h" #include "common/stats/timespan_impl.h" #include "extensions/filters/network/common/redis/client_impl.h" +#include "extensions/filters/network/common/redis/fault_impl.h" #include "extensions/filters/network/common/redis/utility.h" #include "extensions/filters/network/redis_proxy/command_splitter.h" #include "extensions/filters/network/redis_proxy/conn_pool_impl.h" @@ -43,7 +43,9 @@ using Response = ConstSingleton; #define ALL_COMMAND_STATS(COUNTER) \ COUNTER(total) \ COUNTER(success) \ - COUNTER(error) + COUNTER(error) \ + COUNTER(error_fault) \ + COUNTER(delay_fault) /** * Struct definition for all command stats. @see stats_macros.h @@ -59,7 +61,7 @@ class CommandHandler { virtual SplitRequestPtr startRequest(Common::Redis::RespValuePtr&& request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source) PURE; + TimeSource& time_source, bool delay_command_latency) PURE; }; class CommandHandlerBase { @@ -75,10 +77,14 @@ class SplitRequestBase : public SplitRequest { const Common::Redis::RespValue& request); void updateStats(const bool success); - SplitRequestBase(CommandStats& command_stats, TimeSource& time_source) + SplitRequestBase(CommandStats& command_stats, TimeSource& time_source, bool delay_command_latency) : command_stats_(command_stats) { - command_latency_ = std::make_unique( - command_stats_.latency_, time_source); + if (!delay_command_latency) { + command_latency_ = std::make_unique( + command_stats_.latency_, time_source); + } else { + command_latency_ = nullptr; + } } CommandStats& command_stats_; Stats::TimespanPtr command_latency_; @@ -94,14 +100,16 @@ class SingleServerRequest : public SplitRequestBase, public ConnPool::PoolCallba // ConnPool::PoolCallbacks void onResponse(Common::Redis::RespValuePtr&& response) override; void onFailure() override; + void onFailure(std::string error_msg); // RedisProxy::CommandSplitter::SplitRequest void cancel() override; protected: SingleServerRequest(SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source) - : SplitRequestBase(command_stats, time_source), callbacks_(callbacks) {} + TimeSource& time_source, bool delay_command_latency) + : SplitRequestBase(command_stats, time_source, delay_command_latency), callbacks_(callbacks) { + } SplitCallbacks& callbacks_; ConnPool::InstanceSharedPtr conn_pool_; @@ -109,6 +117,57 @@ class SingleServerRequest : public SplitRequestBase, public ConnPool::PoolCallba Common::Redis::RespValuePtr incoming_request_; }; +/** + * ErrorFaultRequest returns an error. + */ +class ErrorFaultRequest : public SingleServerRequest { +public: + static SplitRequestPtr create(SplitCallbacks& callbacks, CommandStats& command_stats, + TimeSource& time_source, bool has_delaydelay_command_latency_fault); + +private: + ErrorFaultRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source, + bool delay_command_latency) + : SingleServerRequest(callbacks, command_stats, time_source, delay_command_latency) {} +}; + +/** + * DelayFaultRequest wraps a request- either a normal request or a fault- and delays it. + */ +class DelayFaultRequest : public SplitRequestBase, public SplitCallbacks { +public: + static std::unique_ptr + create(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source, + Event::Dispatcher& dispatcher, std::chrono::milliseconds delay); + + DelayFaultRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source, + Event::Dispatcher& dispatcher, std::chrono::milliseconds delay) + : SplitRequestBase(command_stats, time_source, false), callbacks_(callbacks), delay_(delay) { + delay_timer_ = dispatcher.createTimer([this]() -> void { onDelayResponse(); }); + } + + // SplitCallbacks + bool connectionAllowed() override { return callbacks_.connectionAllowed(); } + void onAuth(const std::string& password) override { callbacks_.onAuth(password); } + void onAuth(const std::string& username, const std::string& password) override { + callbacks_.onAuth(username, password); + } + void onResponse(Common::Redis::RespValuePtr&& response) override; + + // RedisProxy::CommandSplitter::SplitRequest + void cancel() override; + + SplitRequestPtr wrapped_request_ptr_; + +private: + void onDelayResponse(); + + SplitCallbacks& callbacks_; + std::chrono::milliseconds delay_; + Event::TimerPtr delay_timer_; + Common::Redis::RespValuePtr response_; +}; + /** * SimpleRequest hashes the first argument as the key. */ @@ -116,11 +175,12 @@ class SimpleRequest : public SingleServerRequest { public: static SplitRequestPtr create(Router& router, Common::Redis::RespValuePtr&& incoming_request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source); + TimeSource& time_source, bool delay_command_latency); private: - SimpleRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source) - : SingleServerRequest(callbacks, command_stats, time_source) {} + SimpleRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source, + bool delay_command_latency) + : SingleServerRequest(callbacks, command_stats, time_source, delay_command_latency) {} }; /** @@ -130,11 +190,12 @@ class EvalRequest : public SingleServerRequest { public: static SplitRequestPtr create(Router& router, Common::Redis::RespValuePtr&& incoming_request, SplitCallbacks& callbacks, CommandStats& command_stats, - TimeSource& time_source); + TimeSource& time_source, bool delay_command_latency); private: - EvalRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source) - : SingleServerRequest(callbacks, command_stats, time_source) {} + EvalRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source, + bool delay_command_latency) + : SingleServerRequest(callbacks, command_stats, time_source, delay_command_latency) {} }; /** @@ -150,8 +211,10 @@ class FragmentedRequest : public SplitRequestBase { void cancel() override; protected: - FragmentedRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source) - : SplitRequestBase(command_stats, time_source), callbacks_(callbacks) {} + FragmentedRequest(SplitCallbacks& callbacks, CommandStats& command_stats, TimeSource& time_source, + bool delay_command_latency) + : SplitRequestBase(command_stats, time_source, delay_command_latency), callbacks_(callbacks) { + } struct PendingRequest : public ConnPool::PoolCallbacks { PendingRequest(FragmentedRequest& parent, uint32_t index) : parent_(parent), index_(index) {} @@ -186,11 +249,12 @@ class MGETRequest : public FragmentedRequest, Logger::Loggable { public: - InstanceImpl(RouterPtr&& router, Stats::Scope& scope, const std::string& stat_prefix, - TimeSource& time_source, bool latency_in_micros); + InstanceImpl(Router& router, Stats::Scope& scope, const std::string& stat_prefix, + TimeSource& time_source, bool latency_in_micros, + Common::Redis::FaultManager& fault_manager, Event::Dispatcher& dispatcher); // RedisProxy::CommandSplitter::Instance SplitRequestPtr makeRequest(Common::Redis::RespValuePtr&& request, @@ -287,7 +355,6 @@ class InstanceImpl : public Instance, Logger::Loggable { bool latency_in_micros, CommandHandler& handler); void onInvalidRequest(SplitCallbacks& callbacks); - RouterPtr router_; CommandHandlerFactory simple_command_handler_; CommandHandlerFactory eval_command_handler_; CommandHandlerFactory mget_handler_; @@ -296,6 +363,27 @@ class InstanceImpl : public Instance, Logger::Loggable { TrieLookupTable handler_lookup_table_; InstanceStats stats_; TimeSource& time_source_; + Common::Redis::FaultManager& fault_manager_; + Event::Dispatcher& dispatcher_; +}; + +class CommandSplitterFactoryImpl : public CommandSplitterFactory { +public: + CommandSplitterFactoryImpl(RouterPtr&& router, Common::Redis::FaultManagerPtr fault_manager, + Stats::Scope& scope, const std::string& stat_prefix, + TimeSource& time_source, bool latency_in_micros) + : router_(std::move(router)), fault_manager_(std::move(fault_manager)), scope_(scope), + stat_prefix_(stat_prefix), time_source_(time_source), + latency_in_micros_(latency_in_micros){}; + CommandSplitterPtr create(Event::Dispatcher& dispatcher) override; + +private: + RouterPtr router_; + Common::Redis::FaultManagerPtr fault_manager_; + Stats::Scope& scope_; + const std::string& stat_prefix_; + TimeSource& time_source_; + bool latency_in_micros_; }; } // namespace CommandSplitter diff --git a/source/extensions/filters/network/redis_proxy/config.cc b/source/extensions/filters/network/redis_proxy/config.cc index 2d62f511b393e..c75e61b2fcb08 100644 --- a/source/extensions/filters/network/redis_proxy/config.cc +++ b/source/extensions/filters/network/redis_proxy/config.cc @@ -5,6 +5,7 @@ #include "extensions/common/redis/cluster_refresh_manager_impl.h" #include "extensions/filters/network/common/redis/client_impl.h" +#include "extensions/filters/network/common/redis/fault_impl.h" #include "extensions/filters/network/redis_proxy/command_splitter_impl.h" #include "extensions/filters/network/redis_proxy/proxy_filter.h" #include "extensions/filters/network/redis_proxy/router_impl.h" @@ -86,14 +87,18 @@ Network::FilterFactoryCb RedisProxyFilterConfigFactory::createFilterFactoryFromP auto router = std::make_unique(prefix_routes, std::move(upstreams), context.runtime()); - std::shared_ptr splitter = - std::make_shared( - std::move(router), context.scope(), filter_config->stat_prefix_, context.timeSource(), - proto_config.latency_in_micros()); - return [splitter, filter_config](Network::FilterManager& filter_manager) -> void { + auto fault_manager = std::make_unique( + context.random(), context.runtime(), proto_config.faults()); + + auto splitter_factory = std::make_shared( + std::move(router), std::move(fault_manager), context.scope(), filter_config->stat_prefix_, + context.timeSource(), proto_config.latency_in_micros()); + + return [splitter_factory, refresh_manager, + filter_config](Network::FilterManager& filter_manager) -> void { Common::Redis::DecoderFactoryImpl factory; filter_manager.addReadFilter(std::make_shared( - factory, Common::Redis::EncoderPtr{new Common::Redis::EncoderImpl()}, *splitter, + factory, Common::Redis::EncoderPtr{new Common::Redis::EncoderImpl()}, *splitter_factory, filter_config)); }; } diff --git a/source/extensions/filters/network/redis_proxy/proxy_filter.cc b/source/extensions/filters/network/redis_proxy/proxy_filter.cc index aa2f558cc51af..bbbadb29de5db 100644 --- a/source/extensions/filters/network/redis_proxy/proxy_filter.cc +++ b/source/extensions/filters/network/redis_proxy/proxy_filter.cc @@ -34,10 +34,11 @@ ProxyStats ProxyFilterConfig::generateStats(const std::string& prefix, Stats::Sc } ProxyFilter::ProxyFilter(Common::Redis::DecoderFactory& factory, - Common::Redis::EncoderPtr&& encoder, CommandSplitter::Instance& splitter, + Common::Redis::EncoderPtr&& encoder, + CommandSplitter::CommandSplitterFactory& splitter_factory, ProxyFilterConfigSharedPtr config) - : decoder_(factory.create(*this)), encoder_(std::move(encoder)), splitter_(splitter), - config_(config) { + : decoder_(factory.create(*this)), encoder_(std::move(encoder)), + splitter_factory_(splitter_factory), config_(config) { config_->stats_.downstream_cx_total_.inc(); config_->stats_.downstream_cx_active_.inc(); connection_allowed_ = @@ -57,12 +58,13 @@ void ProxyFilter::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& ca config_->stats_.downstream_cx_tx_bytes_total_, config_->stats_.downstream_cx_tx_bytes_buffered_, nullptr, nullptr}); + splitter_ = splitter_factory_.create(callbacks_->connection().dispatcher()); } void ProxyFilter::onRespValue(Common::Redis::RespValuePtr&& value) { pending_requests_.emplace_back(*this); PendingRequest& request = pending_requests_.back(); - CommandSplitter::SplitRequestPtr split = splitter_.makeRequest(std::move(value), request); + CommandSplitter::SplitRequestPtr split = splitter_->makeRequest(std::move(value), request); if (split) { // The splitter can immediately respond and destroy the pending request. Only store the handle // if the request is still alive. diff --git a/source/extensions/filters/network/redis_proxy/proxy_filter.h b/source/extensions/filters/network/redis_proxy/proxy_filter.h index 1694a2a0640e9..4b7877bc00d79 100644 --- a/source/extensions/filters/network/redis_proxy/proxy_filter.h +++ b/source/extensions/filters/network/redis_proxy/proxy_filter.h @@ -75,7 +75,8 @@ class ProxyFilter : public Network::ReadFilter, public Network::ConnectionCallbacks { public: ProxyFilter(Common::Redis::DecoderFactory& factory, Common::Redis::EncoderPtr&& encoder, - CommandSplitter::Instance& splitter, ProxyFilterConfigSharedPtr config); + CommandSplitter::CommandSplitterFactory& splitter_factory, + ProxyFilterConfigSharedPtr config); ~ProxyFilter() override; // Network::ReadFilter @@ -94,6 +95,8 @@ class ProxyFilter : public Network::ReadFilter, bool connectionAllowed() { return connection_allowed_; } private: + friend class RedisProxyFilterTest; + struct PendingRequest : public CommandSplitter::SplitCallbacks { PendingRequest(ProxyFilter& parent); ~PendingRequest() override; @@ -119,7 +122,8 @@ class ProxyFilter : public Network::ReadFilter, Common::Redis::DecoderPtr decoder_; Common::Redis::EncoderPtr encoder_; - CommandSplitter::Instance& splitter_; + CommandSplitter::CommandSplitterFactory& splitter_factory_; + CommandSplitter::CommandSplitterPtr splitter_; ProxyFilterConfigSharedPtr config_; Buffer::OwnedImpl encoder_buffer_; Network::ReadFilterCallbacks* callbacks_{}; diff --git a/test/extensions/filters/network/common/redis/BUILD b/test/extensions/filters/network/common/redis/BUILD index dffc239544886..e8e445c8b6081 100644 --- a/test/extensions/filters/network/common/redis/BUILD +++ b/test/extensions/filters/network/common/redis/BUILD @@ -60,3 +60,16 @@ envoy_cc_test( "@envoy_api//envoy/extensions/filters/network/redis_proxy/v3:pkg_cc_proto", ], ) + +envoy_cc_test( + name = "fault_test", + srcs = ["fault_test.cc"], + deps = [ + ":redis_mocks", + "//source/common/common:assert_lib", + "//source/extensions/filters/network/common/redis:fault_lib", + "//test/mocks/runtime:runtime_mocks", + "//test/test_common:test_runtime_lib", + "//test/test_common:utility_lib", + ], +) diff --git a/test/extensions/filters/network/common/redis/fault_test.cc b/test/extensions/filters/network/common/redis/fault_test.cc new file mode 100644 index 0000000000000..a80caf5c2d2a3 --- /dev/null +++ b/test/extensions/filters/network/common/redis/fault_test.cc @@ -0,0 +1,206 @@ +#include "envoy/common/random_generator.h" + +#include "common/common/assert.h" + +#include "extensions/filters/network/common/redis/fault_impl.h" + +#include "test/extensions/filters/network/common/redis/mocks.h" +#include "test/mocks/runtime/mocks.h" +#include "test/test_common/printers.h" +#include "test/test_common/test_runtime.h" +#include "test/test_common/utility.h" + +#include "gtest/gtest.h" + +using testing::Return; + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Common { +namespace Redis { + +using RedisProxy = envoy::extensions::filters::network::redis_proxy::v3::RedisProxy; +using FractionalPercent = envoy::type::v3::FractionalPercent; +class FaultTest : public testing::Test { +public: + const std::string RUNTIME_KEY = "runtime_key"; + + void + createCommandFault(RedisProxy::RedisFault* fault, std::string command_str, int delay_seconds, + absl::optional fault_percentage, + absl::optional denominator, + absl::optional runtime_key) { + // We don't set fault type as it isn't used in the test + + auto* commands = fault->mutable_commands(); + auto* command = commands->Add(); + command->assign(command_str); + + fault->set_fault_type(envoy::extensions::filters::network::redis_proxy::v3:: + RedisProxy_RedisFault_RedisFaultType_ERROR); + + addFaultPercentage(fault, fault_percentage, denominator, runtime_key); + addDelay(fault, delay_seconds); + } + + void + createAllKeyFault(RedisProxy::RedisFault* fault, int delay_seconds, + absl::optional fault_percentage, + absl::optional denominator, + absl::optional runtime_key) { + addFaultPercentage(fault, fault_percentage, denominator, runtime_key); + addDelay(fault, delay_seconds); + } + + void + addFaultPercentage(RedisProxy::RedisFault* fault, absl::optional fault_percentage, + absl::optional denominator, + absl::optional runtime_key) { + envoy::config::core::v3::RuntimeFractionalPercent* fault_enabled = + fault->mutable_fault_enabled(); + + if (runtime_key.has_value()) { + fault_enabled->set_runtime_key(runtime_key.value()); + } + auto* percentage = fault_enabled->mutable_default_value(); + if (fault_percentage.has_value()) { + percentage->set_numerator(fault_percentage.value()); + } + if (denominator.has_value()) { + percentage->set_denominator(denominator.value()); + } + } + + void addDelay(RedisProxy::RedisFault* fault, int delay_seconds) { + std::chrono::seconds duration = std::chrono::seconds(delay_seconds); + fault->mutable_delay()->set_seconds(duration.count()); + } + + testing::NiceMock random_; + testing::NiceMock runtime_; +}; + +TEST_F(FaultTest, MakeFaultForTestHelper) { + Common::Redis::FaultSharedPtr fault_ptr = + FaultManagerImpl::makeFaultForTest(FaultType::Error, std::chrono::milliseconds(10)); + + ASSERT_TRUE(fault_ptr->faultType() == FaultType::Error); + ASSERT_TRUE(fault_ptr->delayMs() == std::chrono::milliseconds(10)); +} + +TEST_F(FaultTest, NoFaults) { + RedisProxy redis_config; + auto* faults = redis_config.mutable_faults(); + + TestScopedRuntime scoped_runtime; + FaultManagerImpl fault_manager = FaultManagerImpl(random_, runtime_, *faults); + + const Fault* fault_ptr = fault_manager.getFaultForCommand("get"); + ASSERT_TRUE(fault_ptr == nullptr); +} + +TEST_F(FaultTest, SingleCommandFaultNotEnabled) { + RedisProxy redis_config; + auto* faults = redis_config.mutable_faults(); + createCommandFault(faults->Add(), "get", 0, 0, FractionalPercent::HUNDRED, RUNTIME_KEY); + + TestScopedRuntime scoped_runtime; + FaultManagerImpl fault_manager = FaultManagerImpl(random_, runtime_, *faults); + + EXPECT_CALL(random_, random()).WillOnce(Return(0)); + EXPECT_CALL(runtime_, snapshot()); + const Fault* fault_ptr = fault_manager.getFaultForCommand("get"); + ASSERT_TRUE(fault_ptr == nullptr); +} + +TEST_F(FaultTest, SingleCommandFault) { + // Inject a single fault. Notably we use a different denominator to test that code path; normally + // we use FractionalPercent::HUNDRED. + RedisProxy redis_config; + auto* faults = redis_config.mutable_faults(); + createCommandFault(faults->Add(), "ttl", 0, 5000, FractionalPercent::TEN_THOUSAND, RUNTIME_KEY); + + TestScopedRuntime scoped_runtime; + FaultManagerImpl fault_manager = FaultManagerImpl(random_, runtime_, *faults); + + EXPECT_CALL(random_, random()).WillOnce(Return(1)); + EXPECT_CALL(runtime_.snapshot_, getInteger(RUNTIME_KEY, 50)).WillOnce(Return(10)); + + const Fault* fault_ptr = fault_manager.getFaultForCommand("ttl"); + ASSERT_TRUE(fault_ptr != nullptr); +} + +TEST_F(FaultTest, SingleCommandFaultWithNoDefaultValueOrRuntimeValue) { + // Inject a single fault with no default value or runtime value. + RedisProxy redis_config; + auto* faults = redis_config.mutable_faults(); + createCommandFault(faults->Add(), "ttl", 0, absl::nullopt, absl::nullopt, absl::nullopt); + + TestScopedRuntime scoped_runtime; + FaultManagerImpl fault_manager = FaultManagerImpl(random_, runtime_, *faults); + + EXPECT_CALL(random_, random()).WillOnce(Return(1)); + const Fault* fault_ptr = fault_manager.getFaultForCommand("ttl"); + ASSERT_TRUE(fault_ptr == nullptr); +} + +TEST_F(FaultTest, MultipleFaults) { + // This creates 2 faults, but the map will have 3 entries, as each command points to + // command specific faults AND the general fault. The second fault has no runtime key, + // forcing the runtime key check to be false in application code and falling back to the + // default value. + RedisProxy redis_config; + auto* faults = redis_config.mutable_faults(); + createCommandFault(faults->Add(), "get", 0, 25, FractionalPercent::HUNDRED, RUNTIME_KEY); + createAllKeyFault(faults->Add(), 2, 25, FractionalPercent::HUNDRED, absl::nullopt); + + TestScopedRuntime scoped_runtime; + FaultManagerImpl fault_manager = FaultManagerImpl(random_, runtime_, *faults); + const Fault* fault_ptr; + + // Get command - should have a fault 50% of time + // For the first call we mock the random percentage to be 10%, which will give us the first fault + // with 0s delay. + EXPECT_CALL(random_, random()).WillOnce(Return(1)); + EXPECT_CALL(runtime_.snapshot_, getInteger(_, 25)).WillOnce(Return(10)); + fault_ptr = fault_manager.getFaultForCommand("get"); + ASSERT_TRUE(fault_ptr != nullptr); + ASSERT_EQ(fault_ptr->delayMs(), std::chrono::milliseconds(0)); + + // Another Get; we mock the random percentage to be 25%, giving us the ALL_KEY fault + EXPECT_CALL(random_, random()).WillOnce(Return(25)); + EXPECT_CALL(runtime_.snapshot_, getInteger(_, _)) + .Times(2) + .WillOnce(Return(10)) + .WillOnce(Return(50)); + fault_ptr = fault_manager.getFaultForCommand("get"); + ASSERT_TRUE(fault_ptr != nullptr); + ASSERT_EQ(fault_ptr->delayMs(), std::chrono::milliseconds(2000)); + + // No fault for Get command with mocked random percentage >= 50%. + EXPECT_CALL(random_, random()).WillOnce(Return(50)); + EXPECT_CALL(runtime_.snapshot_, getInteger(_, _)).Times(2); + fault_ptr = fault_manager.getFaultForCommand("get"); + ASSERT_TRUE(fault_ptr == nullptr); + + // Any other command; we mock the random percentage to be 1%, giving us the ALL_KEY fault + EXPECT_CALL(random_, random()).WillOnce(Return(1)); + EXPECT_CALL(runtime_.snapshot_, getInteger(_, _)).WillOnce(Return(10)); + + fault_ptr = fault_manager.getFaultForCommand("ttl"); + ASSERT_TRUE(fault_ptr != nullptr); + ASSERT_EQ(fault_ptr->delayMs(), std::chrono::milliseconds(2000)); + + // No fault for any other command with mocked random percentage >= 25%. + EXPECT_CALL(random_, random()).WillOnce(Return(25)); + EXPECT_CALL(runtime_.snapshot_, getInteger(_, _)); + fault_ptr = fault_manager.getFaultForCommand("ttl"); + ASSERT_TRUE(fault_ptr == nullptr); +} + +} // namespace Redis +} // namespace Common +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/test/extensions/filters/network/redis_proxy/BUILD b/test/extensions/filters/network/redis_proxy/BUILD index 3f83e1eeda4b1..24d847f5306f6 100644 --- a/test/extensions/filters/network/redis_proxy/BUILD +++ b/test/extensions/filters/network/redis_proxy/BUILD @@ -18,16 +18,22 @@ envoy_extension_cc_test( name = "command_splitter_impl_test", srcs = ["command_splitter_impl_test.cc"], extension_name = "envoy.filters.network.redis_proxy", + # This test takes a while to run specially under tsan. + # Shard it to avoid test timeout. + shard_count = 2, deps = [ ":redis_mocks", "//source/common/stats:isolated_store_lib", "//source/common/stats:stats_lib", + "//source/extensions/filters/network/common/redis:fault_lib", "//source/extensions/filters/network/redis_proxy:command_splitter_lib", "//source/extensions/filters/network/redis_proxy:router_interface", "//test/extensions/filters/network/common/redis:redis_mocks", "//test/mocks:common_lib", + "//test/mocks/event:event_mocks", "//test/mocks/stats:stats_mocks", "//test/test_common:simulated_time_system_lib", + "//test/test_common:test_runtime_lib", ], ) @@ -85,6 +91,7 @@ envoy_cc_mock( "//source/extensions/common/redis:cluster_refresh_manager_interface", "//source/extensions/filters/network/common/redis:client_interface", "//source/extensions/filters/network/common/redis:codec_lib", + "//source/extensions/filters/network/common/redis:fault_interface", "//source/extensions/filters/network/redis_proxy:command_splitter_interface", "//source/extensions/filters/network/redis_proxy:conn_pool_interface", "//source/extensions/filters/network/redis_proxy:router_interface", @@ -116,6 +123,7 @@ envoy_extension_cc_benchmark_binary( "//source/common/stats:isolated_store_lib", "//source/common/stats:stats_lib", "//source/extensions/filters/network/redis_proxy:command_splitter_lib", + "//test/mocks/event:event_mocks", "//test/test_common:printers_lib", "//test/test_common:simulated_time_system_lib", ], @@ -148,6 +156,7 @@ envoy_extension_cc_test( extension_name = "envoy.filters.network.redis_proxy", tags = ["fails_on_windows"], deps = [ + "//source/extensions/filters/network/common/redis:fault_lib", "//source/extensions/filters/network/redis_proxy:config", "//test/integration:integration_lib", ], diff --git a/test/extensions/filters/network/redis_proxy/command_lookup_speed_test.cc b/test/extensions/filters/network/redis_proxy/command_lookup_speed_test.cc index edf29c9730921..d7ddc451bcb41 100644 --- a/test/extensions/filters/network/redis_proxy/command_lookup_speed_test.cc +++ b/test/extensions/filters/network/redis_proxy/command_lookup_speed_test.cc @@ -12,10 +12,14 @@ #include "extensions/filters/network/common/redis/supported_commands.h" #include "extensions/filters/network/redis_proxy/command_splitter_impl.h" +#include "test/extensions/filters/network/redis_proxy/mocks.h" +#include "test/mocks/event/mocks.h" #include "test/test_common/simulated_time_system.h" #include "benchmark/benchmark.h" +using testing::NiceMock; + namespace Envoy { namespace Extensions { namespace NetworkFilters { @@ -64,11 +68,13 @@ class CommandLookUpSpeedTest { } } - Router* router_{new NullRouterImpl()}; + RouterPtr router_{std::make_unique()}; Stats::IsolatedStoreImpl store_; Event::SimulatedTimeSystem time_system_; - CommandSplitter::InstanceImpl splitter_{RouterPtr{router_}, store_, "redis.foo.", time_system_, - false}; + NiceMock fault_manager_; + NiceMock dispatcher_; + CommandSplitter::InstanceImpl splitter_{*router_, store_, "redis.foo.", time_system_, + false, fault_manager_, dispatcher_}; NoOpSplitCallbacks callbacks_; CommandSplitter::SplitRequestPtr handle_; }; diff --git a/test/extensions/filters/network/redis_proxy/command_splitter_impl_test.cc b/test/extensions/filters/network/redis_proxy/command_splitter_impl_test.cc index 097cb3d49f4c2..e864cb29e65c9 100644 --- a/test/extensions/filters/network/redis_proxy/command_splitter_impl_test.cc +++ b/test/extensions/filters/network/redis_proxy/command_splitter_impl_test.cc @@ -6,12 +6,14 @@ #include "common/common/fmt.h" #include "common/stats/isolated_store_impl.h" +#include "extensions/filters/network/common/redis/fault_impl.h" #include "extensions/filters/network/common/redis/supported_commands.h" #include "extensions/filters/network/redis_proxy/command_splitter_impl.h" #include "test/extensions/filters/network/common/redis/mocks.h" #include "test/extensions/filters/network/redis_proxy/mocks.h" #include "test/mocks/common.h" +#include "test/mocks/event/mocks.h" #include "test/mocks/stats/mocks.h" #include "test/test_common/simulated_time_system.h" @@ -32,7 +34,12 @@ namespace CommandSplitter { class RedisCommandSplitterImplTest : public testing::Test { public: RedisCommandSplitterImplTest() : RedisCommandSplitterImplTest(false) {} - RedisCommandSplitterImplTest(bool latency_in_macro) : latency_in_micros_(latency_in_macro) {} + RedisCommandSplitterImplTest(bool latency_in_macro) + : RedisCommandSplitterImplTest(latency_in_macro, nullptr) {} + RedisCommandSplitterImplTest(bool latency_in_macro, Common::Redis::FaultSharedPtr fault_ptr) + : latency_in_micros_(latency_in_macro) { + ON_CALL(fault_manager_, getFaultForCommand(_)).WillByDefault(Return(fault_ptr.get())); + } void makeBulkStringArray(Common::Redis::RespValue& value, const std::vector& strings) { std::vector values(strings.size()); @@ -52,14 +59,18 @@ class RedisCommandSplitterImplTest : public testing::Test { const bool latency_in_micros_; ConnPool::MockInstance* conn_pool_{new ConnPool::MockInstance()}; + ConnPool::InstanceSharedPtr conn_pool_shared_ptr_{conn_pool_}; ConnPool::MockInstance* mirror_conn_pool_{new ConnPool::MockInstance()}; ConnPool::InstanceSharedPtr mirror_conn_pool_shared_ptr_{mirror_conn_pool_}; - std::shared_ptr> route_{ - new NiceMock(ConnPool::InstanceSharedPtr{conn_pool_})}; + std::shared_ptr> route_{new NiceMock(conn_pool_shared_ptr_)}; + std::shared_ptr> router_{new NiceMock(route_)}; NiceMock store_; + NiceMock dispatcher_; + NiceMock fault_manager_; + Event::SimulatedTimeSystem time_system_; - InstanceImpl splitter_{std::make_unique>(route_), store_, "redis.foo.", - time_system_, latency_in_micros_}; + InstanceImpl splitter_{*router_, store_, "redis.foo.", time_system_, + latency_in_micros_, fault_manager_, dispatcher_}; MockSplitCallbacks callbacks_; SplitRequestPtr handle_; }; @@ -993,6 +1004,147 @@ INSTANTIATE_TEST_SUITE_P(RedisSingleServerRequestWithLatencyMicrosTest, RedisSingleServerRequestWithLatencyMicrosTest, testing::ValuesIn(Common::Redis::SupportedCommands::simpleCommands())); +// In subclasses of fault test, we mock the expected faults in the constructor, as the +// fault manager is owned by the splitter, which is also generated later in construction +// of the base test class. +class RedisSingleServerRequestWithFaultTest : public RedisSingleServerRequestTest { +public: + NiceMock* timer_; + Event::TimerCb timer_cb_; + int delay_ms_; + Common::Redis::FaultSharedPtr fault_ptr_; +}; + +class RedisSingleServerRequestWithErrorFaultTest : public RedisSingleServerRequestWithFaultTest { +public: + RedisSingleServerRequestWithErrorFaultTest() { + delay_ms_ = 0; + fault_ptr_ = Common::Redis::FaultManagerImpl::makeFaultForTest( + Common::Redis::FaultType::Error, std::chrono::milliseconds(delay_ms_)); + ON_CALL(fault_manager_, getFaultForCommand(_)).WillByDefault(Return(fault_ptr_.get())); + } +}; + +TEST_P(RedisSingleServerRequestWithErrorFaultTest, Fault) { + InSequence s; + + std::string lower_command = absl::AsciiStrToLower(GetParam()); + Common::Redis::RespValuePtr request{new Common::Redis::RespValue()}; + makeBulkStringArray(*request, {GetParam(), "hello"}); + + EXPECT_CALL(callbacks_, connectionAllowed()).WillOnce(Return(true)); + EXPECT_CALL(callbacks_, onResponse_(_)); + handle_ = splitter_.makeRequest(std::move(request), callbacks_); + EXPECT_EQ(nullptr, handle_); + + EXPECT_EQ(1UL, store_.counter(fmt::format("redis.foo.command.{}.total", lower_command)).value()); + EXPECT_EQ(1UL, store_.counter(fmt::format("redis.foo.command.{}.error", lower_command)).value()); + EXPECT_EQ(1UL, + store_.counter(fmt::format("redis.foo.command.{}.error_fault", lower_command)).value()); +}; + +class RedisSingleServerRequestWithErrorWithDelayFaultTest + : public RedisSingleServerRequestWithFaultTest { +public: + RedisSingleServerRequestWithErrorWithDelayFaultTest() { + delay_ms_ = 5; + fault_ptr_ = Common::Redis::FaultManagerImpl::makeFaultForTest( + Common::Redis::FaultType::Error, std::chrono::milliseconds(delay_ms_)); + ON_CALL(fault_manager_, getFaultForCommand(_)).WillByDefault(Return(fault_ptr_.get())); + timer_ = new NiceMock(); + } +}; + +INSTANTIATE_TEST_SUITE_P(RedisSingleServerRequestWithErrorFaultTest, + RedisSingleServerRequestWithErrorFaultTest, + testing::ValuesIn(Common::Redis::SupportedCommands::simpleCommands())); + +TEST_P(RedisSingleServerRequestWithErrorWithDelayFaultTest, Fault) { + InSequence s; + + std::string lower_command = absl::AsciiStrToLower(GetParam()); + Common::Redis::RespValuePtr request{new Common::Redis::RespValue()}; + makeBulkStringArray(*request, {GetParam(), "hello"}); + + // As error faults have zero latency, recorded latency is equal to the delay. + EXPECT_CALL(callbacks_, connectionAllowed()).WillOnce(Return(true)); + EXPECT_CALL(dispatcher_, createTimer_(_)).WillOnce(Invoke([this](Event::TimerCb timer_cb) { + timer_cb_ = timer_cb; + return timer_; + })); + + handle_ = splitter_.makeRequest(std::move(request), callbacks_); + EXPECT_NE(nullptr, handle_); + time_system_.setMonotonicTime(std::chrono::milliseconds(delay_ms_)); + EXPECT_CALL(store_, deliverHistogramToSinks( + Property(&Stats::Metric::name, + fmt::format("redis.foo.command.{}.latency", lower_command)), + delay_ms_)); + EXPECT_CALL(callbacks_, onResponse_(_)); + timer_cb_(); + + EXPECT_EQ(1UL, store_.counter(fmt::format("redis.foo.command.{}.total", lower_command)).value()); + EXPECT_EQ(1UL, store_.counter(fmt::format("redis.foo.command.{}.error", lower_command)).value()); + EXPECT_EQ(1UL, + store_.counter(fmt::format("redis.foo.command.{}.error_fault", lower_command)).value()); +}; + +INSTANTIATE_TEST_SUITE_P(RedisSingleServerRequestWithErrorWithDelayFaultTest, + RedisSingleServerRequestWithErrorWithDelayFaultTest, + testing::ValuesIn(Common::Redis::SupportedCommands::simpleCommands())); + +class RedisSingleServerRequestWithDelayFaultTest : public RedisSingleServerRequestWithFaultTest { +public: + RedisSingleServerRequestWithDelayFaultTest() { + delay_ms_ = 15; + fault_ptr_ = Common::Redis::FaultManagerImpl::makeFaultForTest( + Common::Redis::FaultType::Delay, std::chrono::milliseconds(delay_ms_)); + ON_CALL(fault_manager_, getFaultForCommand(_)).WillByDefault(Return(fault_ptr_.get())); + timer_ = new NiceMock(); + } +}; + +TEST_P(RedisSingleServerRequestWithDelayFaultTest, Fault) { + InSequence s; + + std::string lower_command = absl::AsciiStrToLower(GetParam()); + std::string hash_key = "hello"; + + Common::Redis::RespValuePtr request{new Common::Redis::RespValue()}; + makeBulkStringArray(*request, {GetParam(), "hello"}); + + EXPECT_CALL(callbacks_, connectionAllowed()).WillOnce(Return(true)); + EXPECT_CALL(dispatcher_, createTimer_(_)).WillOnce(Invoke([this](Event::TimerCb timer_cb) { + timer_cb_ = timer_cb; + return timer_; + })); + EXPECT_CALL(*conn_pool_, makeRequest_(hash_key, RespVariantEq(*request), _)) + .WillOnce(DoAll(WithArg<2>(SaveArgAddress(&pool_callbacks_)), Return(&pool_request_))); + + handle_ = splitter_.makeRequest(std::move(request), callbacks_); + + EXPECT_NE(nullptr, handle_); + + EXPECT_CALL(store_, deliverHistogramToSinks( + Property(&Stats::Metric::name, + fmt::format("redis.foo.command.{}.latency", lower_command)), + delay_ms_)); + respond(); + + time_system_.setMonotonicTime(std::chrono::milliseconds(delay_ms_)); + timer_cb_(); + + EXPECT_EQ(1UL, store_.counter(fmt::format("redis.foo.command.{}.total", lower_command)).value()); + EXPECT_EQ(1UL, + store_.counter(fmt::format("redis.foo.command.{}.success", lower_command)).value()); + EXPECT_EQ(1UL, + store_.counter(fmt::format("redis.foo.command.{}.delay_fault", lower_command)).value()); +}; + +INSTANTIATE_TEST_SUITE_P(RedisSingleServerRequestWithDelayFaultTest, + RedisSingleServerRequestWithDelayFaultTest, + testing::ValuesIn(Common::Redis::SupportedCommands::simpleCommands())); + } // namespace CommandSplitter } // namespace RedisProxy } // namespace NetworkFilters diff --git a/test/extensions/filters/network/redis_proxy/config_test.cc b/test/extensions/filters/network/redis_proxy/config_test.cc index a9043af8cd6eb..155b726892841 100644 --- a/test/extensions/filters/network/redis_proxy/config_test.cc +++ b/test/extensions/filters/network/redis_proxy/config_test.cc @@ -169,6 +169,43 @@ stat_prefix: foo cb(connection); } +TEST(RedisProxyFilterConfigFactoryTest, RedisProxyFaultProto) { + const std::string yaml = R"EOF( +prefix_routes: + catch_all_route: + cluster: fake_cluster +stat_prefix: foo +faults: +- fault_type: ERROR + fault_enabled: + default_value: + numerator: 30 + denominator: HUNDRED + runtime_key: "bogus_key" + commands: + - GET +- fault_type: DELAY + fault_enabled: + default_value: + numerator: 20 + denominator: HUNDRED + runtime_key: "bogus_key" + delay: 2s +settings: + op_timeout: 0.02s + )EOF"; + + envoy::extensions::filters::network::redis_proxy::v3::RedisProxy proto_config{}; + TestUtility::loadFromYamlAndValidate(yaml, proto_config); + NiceMock context; + RedisProxyFilterConfigFactory factory; + Network::FilterFactoryCb cb = factory.createFilterFactoryFromProto(proto_config, context); + EXPECT_TRUE(factory.isTerminalFilter()); + Network::MockConnection connection; + EXPECT_CALL(connection, addReadFilter(_)); + cb(connection); +} + // Test that the deprecated extension name still functions. TEST(RedisProxyFilterConfigFactoryTest, DEPRECATED_FEATURE_TEST(DeprecatedExtensionFilterName)) { const std::string deprecated_name = "envoy.redis_proxy"; diff --git a/test/extensions/filters/network/redis_proxy/mocks.cc b/test/extensions/filters/network/redis_proxy/mocks.cc index d51809ba27f69..fa1ff637af615 100644 --- a/test/extensions/filters/network/redis_proxy/mocks.cc +++ b/test/extensions/filters/network/redis_proxy/mocks.cc @@ -26,6 +26,10 @@ MockMirrorPolicy::MockMirrorPolicy(ConnPool::InstanceSharedPtr conn_pool) ON_CALL(*this, shouldMirror(_)).WillByDefault(Return(true)); } +MockFaultManager::MockFaultManager() = default; +MockFaultManager::MockFaultManager(const MockFaultManager&) {} +MockFaultManager::~MockFaultManager() = default; + namespace ConnPool { MockPoolCallbacks::MockPoolCallbacks() = default; @@ -47,6 +51,9 @@ MockSplitCallbacks::~MockSplitCallbacks() = default; MockInstance::MockInstance() = default; MockInstance::~MockInstance() = default; +MockCommandSplitterFactory::MockCommandSplitterFactory() = default; +MockCommandSplitterFactory::~MockCommandSplitterFactory() = default; + } // namespace CommandSplitter } // namespace RedisProxy } // namespace NetworkFilters diff --git a/test/extensions/filters/network/redis_proxy/mocks.h b/test/extensions/filters/network/redis_proxy/mocks.h index b093ad35b9b94..d76a1f2598b71 100644 --- a/test/extensions/filters/network/redis_proxy/mocks.h +++ b/test/extensions/filters/network/redis_proxy/mocks.h @@ -7,6 +7,7 @@ #include "extensions/common/redis/cluster_refresh_manager.h" #include "extensions/filters/network/common/redis/client.h" #include "extensions/filters/network/common/redis/codec_impl.h" +#include "extensions/filters/network/common/redis/fault.h" #include "extensions/filters/network/redis_proxy/command_splitter.h" #include "extensions/filters/network/redis_proxy/conn_pool.h" #include "extensions/filters/network/redis_proxy/router.h" @@ -51,6 +52,15 @@ class MockMirrorPolicy : public MirrorPolicy { ConnPool::InstanceSharedPtr conn_pool_; }; +class MockFaultManager : public Common::Redis::FaultManager { +public: + MockFaultManager(); + MockFaultManager(const MockFaultManager& other); + ~MockFaultManager() override; + + MOCK_METHOD(const Common::Redis::Fault*, getFaultForCommand, (const std::string&), (const)); +}; + namespace ConnPool { class MockPoolCallbacks : public PoolCallbacks { @@ -119,6 +129,16 @@ class MockInstance : public Instance { (const Common::Redis::RespValue& request, SplitCallbacks& callbacks)); }; +class MockCommandSplitterFactory : public CommandSplitterFactory { +public: + MockCommandSplitterFactory(); + ~MockCommandSplitterFactory() override; + + CommandSplitterPtr create(Event::Dispatcher& dispatcher) override { return create_(dispatcher); }; + + MOCK_METHOD(CommandSplitterPtr, create_, (Event::Dispatcher & dispatcher)); +}; + } // namespace CommandSplitter } // namespace RedisProxy } // namespace NetworkFilters diff --git a/test/extensions/filters/network/redis_proxy/proxy_filter_test.cc b/test/extensions/filters/network/redis_proxy/proxy_filter_test.cc index f094c02b665a0..66def36798c08 100644 --- a/test/extensions/filters/network/redis_proxy/proxy_filter_test.cc +++ b/test/extensions/filters/network/redis_proxy/proxy_filter_test.cc @@ -19,6 +19,7 @@ #include "gtest/gtest.h" using testing::_; +using testing::ByMove; using testing::ByRef; using testing::DoAll; using testing::Eq; @@ -131,8 +132,12 @@ class RedisProxyFilterTest : public testing::Test, public Common::Redis::Decoder parseProtoFromYaml(yaml_string); config_ = std::make_shared(proto_config, store_, drain_decision_, runtime_, api_); - filter_ = std::make_unique(*this, Common::Redis::EncoderPtr{encoder_}, splitter_, - config_); + + std::unique_ptr splitter_ptr = + std::make_unique(); + EXPECT_CALL(splitter_factory_, create_(_)).WillOnce(Return(ByMove(std::move(splitter_ptr)))); + filter_ = std::make_unique(*this, Common::Redis::EncoderPtr{encoder_}, + splitter_factory_, config_); filter_->initializeReadFilterCallbacks(filter_callbacks_); EXPECT_EQ(Network::FilterStatus::Continue, filter_->onNewConnection()); EXPECT_EQ(1UL, config_->stats_.downstream_cx_total_.value()); @@ -152,6 +157,10 @@ class RedisProxyFilterTest : public testing::Test, public Common::Redis::Decoder } } + CommandSplitter::MockInstance& getSplitter() { + return reinterpret_cast(*filter_->splitter_); + } + // Common::Redis::DecoderFactory Common::Redis::DecoderPtr create(Common::Redis::DecoderCallbacks& callbacks) override { decoder_callbacks_ = &callbacks; @@ -161,7 +170,7 @@ class RedisProxyFilterTest : public testing::Test, public Common::Redis::Decoder Common::Redis::MockEncoder* encoder_{new Common::Redis::MockEncoder()}; Common::Redis::MockDecoder* decoder_{new Common::Redis::MockDecoder()}; Common::Redis::DecoderCallbacks* decoder_callbacks_{}; - CommandSplitter::MockInstance splitter_; + CommandSplitter::MockCommandSplitterFactory splitter_factory_; Stats::TestUtil::TestStore store_; NiceMock drain_decision_; NiceMock runtime_; @@ -181,12 +190,12 @@ TEST_F(RedisProxyFilterTest, OutOfOrderResponseWithDrainClose) { CommandSplitter::SplitCallbacks* request_callbacks2; EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { Common::Redis::RespValuePtr request1(new Common::Redis::RespValue()); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request1), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request1), _)) .WillOnce(DoAll(WithArg<1>(SaveArgAddress(&request_callbacks1)), Return(request_handle1))); decoder_callbacks_->onRespValue(std::move(request1)); Common::Redis::RespValuePtr request2(new Common::Redis::RespValue()); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request2), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request2), _)) .WillOnce(DoAll(WithArg<1>(SaveArgAddress(&request_callbacks2)), Return(request_handle2))); decoder_callbacks_->onRespValue(std::move(request2)); })); @@ -222,12 +231,12 @@ TEST_F(RedisProxyFilterTest, OutOfOrderResponseDownstreamDisconnectBeforeFlush) CommandSplitter::SplitCallbacks* request_callbacks2; EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { Common::Redis::RespValuePtr request1(new Common::Redis::RespValue()); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request1), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request1), _)) .WillOnce(DoAll(WithArg<1>(SaveArgAddress(&request_callbacks1)), Return(request_handle1))); decoder_callbacks_->onRespValue(std::move(request1)); Common::Redis::RespValuePtr request2(new Common::Redis::RespValue()); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request2), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request2), _)) .WillOnce(DoAll(WithArg<1>(SaveArgAddress(&request_callbacks2)), Return(request_handle2))); decoder_callbacks_->onRespValue(std::move(request2)); })); @@ -251,7 +260,7 @@ TEST_F(RedisProxyFilterTest, DownstreamDisconnectWithActive) { CommandSplitter::SplitCallbacks* request_callbacks1; EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { Common::Redis::RespValuePtr request1(new Common::Redis::RespValue()); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request1), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request1), _)) .WillOnce(DoAll(WithArg<1>(SaveArgAddress(&request_callbacks1)), Return(request_handle1))); decoder_callbacks_->onRespValue(std::move(request1)); })); @@ -269,7 +278,7 @@ TEST_F(RedisProxyFilterTest, ImmediateResponse) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request1)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request1), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request1), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -313,7 +322,7 @@ TEST_F(RedisProxyFilterTest, AuthWhenNotRequired) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -340,7 +349,7 @@ TEST_F(RedisProxyFilterTest, AuthAclWhenNotRequired) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -383,7 +392,7 @@ TEST_F(RedisProxyFilterWithAuthPasswordTest, AuthPasswordCorrect) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -410,7 +419,7 @@ TEST_F(RedisProxyFilterWithAuthPasswordTest, AuthPasswordIncorrect) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -455,7 +464,7 @@ TEST_F(RedisProxyFilterWithAuthAclTest, AuthAclCorrect) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -482,7 +491,7 @@ TEST_F(RedisProxyFilterWithAuthAclTest, AuthAclUsernameIncorrect) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { @@ -509,7 +518,7 @@ TEST_F(RedisProxyFilterWithAuthAclTest, AuthAclPasswordIncorrect) { EXPECT_CALL(*decoder_, decode(Ref(fake_data))).WillOnce(Invoke([&](Buffer::Instance&) -> void { decoder_callbacks_->onRespValue(std::move(request)); })); - EXPECT_CALL(splitter_, makeRequest_(Ref(*request), _)) + EXPECT_CALL(getSplitter(), makeRequest_(Ref(*request), _)) .WillOnce( Invoke([&](const Common::Redis::RespValue&, CommandSplitter::SplitCallbacks& callbacks) -> CommandSplitter::SplitRequest* { diff --git a/test/extensions/filters/network/redis_proxy/redis_proxy_integration_test.cc b/test/extensions/filters/network/redis_proxy/redis_proxy_integration_test.cc index dd0e64375b8cc..b9760e147206c 100644 --- a/test/extensions/filters/network/redis_proxy/redis_proxy_integration_test.cc +++ b/test/extensions/filters/network/redis_proxy/redis_proxy_integration_test.cc @@ -1,6 +1,9 @@ #include #include +#include "common/common/fmt.h" + +#include "extensions/filters/network/common/redis/fault_impl.h" #include "extensions/filters/network/redis_proxy/command_splitter_impl.h" #include "test/integration/integration.h" @@ -52,7 +55,7 @@ const std::string CONFIG = R"EOF( filters: name: redis typed_config: - "@type": type.googleapis.com/envoy.config.filter.network.redis_proxy.v2.RedisProxy + "@type": type.googleapis.com/envoy.extensions.filters.network.redis_proxy.v3.RedisProxy stat_prefix: redis_stats prefix_routes: catch_all_route: @@ -273,6 +276,27 @@ const std::string CONFIG_WITH_ROUTES_AND_AUTH_PASSWORDS = R"EOF( cluster: cluster_2 )EOF"; +// This is a configuration with fault injection enabled. +const std::string CONFIG_WITH_FAULT_INJECTION = CONFIG + R"EOF( + faults: + - fault_type: ERROR + fault_enabled: + default_value: + numerator: 100 + denominator: HUNDRED + commands: + - GET + - fault_type: DELAY + fault_enabled: + default_value: + numerator: 20 + denominator: HUNDRED + runtime_key: "bogus_key" + delay: 2s + commands: + - SET +)EOF"; + // This function encodes commands as an array of bulkstrings as transmitted by Redis clients to // Redis servers, according to the Redis protocol. std::string makeBulkStringArray(std::vector&& command_strings) { @@ -436,6 +460,12 @@ class RedisProxyWithCommandStatsIntegrationTest : public RedisProxyIntegrationTe : RedisProxyIntegrationTest(CONFIG_WITH_COMMAND_STATS, 2) {} }; +class RedisProxyWithFaultInjectionIntegrationTest : public RedisProxyIntegrationTest { +public: + RedisProxyWithFaultInjectionIntegrationTest() + : RedisProxyIntegrationTest(CONFIG_WITH_FAULT_INJECTION, 2) {} +}; + INSTANTIATE_TEST_SUITE_P(IpVersions, RedisProxyIntegrationTest, testing::ValuesIn(TestEnvironment::getIpVersionsForTest()), TestUtility::ipTestParamsToString); @@ -468,6 +498,10 @@ INSTANTIATE_TEST_SUITE_P(IpVersions, RedisProxyWithCommandStatsIntegrationTest, testing::ValuesIn(TestEnvironment::getIpVersionsForTest()), TestUtility::ipTestParamsToString); +INSTANTIATE_TEST_SUITE_P(IpVersions, RedisProxyWithFaultInjectionIntegrationTest, + testing::ValuesIn(TestEnvironment::getIpVersionsForTest()), + TestUtility::ipTestParamsToString); + void RedisProxyIntegrationTest::initialize() { setUpstreamCount(num_upstreams_); setDeterministic(); @@ -1062,5 +1096,25 @@ TEST_P(RedisProxyWithMirrorsIntegrationTest, EnabledViaRuntimeFraction) { redis_client->close(); } +TEST_P(RedisProxyWithFaultInjectionIntegrationTest, ErrorFault) { + std::string fault_response = + fmt::format("-{}\r\n", Extensions::NetworkFilters::Common::Redis::FaultMessages::get().Error); + initialize(); + simpleProxyResponse(makeBulkStringArray({"get", "foo"}), fault_response); + + EXPECT_EQ(1, test_server_->counter("redis.redis_stats.command.get.error")->value()); + EXPECT_EQ(1, test_server_->counter("redis.redis_stats.command.get.error_fault")->value()); +} + +TEST_P(RedisProxyWithFaultInjectionIntegrationTest, DelayFault) { + const std::string& set_request = makeBulkStringArray({"set", "write_only:toto", "bar"}); + const std::string& set_response = ":1\r\n"; + initialize(); + simpleRequestAndResponse(set_request, set_response); + + EXPECT_EQ(1, test_server_->counter("redis.redis_stats.command.set.success")->value()); + EXPECT_EQ(1, test_server_->counter("redis.redis_stats.command.set.delay_fault")->value()); +} + } // namespace } // namespace Envoy