Skip to content

redis: prefixed routing #5658

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mattklein123 merged 19 commits into from
Mar 26, 2019
Merged

redis: prefixed routing #5658

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
a36017e
redis: prefixed routing
Jan 15, 2019
de88470
Formatting
Mar 5, 2019
da0e958
Address comments
Mar 11, 2019
3b12bcb
formatting
Mar 11, 2019
0fde3aa
Formatting
Mar 11, 2019
164159f
indenting
Mar 11, 2019
36b5ee2
Try to fix doc warning
Mar 12, 2019
74c2248
more formatting
Mar 19, 2019
8dd1e5f
update doc and add missing tests to cover prefix lookup edge cases
Mar 19, 2019
0b8e861
formatting
Mar 19, 2019
ce5b011
Valide uniqueness in trie, update typos in doc
Mar 21, 2019
b34640e
formatting
Mar 21, 2019
2168949
Finnish remove_prefix part
Mar 22, 2019
71e619e
fix typo
Mar 22, 2019
2625680
improve docs & tests. Cleanup old comments
Mar 22, 2019
5f9cbc9
Add overwrite_existing
Mar 22, 2019
9abb8f9
formatting
Mar 22, 2019
c6d6cc8
fix typo
Mar 25, 2019
ab39fee
Merge branch 'master' into redis-prefixed-key-routing
Mar 26, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions DEPRECATED.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ A logged warning is expected for each deprecated item that is in deprecation win
Set the `filter_enabled` field instead.
* Use of google.protobuf.Struct for extension opaque configs is deprecated. Use google.protobuf.Any instead or pack
google.protobuf.Struct in google.protobuf.Any.
* Use of `cluster`, found in [redis-proxy.proto](https://github.com/envoyproxy/envoy/blob/master/api/envoy/config/filter/network/redis_proxy/v2/redis_proxy.proto) is deprecated. Set a `PrefixRoutes.catch_all_cluster` instead.

## Version 1.9.0 (Dec 20, 2018)

Expand Down
60 changes: 58 additions & 2 deletions api/envoy/config/filter/network/redis_proxy/v2/redis_proxy.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,13 @@ message RedisProxy {
// Name of cluster from cluster manager. See the :ref:`configuration section
// <arch_overview_redis_configuration>` of the architecture overview for recommendations on
// configuring the backing cluster.
string cluster = 2 [(validate.rules).string.min_bytes = 1];
//
// .. attention::
//
// This field is deprecated. Use a :ref:`catch-all
// route<envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.PrefixRoutes.catch_all_cluster>`
// instead.
string cluster = 2 [deprecated = true];

// Redis connection pool settings.
message ConnPoolSettings {
Expand All @@ -48,10 +54,60 @@ message RedisProxy {
bool enable_hashtagging = 2;
}

// Network settings for the connection pool to the upstream cluster.
// Network settings for the connection pool to the upstream clusters.
ConnPoolSettings settings = 3 [(validate.rules).message.required = true];

// Indicates that latency stat should be computed in microseconds. By default it is computed in
// milliseconds.
bool latency_in_micros = 4;

message PrefixRoutes {
message Route {
// String prefix that must match the beginning of the keys. Envoy will always favor the
// longest match.
string prefix = 1 [(validate.rules).string.min_bytes = 1];

// Indicates if the prefix needs to be removed from the key when forwarded.
bool remove_prefix = 2;

// Upstream cluster to forward the command to.
string cluster = 3;

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this have a min-length validation?

}

// List of prefix routes.
repeated Route routes = 1 [(gogoproto.nullable) = false];

// Indicates that prefix matching should be case insensitive.
bool case_insensitive = 2;

// Optional catch-all route to forward commands that doesn't match any of the routes. The
// catch-all route becomes required when no routes are specified.
string catch_all_cluster = 3;
}

// List of **unique** prefixes used to separate keys from different workloads to different
// clusters. Envoy will always favor the longest match first in case of overlap. A catch-all
// cluster can be used to forward commands to when there is no match. Time complexity of the

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: s/to when/when

// lookups are in O(min(longest key prefix, key length)).
//
// Example:
//
// .. code-block:: yaml
//
// prefix_routes:
// routes:
// - prefix: "ab"
// cluster: "cluster_a"
// - prefix: "abc"
// cluster: "cluster_b"
// # next 2 lines will break uniqueness constraint

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would probably delete this. I think the error message generated would be obvious.

// # - prefix: "abc"
// # cluster: "cluster_c"
// - prefix: "a"
// cluster: "cluster_d"
//

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you add further text to the example along the lines of "When using the above routes, the following prefixes would be sent to: ..." (With some examples)

// See the :ref:`configuration section
// <arch_overview_redis_configuration>` of the architecture overview for recommendations on
// configuring the backing clusters.
PrefixRoutes prefix_routes = 5 [(gogoproto.nullable) = false];
}
5 changes: 4 additions & 1 deletion docs/root/intro/arch_overview/redis.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,9 @@ In this mode, the goals of Envoy are to maintain availability and partition tole
over consistency. This is the key point when comparing Envoy to `Redis Cluster
<https://redis.io/topics/cluster-spec>`_. Envoy is designed as a best-effort cache,
meaning that it will not try to reconcile inconsistent data or keep a globally consistent
view of cluster membership.
view of cluster membership. It also supports routing commands from different workload to
different to different upstream clusters based on their access patterns, eviction, or isolation
requirements.

The Redis project offers a thorough reference on partitioning as it relates to Redis. See
"`Partitioning: how to split data among multiple Redis instances
Expand All @@ -22,6 +24,7 @@ The Redis project offers a thorough reference on partitioning as it relates to R
* Detailed command statistics.
* Active and passive healthchecking.
* Hash tagging.
* Prefix routing.

**Planned future enhancements**:

Expand Down
1 change: 1 addition & 0 deletions docs/root/intro/version_history.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ Version history
* ratelimit: removed deprecated rate limit configuration from bootstrap.
* redis: added :ref:`hashtagging <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.ConnPoolSettings.enable_hashtagging>` to guarantee a given key's upstream.
* redis: added :ref:`latency stats <config_network_filters_redis_proxy_per_command_stats>` for commands.
* redis: added :ref:`prefix routing <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.prefix_routes>` to enable routing commands based on their key's prefix to different upstream.
* redis: added :ref:`success and error stats <config_network_filters_redis_proxy_per_command_stats>` for commands.
* redis: migrate hash function for host selection to `MurmurHash2 <https://sites.google.com/site/murmurhash>`_ from std::hash. MurmurHash2 is compatible with std::hash in GNU libstdc++ 3.4.20 or above. This is typically the case when compiled on Linux and not macOS.
* redis: added :ref:`latency_in_micros <envoy_api_field_config.filter.network.redis_proxy.v2.RedisProxy.latency_in_micros>` to specify the redis commands stats time unit in microseconds.
Expand Down
32 changes: 31 additions & 1 deletion source/common/common/utility.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -568,8 +568,9 @@ template <class Value> struct TrieLookupTable {
* Adds an entry to the Trie at the given Key.
* @param key the key used to add the entry.
* @param value the value to be associated with the key.
* @return false when a value already exists for the given key.
*/
void add(const char* key, Value value) {
bool add(const char* key, Value value) {

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add some explicit unit tests for the functions in this file.

TrieEntry<Value>* current = &root_;
while (uint8_t c = *key) {
if (!current->entries_[c]) {
Expand All @@ -578,7 +579,11 @@ template <class Value> struct TrieLookupTable {
current = current->entries_[c].get();
key++;
}
if (current->value_) {

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This might change the behavior for current callers. Should overwrite_existing instead be a param?

return false;
}
current->value_ = value;
return true;
}

/**
Expand All @@ -599,6 +604,31 @@ template <class Value> struct TrieLookupTable {
return current->value_;
}

/**
* Finds the entry associated with the longest prefix. Complexity is O(min(longest key prefix, key
* length))
* @param key the key used to find.
* @return the value matching the longest prefix based on the key.
*/
Value findPrefix(const char* key) const {

@mattklein123 mattklein123 Mar 22, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

findLongestPrefix ?

const TrieEntry<Value>* current = &root_;
const TrieEntry<Value>* result = nullptr;
while (uint8_t c = *key) {
if (current->value_) {
result = current;
}

// https://github.com/facebook/mcrouter/blob/master/mcrouter/lib/fbi/cpp/Trie-inl.h#L126-L143
current = current->entries_[c].get();
if (current == nullptr) {
return result ? result->value_ : nullptr;
}

key++;
}
return current ? current->value_ : result->value_;
}

TrieEntry<Value> root_;
};

Expand Down
29 changes: 26 additions & 3 deletions source/extensions/filters/network/redis_proxy/BUILD
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,22 @@ envoy_cc_library(
],
)

envoy_cc_library(
name = "router_interface",
hdrs = ["router.h"],
deps = [
":conn_pool_interface",
"@envoy_api//envoy/config/filter/network/redis_proxy/v2:redis_proxy_cc",
],
)

envoy_cc_library(
name = "command_splitter_lib",
srcs = ["command_splitter_impl.cc"],
hdrs = ["command_splitter_impl.h"],
deps = [
":command_splitter_interface",
":conn_pool_interface",
":router_interface",
"//include/envoy/stats:stats_macros",
"//include/envoy/stats:timespan",
"//source/common/common:assert_lib",
Expand All @@ -54,7 +63,6 @@ envoy_cc_library(
hdrs = ["conn_pool_impl.h"],
deps = [
":conn_pool_interface",
"//include/envoy/router:router_interface",
"//include/envoy/thread_local:thread_local_interface",
"//include/envoy/upstream:cluster_manager_interface",
"//source/common/buffer:buffer_lib",
Expand All @@ -73,6 +81,7 @@ envoy_cc_library(
hdrs = ["proxy_filter.h"],
deps = [
":command_splitter_interface",
":router_interface",
"//include/envoy/network:drain_decision_interface",
"//include/envoy/network:filter_interface",
"//include/envoy/upstream:cluster_manager_interface",
Expand All @@ -95,7 +104,21 @@ envoy_cc_library(
"//source/extensions/filters/network/common:factory_base_lib",
"//source/extensions/filters/network/common/redis:codec_lib",
"//source/extensions/filters/network/redis_proxy:command_splitter_lib",
"//source/extensions/filters/network/redis_proxy:conn_pool_lib",
"//source/extensions/filters/network/redis_proxy:proxy_filter_lib",
"//source/extensions/filters/network/redis_proxy:router_lib",
],
)

envoy_cc_library(
name = "router_lib",
srcs = ["router_impl.cc"],
hdrs = ["router_impl.h"],
deps = [
":router_interface",
"//include/envoy/thread_local:thread_local_interface",
"//include/envoy/upstream:cluster_manager_interface",
"//source/common/common:to_lower_table_lib",
"//source/extensions/filters/network/redis_proxy:conn_pool_lib",
"@envoy_api//envoy/config/filter/network/redis_proxy/v2:redis_proxy_cc",
],
)
41 changes: 20 additions & 21 deletions source/extensions/filters/network/redis_proxy/command_splitter_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,15 +59,15 @@ void SingleServerRequest::cancel() {
handle_ = nullptr;
}

SplitRequestPtr SimpleRequest::create(ConnPool::Instance& conn_pool,
SplitRequestPtr SimpleRequest::create(Router& router,
const Common::Redis::RespValue& incoming_request,
SplitCallbacks& callbacks, CommandStats& command_stats,
TimeSource& time_source, bool latency_in_micros) {
std::unique_ptr<SimpleRequest> request_ptr{
new SimpleRequest(callbacks, command_stats, time_source, latency_in_micros)};

request_ptr->handle_ = conn_pool.makeRequest(incoming_request.asArray()[1].asString(),
incoming_request, *request_ptr);
request_ptr->handle_ =
router.makeRequest(incoming_request.asArray()[1].asString(), incoming_request, *request_ptr);
if (!request_ptr->handle_) {
request_ptr->callbacks_.onResponse(Utility::makeError(Response::get().NoUpstreamHost));
return nullptr;
Expand All @@ -76,7 +76,7 @@ SplitRequestPtr SimpleRequest::create(ConnPool::Instance& conn_pool,
return std::move(request_ptr);
}

SplitRequestPtr EvalRequest::create(ConnPool::Instance& conn_pool,
SplitRequestPtr EvalRequest::create(Router& router,
const Common::Redis::RespValue& incoming_request,
SplitCallbacks& callbacks, CommandStats& command_stats,
TimeSource& time_source, bool latency_in_micros) {
Expand All @@ -91,8 +91,8 @@ SplitRequestPtr EvalRequest::create(ConnPool::Instance& conn_pool,

std::unique_ptr<EvalRequest> request_ptr{
new EvalRequest(callbacks, command_stats, time_source, latency_in_micros)};
request_ptr->handle_ = conn_pool.makeRequest(incoming_request.asArray()[3].asString(),
incoming_request, *request_ptr);
request_ptr->handle_ =
router.makeRequest(incoming_request.asArray()[3].asString(), incoming_request, *request_ptr);
if (!request_ptr->handle_) {
command_stats.error_.inc();
request_ptr->callbacks_.onResponse(Utility::makeError(Response::get().NoUpstreamHost));
Expand Down Expand Up @@ -123,7 +123,7 @@ void FragmentedRequest::onChildFailure(uint32_t index) {
onChildResponse(Utility::makeError(Response::get().UpstreamFailure), index);
}

SplitRequestPtr MGETRequest::create(ConnPool::Instance& conn_pool,
SplitRequestPtr MGETRequest::create(Router& router,
const Common::Redis::RespValue& incoming_request,
SplitCallbacks& callbacks, CommandStats& command_stats,
TimeSource& time_source, bool latency_in_micros) {
Expand Down Expand Up @@ -152,8 +152,8 @@ SplitRequestPtr MGETRequest::create(ConnPool::Instance& conn_pool,

single_mget.asArray()[1].asString() = incoming_request.asArray()[i].asString();
ENVOY_LOG(debug, "redis: parallel get: '{}'", single_mget.toString());
pending_request.handle_ = conn_pool.makeRequest(incoming_request.asArray()[i].asString(),
single_mget, pending_request);
pending_request.handle_ =
router.makeRequest(incoming_request.asArray()[i].asString(), single_mget, pending_request);
if (!pending_request.handle_) {
pending_request.onResponse(Utility::makeError(Response::get().NoUpstreamHost));
}
Expand Down Expand Up @@ -195,7 +195,7 @@ void MGETRequest::onChildResponse(Common::Redis::RespValuePtr&& value, uint32_t
}
}

SplitRequestPtr MSETRequest::create(ConnPool::Instance& conn_pool,
SplitRequestPtr MSETRequest::create(Router& router,
const Common::Redis::RespValue& incoming_request,
SplitCallbacks& callbacks, CommandStats& command_stats,
TimeSource& time_source, bool latency_in_micros) {
Expand Down Expand Up @@ -231,8 +231,8 @@ SplitRequestPtr MSETRequest::create(ConnPool::Instance& conn_pool,
single_mset.asArray()[2].asString() = incoming_request.asArray()[i + 1].asString();

ENVOY_LOG(debug, "redis: parallel set: '{}'", single_mset.toString());
pending_request.handle_ = conn_pool.makeRequest(incoming_request.asArray()[i].asString(),
single_mset, pending_request);
pending_request.handle_ =
router.makeRequest(incoming_request.asArray()[i].asString(), single_mset, pending_request);
if (!pending_request.handle_) {
pending_request.onResponse(Utility::makeError(Response::get().NoUpstreamHost));
}
Expand Down Expand Up @@ -270,7 +270,7 @@ void MSETRequest::onChildResponse(Common::Redis::RespValuePtr&& value, uint32_t
}
}

SplitRequestPtr SplitKeysSumResultRequest::create(ConnPool::Instance& conn_pool,
SplitRequestPtr SplitKeysSumResultRequest::create(Router& router,
const Common::Redis::RespValue& incoming_request,
SplitCallbacks& callbacks,
CommandStats& command_stats,
Expand Down Expand Up @@ -299,8 +299,8 @@ SplitRequestPtr SplitKeysSumResultRequest::create(ConnPool::Instance& conn_pool,
single_fragment.asArray()[1].asString() = incoming_request.asArray()[i].asString();
ENVOY_LOG(debug, "redis: parallel {}: '{}'", incoming_request.asArray()[0].asString(),
single_fragment.toString());
pending_request.handle_ = conn_pool.makeRequest(incoming_request.asArray()[i].asString(),
single_fragment, pending_request);
pending_request.handle_ = router.makeRequest(incoming_request.asArray()[i].asString(),
single_fragment, pending_request);
if (!pending_request.handle_) {
pending_request.onResponse(Utility::makeError(Response::get().NoUpstreamHost));
}
Expand Down Expand Up @@ -337,12 +337,11 @@ void SplitKeysSumResultRequest::onChildResponse(Common::Redis::RespValuePtr&& va
}
}

InstanceImpl::InstanceImpl(ConnPool::InstancePtr&& conn_pool, Stats::Scope& scope,
const std::string& stat_prefix, TimeSource& time_source,
bool latency_in_micros)
: conn_pool_(std::move(conn_pool)), simple_command_handler_(*conn_pool_),
eval_command_handler_(*conn_pool_), mget_handler_(*conn_pool_), mset_handler_(*conn_pool_),
split_keys_sum_result_handler_(*conn_pool_),
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."))},
latency_in_micros_(latency_in_micros), time_source_(time_source) {
for (const std::string& command : Common::Redis::SupportedCommands::simpleCommands()) {
Expand Down
Loading