slingshot: configurable cache sizes #3

open
opened by nekomimi.pet targeting main from nekomimi.pet/microcosm-rs: main

I noticed defaults were hardcoded

./slingshot --jetstream us-east-1 --cache-dir ./foyer \
    --cache-memory-mb 128 \
    --cache-disk-gb 2
Changed files
+33 -8
slingshot
+4 -2
slingshot/src/firehose_cache.rs
···
pub async fn firehose_cache(
cache_dir: impl AsRef<Path>,
+
memory_mb: usize,
+
disk_gb: usize,
) -> Result<HybridCache<String, CachedRecord>, String> {
let cache = HybridCacheBuilder::new()
.with_name("firehose")
-
.memory(64 * 2_usize.pow(20))
+
.memory(memory_mb * 2_usize.pow(20))
.with_weighter(|k: &String, v| k.len() + std::mem::size_of_val(v))
.storage(Engine::large())
.with_device_options(
DirectFsDeviceOptions::new(cache_dir)
-
.with_capacity(2_usize.pow(30)) // TODO: configurable (1GB to have something)
+
.with_capacity(disk_gb * 2_usize.pow(30))
.with_file_size(16 * 2_usize.pow(20)), // note: this does limit the max cached item size, warning jumbo records
)
.build()
+22 -5
slingshot/src/main.rs
···
/// where to keep disk caches
#[arg(long)]
cache_dir: PathBuf,
+
/// memory cache size in MB
+
#[arg(long, default_value_t = 64)]
+
cache_memory_mb: usize,
+
/// disk cache size in GB
+
#[arg(long, default_value_t = 1)]
+
cache_disk_gb: usize,
+
/// port for HTTP server (when not using --domain)
+
#[arg(long, default_value_t = 3000)]
+
port: u16,
+
/// port for metrics/prometheus server
+
#[arg(long, default_value_t = 8765)]
+
metrics_port: u16,
/// the domain pointing to this server
///
/// if present:
···
let args = Args::parse();
-
if let Err(e) = install_metrics_server() {
+
if let Err(e) = install_metrics_server(args.metrics_port) {
log::error!("failed to install metrics server: {e:?}");
} else {
-
log::info!("metrics listening at http://0.0.0.0:8765");
+
log::info!("metrics listening at http://0.0.0.0:{}", args.metrics_port);
}
std::fs::create_dir_all(&args.cache_dir).map_err(|e| {
···
log::info!("cache dir ready at at {cache_dir:?}.");
log::info!("setting up firehose cache...");
-
let cache = firehose_cache(cache_dir.join("./firehose")).await?;
+
let cache = firehose_cache(
+
cache_dir.join("./firehose"),
+
args.cache_memory_mb,
+
args.cache_disk_gb,
+
)
+
.await?;
log::info!("firehose cache ready.");
let mut tasks: tokio::task::JoinSet<Result<(), MainTaskError>> = tokio::task::JoinSet::new();
···
args.domain,
args.acme_contact,
args.certs,
+
args.port,
server_shutdown,
)
.await?;
···
Ok(())
}
-
fn install_metrics_server() -> Result<(), metrics_exporter_prometheus::BuildError> {
+
fn install_metrics_server(port: u16) -> Result<(), metrics_exporter_prometheus::BuildError> {
log::info!("installing metrics server...");
let host = [0, 0, 0, 0];
-
let port = 8765;
PrometheusBuilder::new()
.set_quantiles(&[0.5, 0.9, 0.99, 1.0])?
.set_bucket_duration(std::time::Duration::from_secs(300))?
+7 -1
slingshot/src/server.rs
···
domain: Option<String>,
acme_contact: Option<String>,
certs: Option<PathBuf>,
+
port: u16,
shutdown: CancellationToken,
) -> Result<(), ServerError> {
let repo = Arc::new(repo);
···
)
.await
} else {
-
run(TcpListener::bind("127.0.0.1:3000"), app, shutdown).await
+
run(
+
TcpListener::bind(format!("127.0.0.1:{port}")),
+
app,
+
shutdown,
+
)
+
.await
}
}