Skip to content
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

All the fixes #339

Merged
merged 6 commits into from
May 22, 2024
Merged

All the fixes #339

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4751339
All the fixes
alexsnaps May 21, 2024
fc2e7ef
Clean up
alexsnaps May 21, 2024
bc2abfb
Fix tests
alexsnaps May 21, 2024
45b191e
Moar fixes
alexsnaps May 21, 2024
6f90be6
Avoid polluting the logs and cleanups
alexsnaps May 21, 2024
b277704
Remove is_alive and liveness_latency histogram
alexsnaps May 22, 2024
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
17 changes: 12 additions & 5 deletions limitador/src/storage/redis/counters_cache.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use std::sync::Arc;
use std::time::{Duration, SystemTime};
use tokio::select;
use tokio::sync::{Notify, Semaphore};
use tracing::info;

#[derive(Debug)]
pub struct CachedCounterValue {
Expand Down Expand Up @@ -221,17 +222,23 @@ impl Batcher {
return result;
} else {
ready = select! {
_ = self.notifier.notified() => self.batch_ready(max),
_ = async {
loop {
self.notifier.notified().await;
if self.batch_ready(max) {
break;
}
}
} => {
info!("Priority flush!");
true
},
_ = tokio::time::sleep(self.interval) => true,
}
}
}
}

pub fn is_empty(&self) -> bool {
self.updates.is_empty()
}

fn batch_ready(&self, size: usize) -> bool {
self.updates.len() >= size
|| self
Expand Down
10 changes: 9 additions & 1 deletion limitador/src/storage/redis/redis_async.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use redis::{AsyncCommands, RedisError};
use std::collections::HashSet;
use std::str::FromStr;
use std::time::{Duration, Instant};
use tracing::{debug_span, Instrument};
use tracing::{debug_span, warn, Instrument};

// Note: this implementation does not guarantee exact limits. Ensuring that we
// never go over the limits would hurt performance. This implementation
Expand Down Expand Up @@ -215,6 +215,14 @@ impl AsyncRedisStorage {
.clone()
.incr::<&str, i32, u64>("LIMITADOR_LIVE_CHECK", 1)
.await
.map_err(|err| {
Copy link
Member Author

Choose a reason for hiding this comment

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

This is_alive(&self) isn't used anymore afaict... but there is the .then wrt liveness_latency... So... what shall we do?

Copy link
Member

Choose a reason for hiding this comment

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

I haven't found a #[allow(dead_code)] that might affect this (not complaining), also, method is_alive is not found in any other part of the project than here (method definition)

Copy link
Member Author

Choose a reason for hiding this comment

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

@adam-cattermole what you recon we do? 🔥 the metric? Call it from a dedicated Future?

Copy link
Member

Choose a reason for hiding this comment

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

I reckon we just burn the metric/method. We can add this back if needed but for now we could just use the datastore_latency from the batcher calls to redis (I know even this metric isn't perfect due to size of batch etc). If we decide we want this down the line it shouldn't be too hard to add it back

let err = <RedisError as Into<StorageErr>>::into(err);
if !err.is_transient() {
panic!("Unrecoverable Redis error: {}", err);
}
warn!("Live check failure: {}", err);
err
})
.is_ok()
.then(|| histogram!("liveness_latency").record(now.elapsed().as_secs_f64()))
.is_some()
Expand Down
95 changes: 43 additions & 52 deletions limitador/src/storage/redis/redis_cached.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug_span, error, warn, Instrument};
use tracing::{debug_span, error, info, warn, Instrument};

// This is just a first version.
//
Expand Down Expand Up @@ -189,15 +189,13 @@ impl CachedRedisStorage {
AsyncRedisStorage::new_with_conn_manager(redis_conn_manager.clone());

{
let storage = async_redis_storage.clone();
let counters_cache_clone = counters_cache.clone();
let conn = redis_conn_manager.clone();
let p = Arc::clone(&partitioned);
tokio::spawn(async move {
loop {
flush_batcher_and_update_counters(
conn.clone(),
storage.is_alive().await,
counters_cache_clone.clone(),
p.clone(),
batch_size,
Expand Down Expand Up @@ -339,48 +337,53 @@ async fn update_counters<C: ConnectionLike>(
#[tracing::instrument(skip_all)]
async fn flush_batcher_and_update_counters<C: ConnectionLike>(
mut redis_conn: C,
storage_is_alive: bool,
cached_counters: Arc<CountersCache>,
partitioned: Arc<AtomicBool>,
batch_size: usize,
) {
if partitioned.load(Ordering::Acquire) || !storage_is_alive {
if !cached_counters.batcher().is_empty() {
let updated_counters = cached_counters
.batcher()
.consume(batch_size, |counters| {
if !counters.is_empty() && !partitioned.load(Ordering::Acquire) {
info!("Flushing {} counter updates", counters.len());
}
update_counters(&mut redis_conn, counters)
})
.await
.map(|res| {
flip_partitioned(&partitioned, false);
}
} else {
let updated_counters = cached_counters
.batcher()
.consume(batch_size, |counters| {
update_counters(&mut redis_conn, counters)
})
.await
.or_else(|(data, err)| {
if err.is_transient() {
flip_partitioned(&partitioned, true);
let counters = data.len();
let mut reverted = 0;
for (counter, old_value, pending_writes, _) in data {
if cached_counters
.return_pending_writes(&counter, old_value, pending_writes)
.is_err()
{
error!("Couldn't revert writes back to {:?}", &counter);
} else {
reverted += 1;
}
res
})
.or_else(|(data, err)| {
if err.is_transient() {
let new_partition = flip_partitioned(&partitioned, true);
if new_partition {
warn!("Error flushing {}", err);
}
let counters = data.len();
let mut reverted = 0;
for (counter, old_value, pending_writes, _) in data {
if cached_counters
.return_pending_writes(&counter, old_value, pending_writes)
.is_err()
{
error!("Couldn't revert writes back to {:?}", &counter);
} else {
reverted += 1;
}
}
if new_partition {
warn!("Reverted {} of {} counter increments", reverted, counters);
Ok(Vec::new())
} else {
Err(err)
}
})
.expect("Unrecoverable Redis error!");
Ok(Vec::new())
} else {
Err(err)
}
})
.expect("Unrecoverable Redis error!");

for (counter, new_value, remote_deltas, ttl) in updated_counters {
cached_counters.apply_remote_delta(counter, new_value, remote_deltas, ttl);
}
for (counter, new_value, remote_deltas, ttl) in updated_counters {
cached_counters.apply_remote_delta(counter, new_value, remote_deltas, ttl);
}
}

Expand Down Expand Up @@ -529,14 +532,8 @@ mod tests {
assert_eq!(c.hits(&counter), 2);
}

flush_batcher_and_update_counters(
mock_client,
true,
cached_counters.clone(),
partitioned,
100,
)
.await;
flush_batcher_and_update_counters(mock_client, cached_counters.clone(), partitioned, 100)
.await;

let c = cached_counters.get(&counter).unwrap();
assert_eq!(c.hits(&counter), 8);
Expand Down Expand Up @@ -581,14 +578,8 @@ mod tests {
assert_eq!(c.hits(&counter), 5);
}

flush_batcher_and_update_counters(
mock_client,
true,
cached_counters.clone(),
partitioned,
100,
)
.await;
flush_batcher_and_update_counters(mock_client, cached_counters.clone(), partitioned, 100)
.await;

let c = cached_counters.get(&counter).unwrap();
assert_eq!(c.hits(&counter), 5);
Expand Down
Loading