diff --git a/Cargo.toml b/Cargo.toml index 820365e1..220b0473 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -86,7 +86,13 @@ sentry-options = "1.2.4" # - `ssl` is required for SASL/SCRAM # - `librdkafka` must be built from source either way, and `cmake-build` is the # recommended way to do it -rdkafka = { version = "0.39.0", features = ["cmake-build", "ssl", "tracing"] } +# - `libz-static` avoids a runtime dependency on `libz.so.1`; see objectstore-inventory-tracker +rdkafka = { version = "0.39.0", features = [ + "cmake-build", + "libz-static", + "ssl", + "tracing", +] } serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" serde_yaml = "0.9.34-deprecated" diff --git a/objectstore-inventory-tracker/Cargo.toml b/objectstore-inventory-tracker/Cargo.toml index 89d93214..58a8d4fa 100644 --- a/objectstore-inventory-tracker/Cargo.toml +++ b/objectstore-inventory-tracker/Cargo.toml @@ -22,6 +22,7 @@ blake3 = { workspace = true } # Allow a range of possible `rdkafka` versions so library users can pin whichever they use rdkafka = { version = ">=0.29.0, <0.40.0", features = [ "cmake-build", + "libz-static", # for distroless "ssl", "tracing", ], optional = true } diff --git a/objectstore-server/Cargo.toml b/objectstore-server/Cargo.toml index 3e3d446b..20d944ca 100644 --- a/objectstore-server/Cargo.toml +++ b/objectstore-server/Cargo.toml @@ -32,7 +32,7 @@ num_cpus = { workspace = true } objectstore-log = { workspace = true, features = ["init", "sentry"] } objectstore-metrics = { workspace = true } objectstore-options = { workspace = true } -objectstore-service = { workspace = true } +objectstore-service = { workspace = true, features = ["storage-cogs"] } objectstore-types = { workspace = true } papaya = { workspace = true } pin-project-lite = { workspace = true } diff --git a/objectstore-server/docs/architecture.md b/objectstore-server/docs/architecture.md index 3bb704b2..1871bc44 100644 --- a/objectstore-server/docs/architecture.md +++ b/objectstore-server/docs/architecture.md @@ -160,6 +160,8 @@ Key configuration sections: - `http` — HTTP layer parameters (concurrency limit) - `service` — storage service parameters (backend concurrency limit) - `killswitches` — traffic blocking rules +- `storage_cogs` — where backends publish their cost-tracking change stream; corresponds + to config on each backend. See [`change_stream`](objectstore_service::change_stream). - `usecases` — per-use-case properties (expiration policy constraints) - `runtime` — worker threads, metrics interval - `sentry` / `metrics` / `logging` — observability diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index b7f633f7..56af859d 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -41,8 +41,9 @@ use std::time::Duration; use anyhow::Result; use figment::providers::{Env, Format, Serialized, Yaml}; use objectstore_service::backend::local_fs::FileSystemConfig; +use objectstore_service::change_stream::CostTrackerConfig; use objectstore_types::auth::Permission; -use secrecy::{CloneableSecret, SecretBox, SerializableSecret, zeroize::Zeroize}; +use secrecy::{zeroize::Zeroize, CloneableSecret, SecretBox, SerializableSecret}; use serde::{Deserialize, Serialize}; pub use objectstore_log::{LevelFilter, LogFormat, LoggingConfig}; @@ -445,6 +446,32 @@ pub struct Config { /// ``` pub storage: StorageConfig, + /// Cost tracking sink for backends' change streams. + /// + /// A transport owns connections and a send queue, so it is configured once here and + /// shared by every backend. What each backend reports, and how much of it, is + /// configured per backend under [`storage`](Self::storage). + /// + /// Absent is the default, and disables reporting entirely. + /// + /// # Example + /// + /// ```yaml + /// storage_cogs: + /// type: kafka + /// topic: shared-resources-inventory + /// bootstrap_servers: [kafka:9092] + /// ``` + /// + /// # Environment Variables + /// + /// - `OS__STORAGE_COGS__TYPE=kafka` + /// - `OS__STORAGE_COGS__TOPIC=shared-resources-inventory` + /// - `OS__STORAGE_COGS__BOOTSTRAP_SERVERS=kafka:9092` + /// - `OS__STORAGE_COGS__OVERRIDE_PARAMS__=` + #[serde(default)] + pub storage_cogs: Option, + /// Configuration of the internal task runtime. /// /// Controls the thread pool size and behavior of the async runtime powering the server. @@ -635,6 +662,7 @@ impl Default for Config { path: PathBuf::from("data"), }), + storage_cogs: None, runtime: Runtime::default(), logging: LoggingConfig::default(), sentry: Sentry::default(), @@ -868,6 +896,66 @@ mod tests { }); } + #[test] + fn storage_cogs_via_env() { + figment::Jail::expect_with(|jail| { + jail.set_env("OS__STORAGE__TYPE", "bigtable"); + jail.set_env("OS__STORAGE__PROJECT_ID", "my-project"); + jail.set_env("OS__STORAGE__INSTANCE_NAME", "my-instance"); + jail.set_env("OS__STORAGE__TABLE_NAME", "my-table"); + jail.set_env( + "OS__STORAGE__STORAGE_COGS__SHARED_RESOURCE_ID", + "bigtable_objectstore", + ); + jail.set_env("OS__STORAGE__STORAGE_COGS__SAMPLE_RATE", "0.5"); + jail.set_env("OS__STORAGE_COGS__TYPE", "kafka"); + jail.set_env("OS__STORAGE_COGS__TOPIC", "my-topic"); + jail.set_env("OS__STORAGE_COGS__BOOTSTRAP_SERVERS", "[kafka:9092]"); + + let config = Config::load(None).unwrap(); + + let StorageConfig::BigTable(storage) = &dbg!(&config).storage else { + panic!("expected bigtable storage"); + }; + let stream = storage.storage_cogs.as_ref().expect("change stream"); + assert_eq!(stream.shared_resource_id, "bigtable_objectstore"); + assert_eq!(stream.sample_rate, 0.5); + + let CostTrackerConfig::Kafka(kafka) = + config.storage_cogs.as_ref().expect("kafka transport"); + assert_eq!(kafka.topic, "my-topic"); + assert_eq!(kafka.bootstrap_servers, ["kafka:9092"]); + + Ok(()) + }); + } + + #[test] + fn a_backend_stream_defaults_to_reporting_everything_and_no_transport() { + figment::Jail::expect_with(|jail| { + jail.set_env("OS__STORAGE__TYPE", "gcs"); + jail.set_env("OS__STORAGE__BUCKET", "my-bucket"); + jail.set_env( + "OS__STORAGE__STORAGE_COGS__SHARED_RESOURCE_ID", + "gcs_objectstore", + ); + + let config = Config::load(None).unwrap(); + + let StorageConfig::Gcs(storage) = &dbg!(&config).storage else { + panic!("expected gcs storage"); + }; + let stream = storage.storage_cogs.as_ref().expect("change stream"); + assert_eq!(stream.sample_rate, 1.0, "reports everything by default"); + assert!( + config.storage_cogs.is_none(), + "no transport is configured by default" + ); + + Ok(()) + }); + } + #[test] fn metrics_addr_via_env() { figment::Jail::expect_with(|jail| { diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index 111e2292..9143cb4d 100644 --- a/objectstore-server/src/state.rs +++ b/objectstore-server/src/state.rs @@ -62,8 +62,13 @@ impl Services { #[cfg(target_os = "linux")] tokio::spawn(track_allocator_metrics(config.runtime.metrics_interval)); - let backend = - backend::from_config(config.storage.clone(), &ChangeStreamFactory::default()).await?; + // Absent config means no transport, and therefore no reporting. + let streams = config + .storage_cogs + .as_ref() + .map(ChangeStreamFactory::new) + .unwrap_or_default(); + let backend = backend::from_config(config.storage.clone(), &streams).await?; let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency) .with_queue(config.service.concurrency_queue) .with_timeout(config.service.concurrency_timeout)