feat(cogs): publish backend change streams to Kafka

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"

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 }

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 }

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

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__<PROPERTY>=<value>`
+    #[serde(default)]
+    pub storage_cogs: Option<CostTrackerConfig>,
+
     /// 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| {

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)