feat(flags++): added quota limiting to the new flags service and in doing so refactored out a bunch of common stuff

rust/Cargo.toml

@@ -10,6 +10,7 @@ members = [
     "common/dns",
     "common/alloc",
     "common/types",
+    "common/limiters",
     "feature-flags",
     "hook-api",
     "hook-common",

rust/capture/Cargo.toml

@@ -18,7 +18,9 @@ flate2 = { workspace = true }
 governor = { workspace = true }
 health = { path = "../common/health" }
 common-alloc = { path = "../common/alloc" }
+common-redis = { path = "../common/redis" }
 common-types = { path = "../common/types" }
+limiters = { path = "../common/limiters" }
 metrics = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 opentelemetry = { workspace = true }

rust/capture/src/lib.rs

@@ -1,8 +1,6 @@
 pub mod api;
 pub mod config;
-pub mod limiters;
 pub mod prometheus;
-pub mod redis;
 pub mod router;
 pub mod server;
 pub mod sinks;

rust/capture/src/router.rs

@@ -12,9 +12,11 @@ use tower::limit::ConcurrencyLimitLayer;
 use tower_http::cors::{AllowHeaders, AllowOrigin, CorsLayer};
 use tower_http::trace::TraceLayer;
 
-use crate::limiters::token_dropper::TokenDropper;
 use crate::test_endpoint;
-use crate::{limiters::redis::RedisLimiter, redis::Client, sinks, time::TimeSource, v0_endpoint};
+use crate::{sinks, time::TimeSource, v0_endpoint};
+use common_redis::Client;
+use limiters::redis::RedisLimiter;
+use limiters::token_dropper::TokenDropper;
 
 use crate::config::CaptureMode;
 use crate::prometheus::{setup_metrics_recorder, track_metrics};

rust/capture/src/server.rs

@@ -2,27 +2,27 @@ use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
+use common_redis::RedisClient;
 use health::{ComponentStatus, HealthRegistry};
 use time::Duration;
 use tokio::net::TcpListener;
 
 use crate::config::CaptureMode;
 use crate::config::Config;
 
-use crate::limiters::overflow::OverflowLimiter;
-use crate::limiters::redis::{
+use limiters::overflow::OverflowLimiter;
+use limiters::redis::{
     QuotaResource, RedisLimiter, OVERFLOW_LIMITER_CACHE_KEY, QUOTA_LIMITER_CACHE_KEY,
 };
 
-use crate::limiters::token_dropper::TokenDropper;
-use crate::redis::RedisClient;
 use crate::router;
 use crate::router::BATCH_BODY_SIZE;
 use crate::sinks::fallback::FallbackSink;
 use crate::sinks::kafka::KafkaSink;
 use crate::sinks::print::PrintSink;
 use crate::sinks::s3::S3Sink;
 use crate::sinks::Event;
+use limiters::token_dropper::TokenDropper;
 
 async fn create_sink(
     config: &Config,

rust/capture/src/sinks/kafka.rs

@@ -1,8 +1,12 @@
-use crate::limiters::redis::RedisLimiter;
+use crate::api::CaptureError;
+use crate::config::KafkaConfig;
+use crate::prometheus::report_dropped_events;
+use crate::sinks::Event;
 use crate::v0_request::{DataType, ProcessedEvent};
 use async_trait::async_trait;
-
 use health::HealthHandle;
+use limiters::overflow::OverflowLimiter;
+use limiters::redis::RedisLimiter;
 use metrics::{counter, gauge, histogram};
 use rdkafka::error::{KafkaError, RDKafkaErrorCode};
 use rdkafka::message::{Header, OwnedHeaders};
@@ -14,12 +18,6 @@ use tokio::task::JoinSet;
 use tracing::log::{debug, error, info};
 use tracing::{info_span, instrument, Instrument};
 
-use crate::api::CaptureError;
-use crate::config::KafkaConfig;
-use crate::limiters::overflow::OverflowLimiter;
-use crate::prometheus::report_dropped_events;
-use crate::sinks::Event;
-
 struct KafkaContext {
     liveness: HealthHandle,
 }
@@ -376,13 +374,13 @@ impl Event for KafkaSink {
 mod tests {
     use crate::api::CaptureError;
     use crate::config;
-    use crate::limiters::overflow::OverflowLimiter;
     use crate::sinks::kafka::KafkaSink;
     use crate::sinks::Event;
     use crate::utils::uuid_v7;
     use crate::v0_request::{DataType, ProcessedEvent, ProcessedEventMetadata};
     use common_types::CapturedEvent;
     use health::HealthRegistry;
+    use limiters::overflow::OverflowLimiter;
     use rand::distributions::Alphanumeric;
     use rand::Rng;
     use rdkafka::mocking::MockCluster;

rust/capture/src/v0_endpoint.rs

@@ -9,12 +9,12 @@ use axum::http::{HeaderMap, Method};
 use axum_client_ip::InsecureClientIp;
 use base64::Engine;
 use common_types::{CapturedEvent, RawEvent};
+use limiters::token_dropper::TokenDropper;
 use metrics::counter;
 use serde_json::json;
 use serde_json::Value;
 use tracing::instrument;
 
-use crate::limiters::token_dropper::TokenDropper;
 use crate::prometheus::report_dropped_events;
 use crate::v0_request::{
     Compression, DataType, ProcessedEvent, ProcessedEventMetadata, ProcessingContext, RawRequest,

rust/capture/tests/common.rs

@@ -26,11 +26,9 @@ use tokio::time::timeout;
 use tracing::{debug, warn};
 
 use capture::config::{CaptureMode, Config, KafkaConfig};
-use capture::limiters::redis::{
-    QuotaResource, OVERFLOW_LIMITER_CACHE_KEY, QUOTA_LIMITER_CACHE_KEY,
-};
 use capture::server::serve;
 use health::HealthStrategy;
+use limiters::redis::{QuotaResource, OVERFLOW_LIMITER_CACHE_KEY, QUOTA_LIMITER_CACHE_KEY};
 
 pub static DEFAULT_CONFIG: Lazy<Config> = Lazy::new(|| Config {
     print_sink: false,

rust/capture/tests/django_compat.rs

@@ -6,14 +6,14 @@ use base64::engine::general_purpose;
 use base64::Engine;
 use capture::api::{CaptureError, CaptureResponse, CaptureResponseCode};
 use capture::config::CaptureMode;
-use capture::limiters::redis::{QuotaResource, RedisLimiter, QUOTA_LIMITER_CACHE_KEY};
-use capture::limiters::token_dropper::TokenDropper;
-use capture::redis::MockRedisClient;
 use capture::router::router;
 use capture::sinks::Event;
 use capture::time::TimeSource;
 use capture::v0_request::{DataType, ProcessedEvent};
+use common_redis::MockRedisClient;
 use health::HealthRegistry;
+use limiters::redis::{QuotaResource, RedisLimiter, QUOTA_LIMITER_CACHE_KEY};
+use limiters::token_dropper::TokenDropper;
 use serde::Deserialize;
 use serde_json::{json, Value};
 use std::fs::File;

rust/capture/tests/events.rs

@@ -4,7 +4,7 @@ use time::Duration;
 use crate::common::*;
 use anyhow::Result;
 use assert_json_diff::assert_json_include;
-use capture::limiters::redis::QuotaResource;
+use limiters::redis::QuotaResource;
 use reqwest::StatusCode;
 use serde_json::json;
 use uuid::Uuid;

rust/capture/tests/recordings.rs

@@ -2,7 +2,7 @@ use crate::common::*;
 use anyhow::Result;
 use assert_json_diff::assert_json_include;
 use capture::config::CaptureMode;
-use capture::limiters::redis::QuotaResource;
+use limiters::redis::QuotaResource;
 use reqwest::StatusCode;
 use serde_json::{json, value::Value};
 use time::Duration;

rust/common/limiters/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "limiters"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = { workspace = true }
+common-redis = { path = "../redis" }
+governor = { workspace = true }
+metrics = { workspace = true }
+rand = { workspace = true}
+tokio = { workspace = true }
+tracing = { workspace = true }
+time = { workspace = true }
+redis = { version = "0.23.3", features = [
+    "tokio-comp",
+    "cluster",
+    "cluster-async",
+] }

rust/capture/src/limiters/overflow.rs → rust/common/limiters/src/overflow.rs

@@ -72,7 +72,7 @@ impl OverflowLimiter {
 
 #[cfg(test)]
 mod tests {
-    use crate::limiters::overflow::OverflowLimiter;
+    use super::OverflowLimiter;
     use std::num::NonZeroU32;
 
     #[tokio::test]