fix: respect Redis cluster slots when inserting multiple items #8185
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| ### Respect Redis cluster slots when inserting multiple items ([PR #8185](https://github.com/apollographql/router/pull/8185)) | ||
|
|
||
| The existing `insert` code will silently fail rather than reporting an error. This PR fixes that behavior and adds new | ||
| metrics to track Redis client health. | ||
|
|
||
| New metrics: | ||
| * `apollo.router.cache.redis.unresponsive`: counter for 'unresponsive' events raised by the Redis library | ||
| * `kind`: Redis cache purpose (`APQ`, `query planner`, `entity`) | ||
| * `server`: Redis server that became unresponsive | ||
| * `apollo.router.cache.redis.reconnection`: counter for 'reconnect' events raised by the Redis library | ||
| * `kind`: Redis cache purpose (`APQ`, `query planner`, `entity`) | ||
| * `server`: Redis server that required client reconnection | ||
|
|
||
| By [@carodewig](https://github.com/carodewig) in https://github.com/apollographql/router/pull/8185 | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -117,6 +117,7 @@ where | |
| // when each clone is dropped, only when the last instance is dropped. | ||
| struct DropSafeRedisPool { | ||
| pool: Arc<RedisPool>, | ||
| caller: &'static str, | ||
| heartbeat_abort_handle: AbortHandle, | ||
| // Metrics collector handles its own abort and gauges | ||
| _metrics_collector: RedisMetricsCollector, | ||
|
|
@@ -133,10 +134,12 @@ impl Deref for DropSafeRedisPool { | |
| impl Drop for DropSafeRedisPool { | ||
| fn drop(&mut self) { | ||
| let inner = self.pool.clone(); | ||
| let caller = self.caller; | ||
| tokio::spawn(async move { | ||
| let result = inner.quit().await; | ||
| if let Err(err) = result { | ||
| tracing::warn!("Caught error while closing unused Redis connections: {err:?}"); | ||
| record_redis_error(&err, caller); | ||
| } | ||
| }); | ||
| self.heartbeat_abort_handle.abort(); | ||
|
|
@@ -151,7 +154,6 @@ pub(crate) struct RedisCacheStorage { | |
| pub(crate) ttl: Option<Duration>, | ||
| is_cluster: bool, | ||
| reset_ttl: bool, | ||
| } | ||
|
|
||
| fn get_type_of<T>(_: &T) -> &'static str { | ||
|
|
@@ -335,6 +337,7 @@ impl RedisCacheStorage { | |
| // spawn tasks that listen for connection close or reconnect events | ||
| let mut error_rx = client.error_rx(); | ||
| let mut reconnect_rx = client.reconnect_rx(); | ||
| let mut unresponsive_rx = client.unresponsive_rx(); | ||
|
|
||
| i64_up_down_counter_with_unit!( | ||
| "apollo.router.cache.redis.connections", | ||
|
|
@@ -348,22 +351,53 @@ impl RedisCacheStorage { | |
| loop { | ||
| match error_rx.recv().await { | ||
| Ok((error, Some(server))) => { | ||
| tracing::error!("Redis client ({server:?}) error: {error:?}",); | ||
| record_redis_error(&error, caller); | ||
| } | ||
| Ok((error, None)) => { | ||
| tracing::error!("Redis client error: {error:?}",); | ||
| record_redis_error(&error, caller); | ||
| } | ||
| Err(RecvError::Lagged(_)) => continue, | ||
| Err(RecvError::Closed) => break, | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| tokio::spawn(async move { | ||
| loop { | ||
| match unresponsive_rx.recv().await { | ||
| Ok(server) => { | ||
| tracing::debug!("Redis client ({server:?}) unresponsive"); | ||
| u64_counter_with_unit!( | ||
| "apollo.router.cache.redis.unresponsive", | ||
|
Contributor
There was a problem hiding this comment. is it worth distinguishing between the server and client? maybe as a tag or something, not sure; mostly, it'd be nice to know whether the client is struggling (eg, too many buffered commands while the server is still chomping away as expected) or the server is struggling (client is happy but the server has ground to a halt for some reason)
Contributor
Author
There was a problem hiding this comment. I think you'd be able to get the client via your metrics ingest engine - for example, IIRC prometheus adds the 'target' to metrics it scrapes. Or do you mean having a tag for the specific client within the router, if you've got multiple clients in the pool?
Contributor
There was a problem hiding this comment. more the ingestion bit; just some way to distinguish between server and client unresponsiveness, which if there's already some way to figure that out with defaults, then that'd be great
Contributor
Author
There was a problem hiding this comment. Sadly I don't think there's a good way to make that distinction; the unresponsive event is published by fred when it's gone a certain amount of time since hearing from the server per this config. But the other metrics Bryn added around command queue length etc might be a good way to diagnose this live -- if you see a bunch of |
||
| "Counter for Redis client unresponsive events", | ||
| "{event}", | ||
| 1, | ||
| kind = caller, | ||
| server = server.to_string() | ||
| ); | ||
| } | ||
| Err(RecvError::Lagged(_)) => continue, | ||
| Err(RecvError::Closed) => break, | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| tokio::spawn(async move { | ||
| loop { | ||
| match reconnect_rx.recv().await { | ||
| Ok(server) => { | ||
| u64_counter_with_unit!( | ||
| "apollo.router.cache.redis.reconnection", | ||
| "Counter for Redis client reconnection events", | ||
| "{reconnection}", | ||
| 1, | ||
| kind = caller, | ||
| server = server.to_string() | ||
| ); | ||
| tracing::info!("Redis client connected to {server:?}") | ||
| } | ||
| Err(RecvError::Lagged(_)) => continue, | ||
| Err(RecvError::Closed) => break, | ||
| } | ||
|
|
@@ -382,10 +416,8 @@ impl RedisCacheStorage { | |
| }); | ||
| } | ||
|
|
||
| // NB: error is not recorded here as it will be observed by the task following `client.error_rx()` | ||
| let _handle = pooled_client.init().await?; | ||
| let heartbeat_clients = pooled_client.clone(); | ||
| let heartbeat_handle = tokio::spawn(async move { | ||
| heartbeat_clients | ||
|
|
@@ -401,14 +433,14 @@ impl RedisCacheStorage { | |
| Ok(Self { | ||
| inner: Arc::new(DropSafeRedisPool { | ||
| pool: pooled_client_arc, | ||
| caller, | ||
| heartbeat_abort_handle: heartbeat_handle.abort_handle(), | ||
| _metrics_collector: metrics_collector, | ||
| }), | ||
| namespace: namespace.map(Arc::new), | ||
| ttl, | ||
| is_cluster, | ||
| reset_ttl, | ||
| }) | ||
| } | ||
|
|
||
|
|
@@ -418,7 +450,7 @@ impl RedisCacheStorage { | |
|
|
||
| /// Helper method to record Redis errors for metrics | ||
| fn record_error(&self, error: &RedisError) { | ||
| record_redis_error(error, self.inner.caller); | ||
| } | ||
|
|
||
| fn preprocess_urls(urls: Vec<Url>) -> Result<Url, RedisError> { | ||
|
|
@@ -649,6 +681,9 @@ impl RedisCacheStorage { | |
| .set::<(), _, _>(key, value, expiration, None, false) | ||
| .await; | ||
| tracing::trace!("insert result {:?}", r); | ||
| if let Err(err) = r { | ||
| self.record_error(&err); | ||
| } | ||
| } | ||
|
|
||
| pub(crate) async fn insert_multiple<K: KeyType, V: ValueType>( | ||
|
|
@@ -657,29 +692,25 @@ impl RedisCacheStorage { | |
| ttl: Option<Duration>, | ||
| ) { | ||
| tracing::trace!("inserting into redis: {:#?}", data); | ||
| let expiration = ttl | ||
| .or(self.ttl) | ||
| .map(|ttl| Expiration::EX(ttl.as_secs() as i64)); | ||
|
|
||
| // NB: if we were using MSET here, we'd need to split the keys by hash slot. however, fred | ||
| // seems to split the pipeline by hash slot in the background. | ||
| let pipeline = self.inner.next().pipeline(); | ||
| for (key, value) in data { | ||
| let key = self.make_key(key.clone()); | ||
| let _ = pipeline | ||
| .set::<(), _, _>(key, value.clone(), expiration.clone(), None, false) | ||
|
Contributor
There was a problem hiding this comment. if I understand this right, previously when we had no ttl, we'd use not a blocker, I don't think, because a working
Contributor
Author
There was a problem hiding this comment. That's correct! I haven't profiled the But most users will not have been hitting the |
||
| .await; | ||
| } | ||
|
|
||
|
carodewig marked this conversation as resolved.
|
||
| let result: Result<Vec<()>, _> = pipeline.all().await; | ||
|
carodewig marked this conversation as resolved.
|
||
| if let Err(err) = result { | ||
| tracing::trace!("caught error during insert: {err:?}"); | ||
| self.record_error(&err); | ||
| } | ||
| } | ||
|
|
||
| /// Delete keys *without* adding the `namespace` prefix because `keys` is from | ||
|
|
@@ -726,10 +757,20 @@ impl RedisCacheStorage { | |
|
|
||
| #[cfg(test)] | ||
| mod test { | ||
| use std::collections::HashMap; | ||
| use std::time::SystemTime; | ||
|
|
||
| use fred::types::cluster::ClusterRouting; | ||
| use itertools::Itertools; | ||
| use rand::Rng; | ||
| use rand::RngCore; | ||
| use rand::distr::Alphanumeric; | ||
| use serde_json::json; | ||
| use tower::BoxError; | ||
| use url::Url; | ||
|
|
||
| use crate::cache::redis::RedisKey; | ||
| use crate::cache::redis::RedisValue; | ||
| use crate::cache::storage::ValueType; | ||
|
|
||
| #[test] | ||
|
|
@@ -836,4 +877,71 @@ mod test { | |
| let urls = vec![url, url_1]; | ||
| assert!(super::RedisCacheStorage::preprocess_urls(urls).is_err()); | ||
| } | ||
|
|
||
| #[tokio::test(flavor = "multi_thread")] | ||
| async fn test_redis_cluster_insert_get_mget() -> Result<(), BoxError> { | ||
| let config_json = json!({ | ||
| "urls": ["redis-cluster://localhost:7000"], | ||
| "namespace": "test_redis_cluster", | ||
| "required_to_start": true, | ||
| "ttl": "60s" | ||
| }); | ||
| let config = serde_json::from_value(config_json).unwrap(); | ||
| let storage = super::RedisCacheStorage::new(config, "test_redis_cluster").await; | ||
|
|
||
| // only error for lack of storage when running in CI. otherwise, skip this test. | ||
| #[cfg(not(all(feature = "ci", all(target_arch = "x86_64", target_os = "linux"))))] | ||
| if storage.is_err() { | ||
| return Ok(()); | ||
| } | ||
| let storage = storage?; | ||
|
|
||
| // insert values which reflect different cluster slots to properly test cluster behavior | ||
|
carodewig marked this conversation as resolved.
Outdated
|
||
| let mut data = HashMap::default(); | ||
| let expected_value = rand::rng().next_u32() as usize; | ||
| let unique_cluster_slot_count = |data: &HashMap<RedisKey<String>, _>| { | ||
| data.keys() | ||
| .map(|key| ClusterRouting::hash_key(key.0.as_bytes())) | ||
| .unique() | ||
| .count() | ||
| }; | ||
|
|
||
| while unique_cluster_slot_count(&data) < 50 { | ||
| // NB: include {} around key so that this key is what determines the cluster hash slot - adding | ||
| // the namespace will otherwise change the slot | ||
| let key = rand::rng() | ||
| .sample_iter(&Alphanumeric) | ||
| .take(10) | ||
| .map(char::from) | ||
| .collect::<String>(); | ||
| data.insert(RedisKey(format!("{{{}}}", key)), RedisValue(expected_value)); | ||
| } | ||
|
|
||
| // insert values | ||
| let keys: Vec<_> = data.keys().cloned().collect(); | ||
| let data: Vec<_> = data.into_iter().collect(); | ||
| storage.insert_multiple(&data, None).await; | ||
|
|
||
| // make a `get` call for each key and ensure that it has the expected value. this tests both | ||
| // the `get` and `insert_multiple` functions | ||
| for key in &keys { | ||
| let value: RedisValue<usize> = storage | ||
| .get(key.clone()) | ||
| .await | ||
| .ok_or("unable to get value")?; | ||
| assert_eq!(value.0, expected_value); | ||
| } | ||
|
|
||
| // test the `mget` functionality | ||
| let values = storage | ||
| .get_multiple(keys) | ||
| .await | ||
| .ok_or("unable to get_multiple")?; | ||
| for value in values { | ||
| let value: RedisValue<usize> = value.ok_or("missing value")?; | ||
| assert_eq!(value.0, expected_value); | ||
| } | ||
|
Comment on lines
+949
to
+952
Contributor
There was a problem hiding this comment. just to doublecheck because I tripped over it when trying to understand the test: we're setting the same value for all keys and then getting all keys (first as a set of gets and then as one big mget) to check their values, which is just the same int?
Contributor
Author
There was a problem hiding this comment. Yep! The idea with the 'same value' bit is if this test is somehow running twice against the same redis cluster, we're actually getting the value for this test. Probably unnecessary, but didn't add much complexity so I thought it was worth it as a backup. |
||
|
|
||
| Ok(()) | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,3 +30,34 @@ services: | |
| read_only: true | ||
| ports: | ||
| - 8126:8126 | ||
|
|
||
| # redis cluster | ||
|
Contributor
There was a problem hiding this comment. this is the only part of the pr that I'm sort of hesitant about; I don't think it's something to solve here (maybe I can take what's here and iterate on it), but it'd be super nice to not run both standalone redis and clustered redis (my office is in the attic and gets too warm already without having docker run more stuff)
Contributor
Author
There was a problem hiding this comment. I agree! Personally I think it'd be better to only use clustered Redis, but I didn't want to make that change everywhere as part of this PR in case others disagree.
Contributor
There was a problem hiding this comment. whew, spent some time on this and everything is messy apart from just running both!
Contributor
Author
There was a problem hiding this comment. I think it might be good to do something like this in the future, where we test against both clustered and non-clustered redis! ie parameterize the tests and use #[tokio::test]
#[rstest::rstest]
async fn multiple_documents(
#[values(true, false)] clustered: bool,
) -> Result<(), BoxError> {
let config = redis_config(clustered);
todo!()
} |
||
| redis-cluster-7000: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "redis-server", "--protected-mode", "no", "--port", "7000", "--cluster-enabled", "yes" ] | ||
| network_mode: host | ||
| redis-cluster-7001: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "redis-server", "--protected-mode", "no", "--port", "7001", "--cluster-enabled", "yes" ] | ||
| network_mode: host | ||
| redis-cluster-7002: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "redis-server", "--protected-mode", "no", "--port", "7002", "--cluster-enabled", "yes" ] | ||
| network_mode: host | ||
| redis-cluster-7003: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "redis-server", "--protected-mode", "no", "--port", "7003", "--cluster-enabled", "yes" ] | ||
| network_mode: host | ||
| redis-cluster-7004: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "redis-server", "--protected-mode", "no", "--port", "7004", "--cluster-enabled", "yes" ] | ||
| network_mode: host | ||
| redis-cluster-7005: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "redis-server", "--protected-mode", "no", "--port", "7005", "--cluster-enabled", "yes" ] | ||
| network_mode: host | ||
| redis-cluster-startup: | ||
| image: cimg/redis:7.4.5 | ||
| command: [ "sh", "-c", "sleep 30; echo yes | redis-cli --cluster create --cluster-replicas 1 localhost:7000 localhost:7001 localhost:7002 localhost:7003 localhost:7004 localhost:7005" ] | ||
| network_mode: host | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.