feat(cogs): ChangeStream trait, preparation for backend instrumentation

.github/workflows/build.yml

@@ -32,7 +32,9 @@ jobs:
       - name: Install dependencies
         run: |
           sudo apt-get update
-          sudo apt-get install -y llvm
+          # libcurl4-openssl-dev: librdkafka is built from source and links against libcurl,
+          # whose headers are not on the runner by default.
+          sudo apt-get install -y llvm libcurl4-openssl-dev
           curl -sL https://sentry.io/get-cli/ | bash
 
       - name: Install Rust Toolchain

.github/workflows/ci.yml

@@ -154,6 +154,13 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
+      # librdkafka is built from source and links against libcurl, whose headers are not
+      # on the runner by default. Needed by anything that builds the `kafka` feature.
+      - name: Install libcurl-dev
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y libcurl4-openssl-dev
+
       - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
 
       - name: Install Rust Toolchain

Dockerfile.cross

@@ -4,7 +4,7 @@ FROM rust:slim-bookworm
 RUN dpkg --add-architecture amd64 \
     && apt-get update -qq \
     && apt-get upgrade -y \
-    && apt-get install -y --no-install-recommends make protobuf-compiler libprotobuf-dev pkg-config git libssl-dev:amd64 gcc-x86-64-linux-gnu g++-x86-64-linux-gnu \
+    && apt-get install -y --no-install-recommends make protobuf-compiler libprotobuf-dev pkg-config git libssl-dev:amd64 libcurl4-openssl-dev:amd64 gcc-x86-64-linux-gnu g++-x86-64-linux-gnu \
     && rm -rf /var/lib/apt/lists/*
 
 RUN rustup target add x86_64-unknown-linux-gnu

bigtable-bench/src/main.rs

@@ -21,6 +21,7 @@ use yansi::Paint;
 use objectstore_service::backend::bigtable::BigTableBackend;
 use objectstore_service::backend::bigtable::BigTableConfig;
 use objectstore_service::backend::common::Backend;
+use objectstore_service::change_stream::ChangeStreamFactory;
 use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_service::stream;
 use objectstore_types::metadata::{ExpirationPolicy, Metadata};
@@ -92,13 +93,17 @@ async fn main() -> anyhow::Result<()> {
         ),
     }
 
-    let backend = BigTableBackend::new(BigTableConfig {
-        endpoint: args.addr.clone(),
-        project_id: args.project.clone(),
-        instance_name: args.instance.clone(),
-        table_name: args.table.clone(),
-        connections: Some(args.pool),
-    })
+    let backend = BigTableBackend::new(
+        BigTableConfig {
+            endpoint: args.addr.clone(),
+            project_id: args.project.clone(),
+            instance_name: args.instance.clone(),
+            table_name: args.table.clone(),
+            connections: Some(args.pool),
+            storage_cogs: None,
+        },
+        &ChangeStreamFactory::default(),
+    )
     .await
     .context("failed to connect to Bigtable")?;
 

objectstore-server/src/state.rs

@@ -9,6 +9,7 @@ use std::time::Duration;
 use anyhow::Result;
 use bytes::Bytes;
 use futures_util::Stream;
+use objectstore_service::change_stream::ChangeStreamFactory;
 use objectstore_service::concurrency::ConcurrencyLimiter;
 use objectstore_service::id::ObjectContext;
 use objectstore_service::{StorageService, backend};
@@ -61,7 +62,8 @@ impl Services {
         #[cfg(target_os = "linux")]
         tokio::spawn(track_allocator_metrics(config.runtime.metrics_interval));
 
-        let backend = backend::from_config(config.storage.clone()).await?;
+        let backend =
+            backend::from_config(config.storage.clone(), &ChangeStreamFactory::default()).await?;
         let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency)
             .with_queue(config.service.concurrency_queue)
             .with_timeout(config.service.concurrency_timeout)

objectstore-service/Cargo.toml

@@ -20,6 +20,7 @@ futures-util = { workspace = true }
 gcp_auth = { workspace = true }
 humantime = { workspace = true }
 humantime-serde = { workspace = true }
+objectstore-inventory-tracker = { workspace = true, features = ["kafka"], optional = true }
 objectstore-log = { workspace = true }
 objectstore-metrics = { workspace = true }
 objectstore-types = { workspace = true }
@@ -36,8 +37,16 @@ tonic = { workspace = true }
 tracing = { workspace = true }
 uuid = { workspace = true, features = ["v7"] }
 
+[features]
+# Publishes a per-record change stream used to attribute storage cost to app features.
+# Off by default: it pulls in librdkafka, whose native build needs a toolchain the
+# cross-build image would otherwise not need, and sandbox deployments have no topic or
+# consumer for the records anyway.
+storage-cogs = ["dep:objectstore-inventory-tracker"]
+
 [dev-dependencies]
 futures = { workspace = true }
+objectstore-inventory-tracker = { workspace = true, features = ["test-utils"] }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["test-util"] }
 zstd = { workspace = true }

objectstore-service/src/backend/bigtable.rs

@@ -47,6 +47,9 @@ use crate::backend::common::{
     Backend, DeleteResponse, GetResponse, HighVolumeBackend, MetadataResponse, PutResponse,
     TieredGet, TieredMetadata, TieredWrite, Tombstone,
 };
+use crate::change_stream::{
+    ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream,
+};
 use crate::error::{Error, Result};
 use crate::gcp_auth::PrefetchingTokenProvider;
 use crate::id::ObjectId;
@@ -124,6 +127,19 @@ pub struct BigTableConfig {
     ///
     /// - `OS__STORAGE__CONNECTIONS=16` (optional)
     pub connections: Option<usize>,
+
+    /// Reports what this backend stores, for per-usecase cost attribution.
+    ///
+    /// # Default
+    ///
+    /// `None`, which disables reporting for this backend.
+    ///
+    /// # Environment Variables
+    ///
+    /// - `OS__STORAGE__STORAGE_COGS__SHARED_RESOURCE_ID=bigtable_objectstore`
+    /// - `OS__STORAGE__STORAGE_COGS__SAMPLE_RATE=1.0` (optional)
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub storage_cogs: Option<CostTrackerStreamConfig>,
 }
 
 /// Connection timeout used for the initial connection to Bigtable.
@@ -170,6 +186,10 @@ pub struct BigTableBackend {
     instance_path: String,
     table_path: String,
     table_name: String,
+
+    /// Records each write/update/delete performed by the backend. May be
+    /// [`NoopStream`](crate::change_stream::NoopStream).
+    change_stream: Arc<dyn ChangeStream>,
 }
 
 impl fmt::Debug for BigTableBackend {
@@ -694,14 +714,19 @@ impl BigTableBackend {
     ///
     /// Pass an `endpoint` in the config to connect to a local emulator; omit it to use real GCP
     /// credentials. `connections` controls the gRPC connection pool size (defaults to 1).
-    pub async fn new(config: BigTableConfig) -> anyhow::Result<Self> {
+    pub async fn new(
+        config: BigTableConfig,
+        streams: &ChangeStreamFactory,
+    ) -> anyhow::Result<Self> {
         let BigTableConfig {
             endpoint,
             project_id,
             instance_name,
             table_name,
             connections,
+            storage_cogs,
         } = config;
+        let change_stream = streams.build(storage_cogs.as_ref());
 
         let bigtable = if let Some(ref endpoint) = endpoint {
             BigTableConnection::new_with_emulator(
@@ -734,6 +759,7 @@ impl BigTableBackend {
             instance_path: format!("projects/{project_id}/instances/{instance_name}"),
             table_path: client.get_full_table_name(&table_name),
             table_name,
+            change_stream,
         })
     }
 
@@ -952,6 +978,10 @@ impl Backend for BigTableBackend {
 
         Ok(())
     }
+
+    async fn join(&self) {
+        flush_change_stream(&self.change_stream).await;
+    }
 }
 
 #[async_trait::async_trait]
@@ -1311,15 +1341,19 @@ mod tests {
     //
     // Refer to the readme for how to set up the emulator.
 
-    async fn create_test_backend() -> Result<BigTableBackend> {
-        BigTableBackend::new(BigTableConfig {
+    fn test_config() -> BigTableConfig {
+        BigTableConfig {
             endpoint: Some("localhost:8086".into()),
             project_id: "testing".into(),
             instance_name: "objectstore".into(),
             table_name: "objectstore".into(),
             connections: None,
-        })
-        .await
+            storage_cogs: None,
+        }
+    }
+
+    async fn create_test_backend() -> Result<BigTableBackend> {
+        BigTableBackend::new(test_config(), &ChangeStreamFactory::default()).await
     }
 
     fn make_id() -> ObjectId {

objectstore-service/src/backend/gcs.rs

@@ -3,6 +3,7 @@
 use std::borrow::Cow;
 use std::collections::BTreeMap;
 use std::future::Future;
+use std::sync::Arc;
 use std::time::SystemTime;
 use std::{fmt, io};
 
@@ -21,6 +22,9 @@ use crate::backend::common::{
     self, Backend, DeleteResponse, GetResponse, MetadataResponse, MultipartUploadBackend,
     PutResponse,
 };
+use crate::change_stream::{
+    ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream,
+};
 use crate::error::{Error, Result};
 use crate::gcp_auth::PrefetchingTokenProvider;
 use crate::id::ObjectId;
@@ -73,6 +77,19 @@ pub struct GcsConfig {
     ///
     /// - `OS__STORAGE__BUCKET=my-gcs-bucket`
     pub bucket: String,
+
+    /// Reports what this backend stores, for per-usecase cost attribution.
+    ///
+    /// # Default
+    ///
+    /// `None`, which disables reporting for this backend.
+    ///
+    /// # Environment Variables
+    ///
+    /// - `OS__STORAGE__STORAGE_COGS__SHARED_RESOURCE_ID=gcs_objectstore`
+    /// - `OS__STORAGE__STORAGE_COGS__SAMPLE_RATE=1.0` (optional)
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub storage_cogs: Option<CostTrackerStreamConfig>,
 }
 
 /// Default endpoint used to access the GCS JSON API.
@@ -443,12 +460,21 @@ pub struct GcsBackend {
     endpoint: Url,
     bucket: String,
     token_provider: Option<PrefetchingTokenProvider>,
+
+    /// Records each write/update/delete performed by the backend. May be
+    /// [`NoopStream`](crate::change_stream::NoopStream).
+    change_stream: Arc<dyn ChangeStream>,
 }
 
 impl GcsBackend {
     /// Creates an authenticated GCS JSON API backend bound to the bucket in `config`.
-    pub async fn new(config: GcsConfig) -> anyhow::Result<Self> {
-        let GcsConfig { endpoint, bucket } = config;
+    pub async fn new(config: GcsConfig, streams: &ChangeStreamFactory) -> anyhow::Result<Self> {
+        let GcsConfig {
+            endpoint,
+            bucket,
+            storage_cogs,
+        } = config;
+        let change_stream = streams.build(storage_cogs.as_ref());
 
         let token_provider = if endpoint.is_none() {
             Some(PrefetchingTokenProvider::gcp_auth(TOKEN_SCOPES).await?)
@@ -463,6 +489,7 @@ impl GcsBackend {
             endpoint: endpoint_str.parse().context("invalid GCS endpoint URL")?,
             bucket,
             token_provider,
+            change_stream,
         })
     }
 
@@ -846,6 +873,10 @@ impl Backend for GcsBackend {
         })
         .await
     }
+
+    async fn join(&self) {
+        flush_change_stream(&self.change_stream).await;
+    }
 }
 
 #[derive(Debug, Deserialize)]
@@ -1208,12 +1239,16 @@ mod tests {
     //
     // Refer to the readme for how to set up the emulator.
 
-    async fn create_test_backend() -> Result<GcsBackend> {
-        GcsBackend::new(GcsConfig {
+    fn test_config() -> GcsConfig {
+        GcsConfig {
             endpoint: Some("http://localhost:8087".into()),
             bucket: "test-bucket".into(),
-        })
-        .await
+            storage_cogs: None,
+        }
+    }
+
+    async fn create_test_backend() -> Result<GcsBackend> {
+        GcsBackend::new(test_config(), &ChangeStreamFactory::default()).await
     }
 
     fn make_id() -> ObjectId {