diff --git a/fluxer_gifs/Cargo.toml b/fluxer_gifs/Cargo.toml index 599f82745..77de29c7d 100644 --- a/fluxer_gifs/Cargo.toml +++ b/fluxer_gifs/Cargo.toml @@ -14,7 +14,7 @@ fluxer-svc = { path = "../fluxer_svc", default-features = false } hmac = "0.13.0" moka = { version = "0.12.15", features = ["future", "sync"] } reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] } -serde = { version = "1.0.228", features = ["derive"] } +serde = { version = "1.0.228", features = ["derive", "rc"] } serde_json = "1.0.150" sha2 = "0.11.0" tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] } diff --git a/fluxer_gifs/src/klipy.rs b/fluxer_gifs/src/klipy.rs index a7826d5cb..ab3c2b9f9 100644 --- a/fluxer_gifs/src/klipy.rs +++ b/fluxer_gifs/src/klipy.rs @@ -9,7 +9,7 @@ use serde_json::Value; use std::collections::{BTreeMap, HashSet}; use std::sync::LazyLock; use std::time::Duration; -use tokio::time::sleep; +use tokio::time::{sleep, timeout}; const KLIPY_BASE_URL: &str = "https://api.klipy.com/v2"; const KLIPY_DIRECT_BASE_URL: &str = "https://api.klipy.com/api/v1"; @@ -18,6 +18,8 @@ const CLIENT_KEY: &str = "fluxer"; const MAX_RETRIES: usize = 3; const BACKOFF_BASE_DELAY: Duration = Duration::from_secs(1); const KLIPY_RESPONSE_LIMIT_BYTES: usize = 512 * 1024; +const MAX_FEATURED_CATEGORIES: usize = 50; +const FEATURED_CATEGORIES_FETCH_TIMEOUT: Duration = Duration::from_secs(30); const FLUXER_USER_AGENT: &str = "Fluxerbot/1.0 (+https://fluxer.app)"; const KLIPY_PROVIDER_NAME: &str = "klipy"; const KLIPY_FEATURED_CATEGORY_REFRESH_COUNTRY: &str = "US"; @@ -295,6 +297,19 @@ impl KlipyClient { &self, api_key: &str, locale: &str, + ) -> anyhow::Result> { + timeout( + FEATURED_CATEGORIES_FETCH_TIMEOUT, + self.fetch_featured_categories(api_key, locale), + ) + .await + .context("KLIPY featured categories request timed out")? + } + + async fn fetch_featured_categories( + &self, + api_key: &str, + locale: &str, ) -> anyhow::Result> { let normalized_locale = normalize_locale(locale); let response: TagsResponse = self @@ -317,6 +332,7 @@ impl KlipyClient { .map(|tag| tag.searchterm.trim().to_owned()) .filter(|term| !term.is_empty()) .filter(|term| seen.insert(term.clone())) + .take(MAX_FEATURED_CATEGORIES) .collect::>(); let mut categories = Vec::with_capacity(search_terms.len()); @@ -438,7 +454,7 @@ impl KlipyClient { where T: serde::de::DeserializeOwned, { - let response = self + let mut response = self .http_client .get(url) .send() @@ -460,13 +476,21 @@ impl KlipyClient { { anyhow::bail!("KLIPY response declared more than {KLIPY_RESPONSE_LIMIT_BYTES} bytes"); } - let bytes = response - .bytes() + let initial_capacity = response + .content_length() + .and_then(|length| usize::try_from(length).ok()) + .unwrap_or_default(); + let mut bytes = Vec::with_capacity(initial_capacity); + while let Some(chunk) = response + .chunk() .await .map_err(reqwest::Error::without_url) - .with_context(|| format!("KLIPY {label} response body failed"))?; - if bytes.len() > KLIPY_RESPONSE_LIMIT_BYTES { - anyhow::bail!("KLIPY {label} response exceeded {KLIPY_RESPONSE_LIMIT_BYTES} bytes"); + .with_context(|| format!("KLIPY {label} response body failed"))? + { + if bytes.len().saturating_add(chunk.len()) > KLIPY_RESPONSE_LIMIT_BYTES { + anyhow::bail!("KLIPY {label} response exceeded {KLIPY_RESPONSE_LIMIT_BYTES} bytes"); + } + bytes.extend_from_slice(&chunk); } serde_json::from_slice(&bytes) .with_context(|| format!("failed to parse KLIPY {label} response")) diff --git a/fluxer_gifs/src/main.rs b/fluxer_gifs/src/main.rs index 65dba3b4c..3d2666d6d 100644 --- a/fluxer_gifs/src/main.rs +++ b/fluxer_gifs/src/main.rs @@ -32,7 +32,7 @@ async fn main() -> anyhow::Result<()> { fluxer_svc::router::run_router(&config, router, transport).await } Mode::Shard => { - let shard = GifsShard::new(&config)?; + let shard = GifsShard::new()?; fluxer_svc::shard::run_shard(&config, shard, transport).await } } diff --git a/fluxer_gifs/src/shard_impl.rs b/fluxer_gifs/src/shard_impl.rs index a7c1bd354..ca1fcbfab 100644 --- a/fluxer_gifs/src/shard_impl.rs +++ b/fluxer_gifs/src/shard_impl.rs @@ -2,15 +2,21 @@ use crate::klipy::{KlipyClient, build_share_url, extract_slug_from_url, resolve_cache_key}; use crate::media_proxy::MediaProxyUrlBuilder; -use crate::types::{GifCategoryTag, GifItem, GifRequest, GifServiceResponse}; -use fluxer_svc::config::ServiceConfig; +use crate::types::{GifCategoryTag, GifItem, GifMediaFormat, GifRequest, GifServiceResponse}; +use fluxer_svc::config::optional_env; use fluxer_svc::shard::ShardService; -use moka::future::Cache; +use moka::notification::RemovalCause; +use moka::ops::compute::{CompResult, Op}; +use moka::{Expiry, future::Cache}; use std::collections::HashSet; +use std::fmt::Write; use std::future::Future; -use std::sync::Arc; +use std::mem::size_of; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; -use tokio::sync::Mutex; +use tokio::sync::{Semaphore, TryAcquireError}; +use tokio::time::MissedTickBehavior; const SEARCH_SOFT_TTL: Duration = Duration::from_secs(30); const SEARCH_HARD_TTL: Duration = Duration::from_secs(5 * 60); @@ -22,6 +28,24 @@ const CATEGORIES_SOFT_TTL: Duration = Duration::from_secs(24 * 60 * 60); const CATEGORIES_HARD_TTL: Duration = Duration::from_secs(48 * 60 * 60); const RESOLVE_SOFT_TTL: Duration = Duration::from_secs(30 * 60); const RESOLVE_HARD_TTL: Duration = Duration::from_secs(2 * 60 * 60); +const DEFAULT_SHARD_CACHE_MAX_BYTES: u64 = 512 * 1024 * 1024; +const MIN_SHARD_CACHE_MAX_BYTES: u64 = 16 * 1024 * 1024; +const CACHE_BUDGET_PARTS: u64 = 16; +const GIF_LIST_CACHE_PARTS: u64 = 10; +const CATEGORY_CACHE_PARTS: u64 = 1; +const SUGGESTION_CACHE_PARTS: u64 = 1; +const RESOLVED_CACHE_PARTS: u64 = 4; +const MAX_CONCURRENT_REFRESHES: usize = 8; +const CACHE_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(30); +const CACHE_ENTRY_OVERHEAD_BYTES: u64 = 256; +const BTREE_ENTRY_OVERHEAD_BYTES: u64 = 64; +const ARC_HEADER_BYTES: u64 = (2 * size_of::()) as u64; +const METRICS_ORDERING: Ordering = Ordering::Relaxed; + +const _: () = assert!( + GIF_LIST_CACHE_PARTS + CATEGORY_CACHE_PARTS + SUGGESTION_CACHE_PARTS + RESOLVED_CACHE_PARTS + == CACHE_BUDGET_PARTS +); #[derive(Clone)] pub struct GifsShard { @@ -30,20 +54,51 @@ pub struct GifsShard { struct GifsShardInner { klipy: KlipyClient, - gif_lists: Cache>>, - categories: Cache>>, - suggestions: Cache>>, - resolved: Cache>>, - refreshing: Mutex>, + gif_lists: ManagedCache>, + categories: ManagedCache>, + suggestions: ManagedCache>, + resolved: ManagedCache>, + refreshing: Mutex>, + refresh_permits: Arc, +} + +struct ManagedCache { + cache: Cache>, + telemetry: Arc, + max_weight_bytes: u64, +} + +impl Clone for ManagedCache { + fn clone(&self) -> Self { + Self { + cache: self.cache.clone(), + telemetry: Arc::clone(&self.telemetry), + max_weight_bytes: self.max_weight_bytes, + } + } } -#[derive(Debug, Clone)] struct Cached { - data: T, + data: Arc, stored_at: Instant, + generation: u64, + policy: CachePolicy, + retained_bytes: u32, } -#[derive(Debug, Clone, Copy)] +impl Clone for Cached { + fn clone(&self) -> Self { + Self { + data: Arc::clone(&self.data), + stored_at: self.stored_at, + generation: self.generation, + policy: self.policy, + retained_bytes: self.retained_bytes, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] struct CachePolicy { soft_ttl: Duration, hard_ttl: Duration, @@ -51,15 +106,252 @@ struct CachePolicy { impl CachePolicy { const fn new(soft_ttl: Duration, hard_ttl: Duration) -> Self { + assert!(soft_ttl.as_nanos() <= hard_ttl.as_nanos()); Self { soft_ttl, hard_ttl } } } -impl Cached { - fn new(data: T) -> Self { +struct CachedExpiry; + +impl Expiry> for CachedExpiry { + fn expire_after_create( + &self, + _key: &String, + value: &Cached, + _created_at: Instant, + ) -> Option { + Some(value.policy.hard_ttl) + } + + fn expire_after_update( + &self, + _key: &String, + value: &Cached, + _updated_at: Instant, + _duration_until_expiry: Option, + ) -> Option { + Some(value.policy.hard_ttl) + } +} + +#[derive(Clone, Hash, PartialEq, Eq)] +struct RefreshKey { + cache_name: &'static str, + key: String, +} + +struct CacheTelemetry { + name: &'static str, + hits_total: AtomicU64, + misses_total: AtomicU64, + size_evictions_total: AtomicU64, + expired_evictions_total: AtomicU64, + refresh_success_total: AtomicU64, + refresh_error_total: AtomicU64, + refresh_superseded_total: AtomicU64, + refresh_duplicate_total: AtomicU64, + refresh_saturated_total: AtomicU64, + refresh_in_flight: AtomicU64, + next_generation: AtomicU64, +} + +impl CacheTelemetry { + fn new(name: &'static str) -> Self { Self { - data, + name, + hits_total: AtomicU64::new(0), + misses_total: AtomicU64::new(0), + size_evictions_total: AtomicU64::new(0), + expired_evictions_total: AtomicU64::new(0), + refresh_success_total: AtomicU64::new(0), + refresh_error_total: AtomicU64::new(0), + refresh_superseded_total: AtomicU64::new(0), + refresh_duplicate_total: AtomicU64::new(0), + refresh_saturated_total: AtomicU64::new(0), + refresh_in_flight: AtomicU64::new(0), + next_generation: AtomicU64::new(1), + } + } + + fn next_generation(&self) -> u64 { + self.next_generation + .fetch_update(METRICS_ORDERING, METRICS_ORDERING, |generation| { + generation.checked_add(1) + }) + .expect("GIF cache generation exhausted") + } + + fn record_removal(&self, cause: RemovalCause) { + match cause { + RemovalCause::Expired => { + self.expired_evictions_total.fetch_add(1, METRICS_ORDERING); + } + RemovalCause::Size => { + self.size_evictions_total.fetch_add(1, METRICS_ORDERING); + } + RemovalCause::Explicit | RemovalCause::Replaced => {} + } + } +} + +struct RefreshRegistration { + inner: Arc, + telemetry: Arc, + key: RefreshKey, +} + +impl Drop for RefreshRegistration { + fn drop(&mut self) { + let removed = self + .inner + .refreshing + .lock() + .expect("GIF refresh registry mutex poisoned") + .remove(&self.key); + assert!(removed, "GIF refresh registry lost an active key"); + self.telemetry + .refresh_in_flight + .fetch_sub(1, METRICS_ORDERING); + } +} + +#[derive(Clone, Copy)] +struct CacheBudgets { + total: u64, + gif_lists: u64, + categories: u64, + suggestions: u64, + resolved: u64, +} + +impl CacheBudgets { + fn from_env() -> anyhow::Result { + let total = match optional_env("FLUXER_GIFS_SHARD_CACHE_MAX_BYTES") { + Some(value) => value.parse::().map_err(|error| { + anyhow::anyhow!("invalid FLUXER_GIFS_SHARD_CACHE_MAX_BYTES: {error}") + })?, + None => DEFAULT_SHARD_CACHE_MAX_BYTES, + }; + if total < MIN_SHARD_CACHE_MAX_BYTES { + anyhow::bail!( + "FLUXER_GIFS_SHARD_CACHE_MAX_BYTES must be at least {MIN_SHARD_CACHE_MAX_BYTES}" + ); + } + + let part = total / CACHE_BUDGET_PARTS; + let categories = part.saturating_mul(CATEGORY_CACHE_PARTS); + let suggestions = part.saturating_mul(SUGGESTION_CACHE_PARTS); + let resolved = part.saturating_mul(RESOLVED_CACHE_PARTS); + let gif_lists = total + .checked_sub(categories) + .and_then(|remaining| remaining.checked_sub(suggestions)) + .and_then(|remaining| remaining.checked_sub(resolved)) + .expect("GIF cache budget partition exceeds total"); + assert!(gif_lists >= part.saturating_mul(GIF_LIST_CACHE_PARTS)); + + Ok(Self { + total, + gif_lists, + categories, + suggestions, + resolved, + }) + } +} + +trait HeapSize { + fn heap_bytes(&self) -> u64; +} + +impl HeapSize for String { + fn heap_bytes(&self) -> u64 { + usize_bytes(self.capacity()) + } +} + +impl HeapSize for Vec +where + T: HeapSize, +{ + fn heap_bytes(&self) -> u64 { + self.iter().fold( + usize_bytes(self.capacity()).saturating_mul(usize_bytes(size_of::())), + |bytes, value| bytes.saturating_add(value.heap_bytes()), + ) + } +} + +impl HeapSize for Option +where + T: HeapSize, +{ + fn heap_bytes(&self) -> u64 { + self.as_ref().map(HeapSize::heap_bytes).unwrap_or_default() + } +} + +impl HeapSize for GifMediaFormat { + fn heap_bytes(&self) -> u64 { + self.src + .heap_bytes() + .saturating_add(self.proxy_src.heap_bytes()) + } +} + +impl HeapSize for GifItem { + fn heap_bytes(&self) -> u64 { + let string_bytes = [ + &self.id, + &self.slug, + &self.provider, + &self.title, + &self.url, + &self.src, + &self.proxy_src, + ] + .into_iter() + .fold(0_u64, |bytes, value| { + bytes.saturating_add(value.heap_bytes()) + }); + let media_bytes = self.media.iter().fold(0_u64, |bytes, (key, value)| { + bytes + .saturating_add(usize_bytes(size_of::())) + .saturating_add(usize_bytes(size_of::())) + .saturating_add(BTREE_ENTRY_OVERHEAD_BYTES) + .saturating_add(key.heap_bytes()) + .saturating_add(value.heap_bytes()) + }); + + string_bytes + .saturating_add(self.placeholder.heap_bytes()) + .saturating_add(media_bytes) + } +} + +impl HeapSize for GifCategoryTag { + fn heap_bytes(&self) -> u64 { + self.name + .heap_bytes() + .saturating_add(self.src.heap_bytes()) + .saturating_add(self.proxy_src.heap_bytes()) + .saturating_add(self.gif.heap_bytes()) + } +} + +impl Cached +where + T: HeapSize, +{ + fn new(data: T, policy: CachePolicy, generation: u64) -> Self { + let retained_bytes = ARC_HEADER_BYTES + .saturating_add(usize_bytes(size_of::())) + .saturating_add(data.heap_bytes()); + Self { + data: Arc::new(data), stored_at: Instant::now(), + generation, + policy, + retained_bytes: u32::try_from(retained_bytes).unwrap_or(u32::MAX).max(1), } } @@ -69,52 +361,69 @@ impl Cached { } impl GifsShard { - pub fn new(config: &ServiceConfig) -> anyhow::Result { + pub fn new() -> anyhow::Result { let media_proxy = MediaProxyUrlBuilder::from_env()?; let klipy = KlipyClient::new(media_proxy)?; - let max_capacity = config.cache_max_entries; - let max_cache_ttl = CATEGORIES_HARD_TTL; - Ok(Self { - inner: Arc::new(GifsShardInner { - klipy, - gif_lists: build_cache(max_capacity, max_cache_ttl), - categories: build_cache(max_capacity, max_cache_ttl), - suggestions: build_cache(max_capacity, max_cache_ttl), - resolved: build_cache(max_capacity, max_cache_ttl), - refreshing: Mutex::new(HashSet::new()), - }), - }) + let budgets = CacheBudgets::from_env()?; + tracing::info!( + total_cache_max_bytes = budgets.total, + gif_list_cache_max_bytes = budgets.gif_lists, + category_cache_max_bytes = budgets.categories, + suggestion_cache_max_bytes = budgets.suggestions, + resolved_cache_max_bytes = budgets.resolved, + "configured GIF shard cache budgets" + ); + + let inner = Arc::new(GifsShardInner { + klipy, + gif_lists: build_cache("gif_lists", budgets.gif_lists), + categories: build_cache("categories", budgets.categories), + suggestions: build_cache("suggestions", budgets.suggestions), + resolved: build_cache("resolved", budgets.resolved), + refreshing: Mutex::new(HashSet::new()), + refresh_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_REFRESHES)), + }); + spawn_cache_maintenance(&inner); + Ok(Self { inner }) } async fn get_cached( &self, - cache: Cache>, + cache: ManagedCache, key: String, policy: CachePolicy, fetch: Fetch, - ) -> anyhow::Result + ) -> anyhow::Result> where - T: Clone + Send + Sync + 'static, + T: HeapSize + Send + Sync + 'static, Fetch: Fn() -> Fut + Clone + Send + Sync + 'static, Fut: Future> + Send + 'static, { - if let Some(cached) = cache.get(&key).await { + if let Some(cached) = cache.cache.get(&key).await { + assert_eq!(cached.policy, policy, "GIF cache policy mismatch for {key}"); let age = cached.age(); - if age <= policy.soft_ttl { - return Ok(cached.data); - } if age <= policy.hard_ttl { - self.trigger_background_refresh(cache, key, policy, fetch); + cache.telemetry.hits_total.fetch_add(1, METRICS_ORDERING); + if age > policy.soft_ttl { + self.trigger_background_refresh(cache, key, policy, cached.generation, fetch); + } return Ok(cached.data); } - cache.invalidate(&key).await; + cache.cache.invalidate(&key).await; } + cache.telemetry.misses_total.fetch_add(1, METRICS_ORDERING); let fetch_for_load = fetch.clone(); + let telemetry_for_load = Arc::clone(&cache.telemetry); let cached = cache + .cache .try_get_with(key, async move { let data = fetch_for_load().await?; - Ok::, anyhow::Error>(Cached::new(data)) + Ok::, anyhow::Error>(Cached::new( + data, + policy, + telemetry_for_load.next_generation(), + )) }) .await .map_err(|error| anyhow::anyhow!("{}", error.as_ref()))?; @@ -123,36 +432,101 @@ impl GifsShard { fn trigger_background_refresh( &self, - cache: Cache>, + cache: ManagedCache, key: String, - _policy: CachePolicy, + policy: CachePolicy, + observed_generation: u64, fetch: Fetch, ) where - T: Clone + Send + Sync + 'static, + T: HeapSize + Send + Sync + 'static, Fetch: Fn() -> Fut + Clone + Send + Sync + 'static, Fut: Future> + Send + 'static, { - let this = self.clone(); - tokio::spawn(async move { - { - let mut refreshing = this.inner.refreshing.lock().await; - if !refreshing.insert(key.clone()) { - return; - } + let permit = match self.inner.refresh_permits.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(TryAcquireError::NoPermits) => { + cache + .telemetry + .refresh_saturated_total + .fetch_add(1, METRICS_ORDERING); + return; } + Err(TryAcquireError::Closed) => panic!("GIF refresh semaphore unexpectedly closed"), + }; + let refresh_key = RefreshKey { + cache_name: cache.telemetry.name, + key: key.clone(), + }; + { + let mut refreshing = self + .inner + .refreshing + .lock() + .expect("GIF refresh registry mutex poisoned"); + if !refreshing.insert(refresh_key.clone()) { + cache + .telemetry + .refresh_duplicate_total + .fetch_add(1, METRICS_ORDERING); + return; + } + } + cache + .telemetry + .refresh_in_flight + .fetch_add(1, METRICS_ORDERING); + let registration = RefreshRegistration { + inner: Arc::clone(&self.inner), + telemetry: Arc::clone(&cache.telemetry), + key: refresh_key, + }; - let result = fetch().await; - match result { + tokio::spawn(async move { + let _permit = permit; + let _registration = registration; + match fetch().await { Ok(data) => { - cache.insert(key.clone(), Cached::new(data)).await; + let replacement = Cached::new(data, policy, cache.telemetry.next_generation()); + let result = cache + .cache + .entry(key.clone()) + .and_compute_with(move |current| { + let operation = match current { + Some(entry) if entry.value().generation == observed_generation => { + Op::Put(replacement) + } + Some(_) => Op::Nop, + None => Op::Put(replacement), + }; + std::future::ready(operation) + }) + .await; + match result { + CompResult::Inserted(_) | CompResult::ReplacedWith(_) => { + cache + .telemetry + .refresh_success_total + .fetch_add(1, METRICS_ORDERING); + } + CompResult::Unchanged(_) => { + cache + .telemetry + .refresh_superseded_total + .fetch_add(1, METRICS_ORDERING); + } + CompResult::StillNone(_) | CompResult::Removed(_) => { + panic!("GIF cache refresh produced an impossible compute result"); + } + } } Err(error) => { + cache + .telemetry + .refresh_error_total + .fetch_add(1, METRICS_ORDERING); tracing::debug!(error = %error, cache_key = %key, "background GIF cache refresh failed"); } } - - let mut refreshing = this.inner.refreshing.lock().await; - refreshing.remove(&key); }); } @@ -334,6 +708,76 @@ impl GifsShard { .await?; Ok(GifServiceResponse::Resolved { gif }) } + + fn render_cache_metrics(&self, output: &mut String, cache: &ManagedCache) { + let telemetry = cache.telemetry.as_ref(); + let name = telemetry.name; + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_budget_bytes{{cache=\"{name}\"}} {}", + cache.max_weight_bytes + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_entries{{cache=\"{name}\"}} {}", + cache.cache.entry_count() + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_weighted_size_bytes{{cache=\"{name}\"}} {}", + cache.cache.weighted_size() + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_hits_total{{cache=\"{name}\"}} {}", + telemetry.hits_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_misses_total{{cache=\"{name}\"}} {}", + telemetry.misses_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_evictions_total{{cache=\"{name}\",cause=\"size\"}} {}", + telemetry.size_evictions_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_evictions_total{{cache=\"{name}\",cause=\"expired\"}} {}", + telemetry.expired_evictions_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"success\"}} {}", + telemetry.refresh_success_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"error\"}} {}", + telemetry.refresh_error_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"superseded\"}} {}", + telemetry.refresh_superseded_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"duplicate\"}} {}", + telemetry.refresh_duplicate_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_refreshes_total{{cache=\"{name}\",result=\"saturated\"}} {}", + telemetry.refresh_saturated_total.load(METRICS_ORDERING) + ); + let _ = writeln!( + output, + "fluxer_gifs_shard_cache_refreshes_in_flight{{cache=\"{name}\"}} {}", + telemetry.refresh_in_flight.load(METRICS_ORDERING) + ); + } } impl ShardService for GifsShard { @@ -344,6 +788,36 @@ impl ShardService for GifsShard { "gifs" } + fn render_prometheus_metrics(&self, output: &mut String) { + let _ = writeln!(output, "# TYPE fluxer_gifs_shard_cache_budget_bytes gauge"); + let _ = writeln!(output, "# TYPE fluxer_gifs_shard_cache_entries gauge"); + let _ = writeln!( + output, + "# TYPE fluxer_gifs_shard_cache_weighted_size_bytes gauge" + ); + let _ = writeln!(output, "# TYPE fluxer_gifs_shard_cache_hits_total counter"); + let _ = writeln!( + output, + "# TYPE fluxer_gifs_shard_cache_misses_total counter" + ); + let _ = writeln!( + output, + "# TYPE fluxer_gifs_shard_cache_evictions_total counter" + ); + let _ = writeln!( + output, + "# TYPE fluxer_gifs_shard_cache_refreshes_total counter" + ); + let _ = writeln!( + output, + "# TYPE fluxer_gifs_shard_cache_refreshes_in_flight gauge" + ); + self.render_cache_metrics(output, &self.inner.gif_lists); + self.render_cache_metrics(output, &self.inner.categories); + self.render_cache_metrics(output, &self.inner.suggestions); + self.render_cache_metrics(output, &self.inner.resolved); + } + async fn handle(&self, request: GifRequest) -> anyhow::Result { let response = match request { GifRequest::IsAvailable { api_key } => self.handle_available(api_key).await, @@ -399,12 +873,57 @@ impl ShardService for GifsShard { } } -fn build_cache(max_capacity: u64, time_to_live: Duration) -> Cache> +fn usize_bytes(value: usize) -> u64 { + u64::try_from(value).expect("usize must fit in u64") +} + +fn cache_entry_weight(key: &String, value: &Cached) -> u32 { + let bytes = CACHE_ENTRY_OVERHEAD_BYTES + .saturating_add(usize_bytes(size_of::>())) + .saturating_add(usize_bytes(size_of::())) + .saturating_add(usize_bytes(key.capacity())) + .saturating_add(u64::from(value.retained_bytes)); + u32::try_from(bytes).unwrap_or(u32::MAX).max(1) +} + +fn build_cache(name: &'static str, max_weight_bytes: u64) -> ManagedCache where - T: Clone + Send + Sync + 'static, + T: HeapSize + Send + Sync + 'static, { - Cache::builder() - .max_capacity(max_capacity) - .time_to_live(time_to_live) - .build() + assert!(max_weight_bytes > 0); + let telemetry = Arc::new(CacheTelemetry::new(name)); + let listener_telemetry = Arc::clone(&telemetry); + let cache = Cache::>::builder() + .max_capacity(max_weight_bytes) + .weigher(cache_entry_weight::) + .expire_after(CachedExpiry) + .eviction_listener(move |_key, _value, cause| { + listener_telemetry.record_removal(cause); + }) + .build(); + ManagedCache { + cache, + telemetry, + max_weight_bytes, + } +} + +fn spawn_cache_maintenance(inner: &Arc) { + let weak_inner = Arc::downgrade(inner); + tokio::spawn(async move { + let mut interval = tokio::time::interval(CACHE_MAINTENANCE_INTERVAL); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + interval.tick().await; + let Some(inner) = weak_inner.upgrade() else { + return; + }; + tokio::join!( + inner.gif_lists.cache.run_pending_tasks(), + inner.categories.cache.run_pending_tasks(), + inner.suggestions.cache.run_pending_tasks(), + inner.resolved.cache.run_pending_tasks(), + ); + } + }); } diff --git a/fluxer_gifs/src/types.rs b/fluxer_gifs/src/types.rs index 7cc3e09d2..052e1f482 100644 --- a/fluxer_gifs/src/types.rs +++ b/fluxer_gifs/src/types.rs @@ -2,6 +2,7 @@ use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; +use std::sync::Arc; #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "op", rename_all_fields = "snake_case")] @@ -56,16 +57,16 @@ pub enum GifServiceResponse { Available { available: bool, }, - SearchResults(Vec), + SearchResults(Arc>), Featured { - gifs: Vec, - categories: Vec, + gifs: Arc>, + categories: Arc>, }, - TrendingResults(Vec), - Suggestions(Vec), + TrendingResults(Arc>), + Suggestions(Arc>), Registered, Resolved { - gif: Option, + gif: Arc>, }, ShareUrl { url: String, diff --git a/fluxer_svc/src/metrics.rs b/fluxer_svc/src/metrics.rs index a42086bd2..b725b4de4 100644 --- a/fluxer_svc/src/metrics.rs +++ b/fluxer_svc/src/metrics.rs @@ -2,9 +2,11 @@ use libc::{CLOCK_MONOTONIC, clock_gettime, timespec}; use std::fmt::Write; +use std::sync::Arc; use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; const ORDERING: Ordering = Ordering::Relaxed; +pub(crate) type AdditionalMetricsRenderer = Arc; const HISTOGRAM_BUCKETS_MS: &[u64] = &[ 1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000, @@ -71,6 +73,7 @@ pub struct ServiceMetrics { pub shard_forwards_total: AtomicU64, pub request_duration: Histogram, start_ms: AtomicI64, + additional_renderer: Option, } impl Default for ServiceMetrics { @@ -83,11 +86,19 @@ impl Default for ServiceMetrics { shard_forwards_total: AtomicU64::new(0), request_duration: Histogram::new(), start_ms: AtomicI64::new(0), + additional_renderer: None, } } } impl ServiceMetrics { + pub(crate) fn with_additional_renderer(renderer: AdditionalMetricsRenderer) -> Self { + Self { + additional_renderer: Some(renderer), + ..Self::default() + } + } + pub fn init(&self) { self.start_ms.store(now_ms(), ORDERING); } @@ -167,6 +178,10 @@ impl ServiceMetrics { uptime_ms as f64 / 1000.0, ); + if let Some(renderer) = self.additional_renderer.as_ref() { + renderer(&mut out); + } + out } } diff --git a/fluxer_svc/src/shard.rs b/fluxer_svc/src/shard.rs index acd9a175b..bf9be4196 100644 --- a/fluxer_svc/src/shard.rs +++ b/fluxer_svc/src/shard.rs @@ -16,6 +16,7 @@ pub trait ShardService: Send + Sync + 'static { type Response: serde::Serialize + serde::de::DeserializeOwned + Send + 'static; fn service_name(&self) -> &str; + fn render_prometheus_metrics(&self, _output: &mut String) {} fn handle( &self, request: Self::Request, @@ -36,7 +37,12 @@ where let shard_subject = format!("svc.{name}.shard.{shard_id}"); let health_addr = config.listen_addr; - let metrics = Arc::new(ServiceMetrics::default()); + let metrics_service = service.clone(); + let additional_renderer: Arc = + Arc::new(move |output| metrics_service.render_prometheus_metrics(output)); + let metrics = Arc::new(ServiceMetrics::with_additional_renderer( + additional_renderer, + )); metrics.init(); let is_serving = Arc::new(AtomicBool::new(false));