From 836790e44194f0fd72ef6f14d285b7e257183863 Mon Sep 17 00:00:00 2001 From: Matt Hammerly Date: Fri, 14 Aug 2026 16:07:40 -0700 Subject: [PATCH] feat(cogs): report backend changes to a change stream --- objectstore-service/docs/architecture.md | 62 +++- objectstore-service/src/backend/bigtable.rs | 300 ++++++++++++++- objectstore-service/src/backend/gcs.rs | 349 ++++++++++++++++-- .../src/change_stream/factory.rs | 72 ++++ objectstore-service/src/change_stream/mod.rs | 3 + 5 files changed, 720 insertions(+), 66 deletions(-) diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 0b745d46..34ff2d1c 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -108,24 +108,68 @@ sequences. # Cost of Goods Sold (COGS) Accounting -[`StorageService::new`] wraps the configured backend in a -[`CountingBackend`](backend::counting::CountingBackend), a -[`Backend`](backend::common::Backend) decorator that increments the -`objectstore.cogs.usage` counter (tagged with an `app_feature` derived from the -usecase) once per operation. Multipart operations are also counted. +Objectstore emits attribution data that can break Objectstore costs (compute and +storage) down proportionally by usecase (or `app_feature`, as it's called in our +COGS pipelines). To calculate, for example, the compute costs for the +`attachments` usecase, multiply Objectstore's overall compute cost by the +`attachments` usecase's proportional weight in our compute attribution data. + +## Compute COGS + +Objectstore emits the `objectstore.cogs.usage` counter with an `app_feature` +label derived from the usecase once per operation. Multipart and batch +operations are also counted. This counter can be straightforwardly summed by +`app_feature`. + +The counter is incremented in the [`CountingBackend`](backend::counting::CountingBackend) +decorator which [`StorageService::new`] applies to its backend. Wrapping the +outermost decorator owned by `StorageService` covers every operation called by +`StorageService` itself as well as batched operations that are run through +[`StreamExecutor`](crate::streaming::StreamExecutor). For COGS purposes we use operation count as a proxy for compute cost under the assumption that each operation we serve has a basically flat CPU cost. Large payloads take longer, but they can be streamed in the background while other operations are served so they don't really cost more. -Wrapping the outermost backend owned by `StorageService` covers every operation -called by `StorageService` itself as well as batched operations that are run -through [`StreamExecutor`](crate::streaming::StreamExecutor). - Notably, operations that fail before reaching `StorageService` (e.g. auth or rate-limiting failures at a higher layer) are not counted. +## Storage COGS + +Each backend reports every write/overwrite, TTI bump, and delete it performs on +stored objects to a [`ChangeStream`](change_stream::ChangeStream). To +turn this change stream into COGS data, a stream consumer has to merge each +change event into an external table to update an inventory of objects. The +inventory table can be queried to break down each backend's storage utilization +by `app_feature`. Note that [`NoopStream`](change_stream::NoopStream) is used +unless the backend's config includes a +[`CostTrackerStreamConfig`](change_stream::CostTrackerStreamConfig) and the +service has a usable transport for it, and unless the `storage-cogs` feature is +compiled in at all. + +Each row in the inventory table has an anonymized hash of an `ObjectId` as well +as the row's size, expiry, Sentry org/project, `app_feature`, and relevant +backend. When using [`TieredStorage`](backend::tiered::TieredStorage)'s +long-term backend the inventory table will contain _two rows_ for an object: a +row for the actual object and its size in long-term backend, and a separate row +for the tombstone and the tombstone's size in the high-volume backend. + +`ChangeStream` is not aware of any automatic garbage collection that backends +may perform. Expired objects must be filtered out when querying the inventory +table. + +Under the hood, `CostTrackerStream` uses +[`InventoryTracker`](objectstore_inventory_tracker::InventoryTracker) to publish +change events; it is generic over the transport rather than tied to Kafka. Each +backend has its own sampling rate to lessen the load put on the stream +processor. Sampling decisions are made +based on [`ObjectId`](id::ObjectId). Each change event includes the sampling rate that was in +effect at the time so that consumers can smooth over the effects of changing the +sampling rate. When aggregating, divide each row's value by its `sample_rate`. + +See also: [`objectstore_inventory_tracker`] documentation. + # Metadata and Payload Every object consists of structured **metadata** and a binary **payload**. diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs index f1a54119..3b7070c7 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/objectstore-service/src/backend/bigtable.rs @@ -471,11 +471,15 @@ fn bumped_tti_metadata(metadata: &Metadata) -> Metadata { } /// Builds the three mutations that write an object row: clear existing data, -/// then set the payload and metadata cells. +/// then set the payload and metadata cells. Returns them with the resulting row size. /// /// Used by both [`BigTableBackend::put_row`] (unconditional write) and /// [`BigTableBackend::put_non_tombstone`] (conditional write). -fn object_mutations(mut metadata: Metadata, payload: Vec) -> Result<[v2::Mutation; 3]> { +fn object_mutations( + path: &[u8], + mut metadata: Metadata, + payload: Vec, +) -> Result<([v2::Mutation; 3], u64)> { let (family, timestamp_micros) = match metadata.time_expires { None => (FAMILY_MANUAL, -1), Some(deadline) => (FAMILY_GC, system_time_to_micros(deadline)?), @@ -487,7 +491,7 @@ fn object_mutations(mut metadata: Metadata, payload: Vec) -> Result<[v2::Mut let metadata_bytes = serde_json::to_vec(&metadata) .map_err(|cause| Error::serde("failed to serialize metadata", cause))?; - Ok([ + let mutations = [ // NB: We explicitly delete the row to clear metadata on overwrite. delete_row_mutation(), mutation(mutation::Mutation::SetCell(mutation::SetCell { @@ -502,7 +506,33 @@ fn object_mutations(mut metadata: Metadata, payload: Vec) -> Result<[v2::Mut timestamp_micros, value: metadata_bytes, })), - ]) + ]; + + let size = row_size(path, &mutations); + Ok((mutations, size)) +} + +/// Approximates the bytes a row occupies, as its key plus every cell value written. +/// +/// This function does not distinguish between object rows and tombstone rows. It does not +/// include Bigtable's own overhead. +fn row_size(path: &[u8], mutations: &[v2::Mutation]) -> u64 { + let cells: usize = mutations + .iter() + .filter_map(|m| match &m.mutation { + Some(mutation::Mutation::SetCell(cell)) => Some(cell.value.len()), + _ => None, + }) + .sum(); + + (path.len() + cells) as u64 +} + +/// The moment a row written now under `policy` is expected to be reclaimed. +/// +/// Returns `None` for [`ExpirationPolicy::Manual`], which never expires on its own. +fn expiry_from_policy(policy: ExpirationPolicy, now: SystemTime) -> Option { + policy.expires_in().map(|ttl| now + ttl) } /// Metadata carried by tombstone rows in the `t` (tombstone-meta) column. @@ -825,15 +855,17 @@ impl BigTableBackend { Ok(response.into_inner()) } + /// Writes an object row, returning the size of the row it wrote. async fn put_row( &self, path: Vec, metadata: Metadata, payload: Vec, action: &'static str, - ) -> Result { - let mutations = object_mutations(metadata, payload)?; - self.mutate(path, mutations, action).await + ) -> Result<(v2::MutateRowResponse, u64)> { + let (mutations, size) = object_mutations(&path, metadata, payload)?; + let response = self.mutate(path, mutations, action).await?; + Ok((response, size)) } async fn put_tombstone_row( @@ -849,6 +881,8 @@ impl BigTableBackend { /// Best-effort TTI bump for a row. /// /// If the payload isn't loaded, it will be fetched. Failures are ignored silently. + /// + /// A successful bump is reported to the [`ChangeStream`]. #[tracing::instrument(level = "debug", fields(?hv_id, loaded), skip_all)] async fn bump_tti(&self, path: Vec, row: &RowData, loaded: bool, hv_id: &ObjectId) { let expiration_policy = row.expiration_policy(); @@ -863,17 +897,30 @@ impl BigTableBackend { } }; + let now = SystemTime::now(); let tombstone = Tombstone { target, expiration_policy, }; - let _ = self.put_tombstone_row(path, &tombstone, "tti-bump").await; + if self + .put_tombstone_row(path, &tombstone, "tti-bump") + .await + .is_ok() + { + self.change_stream + .update(hv_id, expiry_from_policy(expiration_policy, now)); + } } RowData::Object { metadata, payload } if loaded => { let bumped = bumped_tti_metadata(metadata); - let _ = self + let expires_at = bumped.time_expires; + if self .put_row(path, bumped, payload.clone(), "tti-bump") - .await; + .await + .is_ok() + { + self.change_stream.update(hv_id, expires_at); + } } RowData::Object { metadata, .. } => { let payload_read = self @@ -882,7 +929,14 @@ impl BigTableBackend { if let Ok(Some(RowData::Object { payload, .. })) = payload_read { let bumped = bumped_tti_metadata(metadata); - let _ = self.put_row(path, bumped, payload, "tti-bump").await; + let expires_at = bumped.time_expires; + if self + .put_row(path, bumped, payload, "tti-bump") + .await + .is_ok() + { + self.change_stream.update(hv_id, expires_at); + } } } } @@ -943,8 +997,10 @@ impl Backend for BigTableBackend { payload.push(chunk); } - self.put_row(path, metadata.clone(), payload.into_bytes().into(), "put") + let (_, size) = self + .put_row(path, metadata.clone(), payload.into_bytes().into(), "put") .await?; + self.change_stream.write(id, size, metadata.time_expires); Ok(()) } @@ -975,6 +1031,7 @@ impl Backend for BigTableBackend { let path = id.as_storage_path().to_string().into_bytes(); self.mutate(path, [delete_row_mutation()], "delete").await?; + self.change_stream.delete(id); Ok(()) } @@ -996,7 +1053,7 @@ impl HighVolumeBackend for BigTableBackend { objectstore_log::debug!("Conditional put to Bigtable backend"); let path = id.as_storage_path().to_string().into_bytes(); - let mutations = object_mutations(metadata.clone(), payload.to_vec())?; + let (mutations, size) = object_mutations(&path, metadata.clone(), payload.to_vec())?; for _ in 0..CAS_RETRY_COUNT { let write_succeeded = self @@ -1009,6 +1066,7 @@ impl HighVolumeBackend for BigTableBackend { .await?; if write_succeeded { + self.change_stream.write(id, size, metadata.time_expires); return Ok(None); } @@ -1117,6 +1175,7 @@ impl HighVolumeBackend for BigTableBackend { .await?; if write_succeeded { + self.change_stream.delete(id); return Ok(None); } @@ -1163,14 +1222,44 @@ impl HighVolumeBackend for BigTableBackend { (None, None) => tombstone_predicate(), }; - let mutations = match write { - TieredWrite::Tombstone(tombstone) => tombstone_mutations(&tombstone, now)?.into(), - TieredWrite::Object(m, p) => object_mutations(m, p.to_vec())?.into(), - TieredWrite::Delete => vec![delete_row_mutation()], + // Get the correct set of mutations to apply as well as the new expiration date. + // If we're deleting something, `expires_at` is `None`. If we're writing something + // without an expiration date, `expires_at` is `Some(None)`. + let (mutations, expires_at): (Vec, Option>) = match write { + TieredWrite::Tombstone(tombstone) => ( + tombstone_mutations(&tombstone, now)?.into(), + Some(expiry_from_policy(tombstone.expiration_policy, now)), + ), + TieredWrite::Object(m, p) => { + let expires_at = m.time_expires; + let (mutations, _) = object_mutations(&path, m, p.to_vec())?; + (mutations.into(), Some(expires_at)) + } + TieredWrite::Delete => (vec![delete_row_mutation()], None), }; - self.check_and_mutate(path, predicate, mutations, "compare_and_write") - .await + let written = self + .check_and_mutate( + path.clone(), + predicate, + mutations.clone(), + "compare_and_write", + ) + .await?; + + match (written, expires_at) { + // Don't record anything if the write didn't succeed + (false, _) => {} + // We wrote something (the inner `expires_at` is `None` for manual GC) + (true, Some(expires_at)) => { + self.change_stream + .write(id, row_size(&path, &mutations), expires_at) + } + // We deleted something + (true, None) => self.change_stream.delete(id), + } + + Ok(written) } } @@ -1330,6 +1419,11 @@ mod tests { use std::collections::BTreeMap; use anyhow::Result; + #[cfg(feature = "storage-cogs")] + use objectstore_inventory_tracker::OpType; + #[cfg(feature = "storage-cogs")] + use objectstore_inventory_tracker::test_utils::DummyProducer; + use objectstore_types::scope::{Scope, Scopes}; use super::*; @@ -1356,6 +1450,20 @@ mod tests { BigTableBackend::new(test_config(), &ChangeStreamFactory::default()).await } + #[cfg(feature = "storage-cogs")] + async fn create_test_backend_with_change_stream() -> Result<(BigTableBackend, DummyProducer)> { + let (streams, producer) = crate::change_stream::dummy_factory(); + let config = BigTableConfig { + storage_cogs: Some(CostTrackerStreamConfig { + shared_resource_id: "bigtable_objectstore".into(), + sample_rate: 1.0, + }), + ..test_config() + }; + + Ok((BigTableBackend::new(config, &streams).await?, producer)) + } + fn make_id() -> ObjectId { ObjectId::random(ObjectContext { usecase: "testing".into(), @@ -1377,7 +1485,7 @@ mod tests { if metadata.time_expires.is_none() { metadata.time_expires = metadata.expiration_policy.expires_in().map(|ttl| now + ttl); } - let mutations = object_mutations(metadata, payload.to_vec())?; + let (mutations, _) = object_mutations(&path, metadata, payload.to_vec())?; backend.mutate(path, mutations, "test-setup").await?; Ok(()) } @@ -2393,4 +2501,156 @@ mod tests { Ok(()) } + + #[test] + fn row_size_counts_the_key_and_every_cell() { + let path = b"attachments/org.1/objects/abc"; + let (mutations, size) = + object_mutations(path, Metadata::default(), b"0123456789".to_vec()).unwrap(); + + // The key, the 10-byte payload, and the serialized metadata. `object_mutations` + // stamps the size into the metadata before serializing it, so the expected length + // has to account for that too. + let stamped = Metadata { + size: Some(10), + ..Default::default() + }; + let metadata_len = serde_json::to_vec(&stamped).unwrap().len(); + let expected = (path.len() + 10 + metadata_len) as u64; + + assert_eq!(row_size(path, &mutations), expected); + assert_eq!(size, expected, "the size handed back matches the mutations"); + } + + #[test] + fn row_size_is_nonzero_for_tombstones() { + let path = b"attachments/org.1/objects/abc"; + let tombstone = Tombstone { + target: ObjectId::from_storage_path("attachments/org.1/objects/abc/0199").unwrap(), + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(60)), + }; + let mutations = tombstone_mutations(&tombstone, SystemTime::now()).unwrap(); + + assert!(row_size(path, &mutations) > path.len() as u64); + } + + #[test] + fn expiry_from_policy_resolves_only_timeout_policies() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1000); + + assert_eq!(expiry_from_policy(ExpirationPolicy::Manual, now), None); + assert_eq!( + expiry_from_policy(ExpirationPolicy::TimeToLive(Duration::from_secs(30)), now), + Some(now + Duration::from_secs(30)) + ); + assert_eq!( + expiry_from_policy(ExpirationPolicy::TimeToIdle(Duration::from_secs(30)), now), + Some(now + Duration::from_secs(30)) + ); + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_writes_and_deletes() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(3600)), + time_expires: Some(SystemTime::now() + Duration::from_secs(3600)), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single::(b"hello".to_vec()), + ) + .await?; + backend.delete_object(&id).await?; + + let records = producer.records(); + assert_eq!(records.len(), 2); + + assert_eq!(records[0].op_type, OpType::Write); + assert_eq!(records[0].app_feature, "testing"); + assert_eq!(records[0].shared_resource_id, "bigtable_objectstore"); + // Key plus payload plus metadata, so strictly more than the payload alone. + assert!(records[0].size.unwrap() > b"hello".len() as u64); + assert!(records[0].expiration_time.is_some()); + + assert_eq!(records[1].op_type, OpType::Delete); + assert_eq!(records[1].size, None); + assert_eq!(records[1].record_id, records[0].record_id); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_tombstone_rows() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let target = new_test_revision(&id); + + let tombstone = Tombstone { + target: target.clone(), + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(3600)), + }; + let written = backend + .compare_and_write(&id, None, TieredWrite::Tombstone(tombstone)) + .await?; + assert!(written); + + let records = producer.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].op_type, OpType::Write); + assert!( + records[0].size.unwrap() > 0, + "tombstone rows occupy storage and must not report zero" + ); + assert!(records[0].expiration_time.is_some()); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_tti_bump_as_an_update() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(3600)), + time_expires: Some(SystemTime::now() + Duration::from_secs(1)), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single::(b"hello".to_vec()), + ) + .await?; + producer.clear(); + + // The stored deadline is far enough below `now + tti` to clear the debounce. + backend.get_tiered_object(&id, None).await?; + + let records = producer.records(); + assert_eq!(records.len(), 1, "expected exactly one bump report"); + assert_eq!(records[0].op_type, OpType::Update); + assert_eq!(records[0].size, None, "a bump does not change the size"); + assert!(records[0].expiration_time.is_some()); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + fn new_test_revision(id: &ObjectId) -> ObjectId { + ObjectId { + context: id.context.clone(), + key: format!("{}/{}", id.key, uuid::Uuid::now_v7()), + } + } } diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 6ea1cffe..58b868de 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -92,6 +92,23 @@ pub struct GcsConfig { pub storage_cogs: Option, } +/// Response header carrying the size GCS stored, in bytes. +/// +/// Differs from `Content-Length`, which describes the transfer, once an object is +/// content-encoded. +const STORED_CONTENT_LENGTH: &str = "x-goog-stored-content-length"; + +/// Reads [`STORED_CONTENT_LENGTH`] off a GCS response. +fn stored_content_length(response: &reqwest::Response) -> Option { + response + .headers() + .get(STORED_CONTENT_LENGTH)? + .to_str() + .ok()? + .parse() + .ok() +} + /// Default endpoint used to access the GCS JSON API. const DEFAULT_ENDPOINT: &str = "https://storage.googleapis.com"; /// Permission scopes required for accessing GCS. @@ -156,6 +173,16 @@ struct GcsObject { } impl GcsObject { + /// Bytes this object's custom metadata occupies, keys included. + /// + /// GCS stores metadata alongside the payload, so it counts toward an object's size. + fn metadata_size(&self) -> u64 { + self.metadata + .iter() + .map(|(key, value)| key.to_string().len() as u64 + value.len() as u64) + .sum() + } + /// Converts our Metadata type to GCS JSON object metadata. pub fn from_metadata(metadata: &Metadata) -> Self { let mut gcs_object = GcsObject { @@ -590,8 +617,15 @@ impl GcsBackend { /// Fetches the GCS object metadata (without the payload), bumps TTI if /// needed, and returns the parsed [`Metadata`]. + /// + /// `id` is only used to attribute a TTI bump to the right record in the change stream; the + /// request itself is addressed by `object_url`. #[tracing::instrument(level = "debug", fields(%object_url), skip(self))] - async fn fetch_gcs_metadata(&self, object_url: &Url) -> Result> { + async fn fetch_gcs_metadata( + &self, + id: &ObjectId, + object_url: &Url, + ) -> Result> { let metadata_opt = self .with_retry("get_metadata", || async { let resp = self @@ -639,18 +673,27 @@ impl GcsBackend { // TODO: Schedule into background persistently so this doesn't get lost on restarts if let Some(new_expire_at) = metadata.check_tti_bump(access_time) { - self.update_custom_time( - object_url.clone(), - new_expire_at, - &generation, - &metageneration, - ) - .await?; + let bumped = self + .update_custom_time( + object_url.clone(), + new_expire_at, + &generation, + &metageneration, + ) + .await?; + + // Only report a deadline that actually moved. + if bumped { + self.change_stream.update(id, Some(new_expire_at)); + } } Ok(Some(metadata)) } + /// Moves an object's `customTime`, which is what its lifecycle expiry is anchored to. + /// + /// Returns whether the update was actually applied. #[tracing::instrument(level = "debug", fields(%object_url), skip(self))] async fn update_custom_time( &self, @@ -658,7 +701,7 @@ impl GcsBackend { custom_time: SystemTime, generation: &str, metageneration: &str, - ) -> Result<()> { + ) -> Result { #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct CustomTimeRequest { @@ -684,14 +727,14 @@ impl GcsBackend { { Ok(response) => { response.drain_body().await; - Ok(()) + Ok(true) } // Bumping TTI is opportunistic. A concurrent metadata writer won the CAS race, // so leave its update intact and let a future read evaluate the TTI again. Err(Error::BackendResponse { status: StatusCode::PRECONDITION_FAILED, .. - }) => Ok(()), + }) => Ok(false), Err(error) => Err(error), } }) @@ -744,7 +787,7 @@ impl Backend for GcsBackend { ) .part( "media", - multipart::Part::stream(Body::wrap_stream(stream)) + multipart::Part::stream(Body::wrap_stream(stream.boxed())) .mime_str(&metadata.content_type) .map_err(|e| Error::Generic { context: format!("invalid mime type: {}", metadata.content_type), @@ -757,16 +800,45 @@ impl Backend for GcsBackend { // set the header *after* writing the multipart form into the request. let content_type = format!("multipart/related; boundary={}", multipart.boundary()); - self.request(Method::POST, self.upload_url(id, "multipart")?) + let response = self + .request(Method::POST, self.upload_url(id, "multipart")?) .await? .multipart(multipart) .header(header::CONTENT_TYPE, content_type) .send_traced() .await .check_error("GCS: upload object") - .await? - .drain_body() - .await; + .await?; + + // Try to get the object size from the `STORED_CONTENT_LENGTH` response header. + // If we can't, deserialize the JSON response as it contains the object resource + // which contains the object size. + let stored_size = stored_content_length(&response); + let stored_size = match stored_size { + Some(size) => { + response.drain_body().await; + Some(size) + } + None => response + .json::() + .await + .ok() + .and_then(|object| object.size) + .and_then(|size| size.parse().ok()), + }; + + match stored_size { + Some(payload_size) => { + self.change_stream.write( + id, + payload_size + gcs_metadata.metadata_size(), + metadata.time_expires, + ); + } + None => { + objectstore_metrics::count!("cost_tracker.unreported", reason = "no_stored_size"); + } + } Ok(()) } @@ -776,7 +848,7 @@ impl Backend for GcsBackend { objectstore_log::debug!("Reading from GCS backend"); let object_url = self.object_url(id)?; - let Some(metadata) = self.fetch_gcs_metadata(&object_url).await? else { + let Some(metadata) = self.fetch_gcs_metadata(id, &object_url).await? else { return Ok(None); }; @@ -842,7 +914,7 @@ impl Backend for GcsBackend { async fn get_metadata(&self, id: &ObjectId) -> Result { objectstore_log::debug!("Reading metadata from GCS backend"); let object_url = self.object_url(id)?; - self.fetch_gcs_metadata(&object_url).await + self.fetch_gcs_metadata(id, &object_url).await } #[tracing::instrument(level = "debug", skip(self))] @@ -850,28 +922,35 @@ impl Backend for GcsBackend { objectstore_log::debug!("Deleting from GCS backend"); let object_url = self.object_url(id)?; - self.with_retry("delete", || async { - let resp = self - .request(Method::DELETE, object_url.clone()) - .await? - .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: delete object", e))?; + let deleted = self + .with_retry("delete", || async { + let resp = self + .request(Method::DELETE, object_url.clone()) + .await? + .send_traced() + .await + .map_err(|e| Error::reqwest("GCS: delete object", e))?; - // Do not error for objects that do not exist - if resp.status() == StatusCode::NOT_FOUND { - resp.drain_body().await; - return Ok(()); - } + // Do not error for objects that do not exist + if resp.status() == StatusCode::NOT_FOUND { + resp.drain_body().await; + return Ok(false); + } - resp.check_error("GCS: delete object") - .await? - .drain_body() - .await; + resp.check_error("GCS: delete object") + .await? + .drain_body() + .await; - Ok(()) - }) - .await + Ok(true) + }) + .await?; + + if deleted { + self.change_stream.delete(id); + } + + Ok(()) } async fn join(&self) { @@ -1229,6 +1308,14 @@ mod tests { use anyhow::Result; use objectstore_types::scope::{Scope, Scopes}; + #[cfg(feature = "storage-cogs")] + use objectstore_inventory_tracker::OpType; + #[cfg(feature = "storage-cogs")] + use objectstore_inventory_tracker::test_utils::DummyProducer; + + #[cfg(feature = "storage-cogs")] + use crate::stream::ClientError; + use super::*; use crate::id::ObjectContext; use crate::multipart::CompletedPart; @@ -1251,6 +1338,20 @@ mod tests { GcsBackend::new(test_config(), &ChangeStreamFactory::default()).await } + #[cfg(feature = "storage-cogs")] + async fn create_test_backend_with_change_stream() -> Result<(GcsBackend, DummyProducer)> { + let (streams, producer) = crate::change_stream::dummy_factory(); + let config = GcsConfig { + storage_cogs: Some(CostTrackerStreamConfig { + shared_resource_id: "gcs_objectstore".into(), + sample_rate: 1.0, + }), + ..test_config() + }; + + Ok((GcsBackend::new(config, &streams).await?, producer)) + } + fn make_id() -> ObjectId { ObjectId::random(ObjectContext { usecase: "testing".into(), @@ -2082,4 +2183,178 @@ mod tests { Ok(()) } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_the_size_gcs_stored() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let payload = vec![b'x'; 4096]; + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(3600)), + time_expires: Some(SystemTime::now() + Duration::from_secs(3600)), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single::(payload.clone()), + ) + .await?; + + let records = producer.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].op_type, OpType::Write); + assert_eq!(records[0].shared_resource_id, "gcs_objectstore"); + assert_eq!(records[0].app_feature, "testing"); + assert_eq!( + records[0].size, + Some(payload.len() as u64 + GcsObject::from_metadata(&metadata).metadata_size()) + ); + assert!(records[0].expiration_time.is_some()); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_size_includes_metadata_keys_and_values() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let payload = b"tiny".to_vec(); + + let bare = Metadata::default(); + backend + .put_object( + &make_id(), + &bare, + stream::single::(payload.clone()), + ) + .await?; + + let annotated = Metadata { + custom: BTreeMap::from_iter([("a-fairly-long-metadata-key".into(), "value".into())]), + ..Default::default() + }; + backend + .put_object( + &make_id(), + &annotated, + stream::single::(payload.clone()), + ) + .await?; + + let records = producer.records(); + assert_eq!(records.len(), 2); + + let bare_size = records[0].size.unwrap(); + let annotated_size = records[1].size.unwrap(); + assert_eq!( + bare_size, + payload.len() as u64, + "default metadata contributes no custom keys" + ); + assert!( + annotated_size > bare_size, + "same payload, more metadata: {annotated_size} should exceed {bare_size}" + ); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_nothing_when_the_object_was_already_gone() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + + backend.delete_object(&make_id()).await?; + + assert!(producer.records().is_empty()); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_deletes() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + + backend + .put_object( + &id, + &Metadata::default(), + stream::single::(b"hi".to_vec()), + ) + .await?; + producer.clear(); + + backend.delete_object(&id).await?; + + let records = producer.records(); + assert_eq!(records.len(), 1, "a retried delete must report only once"); + assert_eq!(records[0].op_type, OpType::Delete); + assert_eq!(records[0].size, None); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_tti_bump_as_an_update() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(3600)), + time_expires: Some(SystemTime::now() + Duration::from_secs(1)), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single::(b"hi".to_vec()), + ) + .await?; + producer.clear(); + + backend.get_metadata(&id).await?; + + let records = producer.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].op_type, OpType::Update); + assert_eq!(records[0].size, None); + assert!(records[0].expiration_time.is_some()); + + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn change_stream_reports_nothing_when_tti_is_not_bumped() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let metadata = Metadata { + expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(3600)), + time_expires: Some(SystemTime::now() + Duration::from_secs(3600)), + ..Default::default() + }; + + backend + .put_object( + &id, + &metadata, + stream::single::(b"hi".to_vec()), + ) + .await?; + producer.clear(); + + backend.get_metadata(&id).await?; + + assert!(producer.records().is_empty()); + + Ok(()) + } } diff --git a/objectstore-service/src/change_stream/factory.rs b/objectstore-service/src/change_stream/factory.rs index a42a2a47..dfc33548 100644 --- a/objectstore-service/src/change_stream/factory.rs +++ b/objectstore-service/src/change_stream/factory.rs @@ -6,6 +6,8 @@ use std::sync::Arc; #[cfg(feature = "storage-cogs")] use objectstore_inventory_tracker::SharedProducer; +#[cfg(all(test, feature = "storage-cogs"))] +use objectstore_inventory_tracker::test_utils; #[cfg(feature = "storage-cogs")] use serde::{Deserialize, Serialize}; @@ -106,3 +108,73 @@ fn build_kafka_producer( } } } + +/// A [`ChangeStreamFactory`] that reports into the returned producer. +#[cfg(all(test, feature = "storage-cogs"))] +pub(crate) fn dummy_factory() -> (ChangeStreamFactory, test_utils::DummyProducer) { + use objectstore_inventory_tracker::Producer as _; + + let producer = test_utils::DummyProducer::default(); + let factory = ChangeStreamFactory { + producer: Some(producer.clone().shared()), + }; + + (factory, producer) +} + +#[cfg(all(test, feature = "storage-cogs"))] +mod tests { + use super::*; + + fn config() -> CostTrackerStreamConfig { + CostTrackerStreamConfig { + shared_resource_id: "bigtable_objectstore".into(), + sample_rate: 1.0, + } + } + + fn reports(stream: &Arc) -> bool { + !format!("{stream:?}").contains("NoopStream") + } + + #[test] + fn a_backend_without_a_change_stream_config_reports_nothing() { + let (factory, _producer) = dummy_factory(); + + assert!(!reports(&factory.build(None))); + } + + #[test] + fn a_configured_backend_without_a_transport_reports_nothing() { + let factory = ChangeStreamFactory::default(); + + assert!(!reports(&factory.build(Some(&config())))); + } + + #[test] + fn a_configured_backend_reports_through_the_transport() { + let (factory, producer) = dummy_factory(); + let stream = factory.build(Some(&config())); + + assert!(reports(&stream)); + + stream.delete(&crate::id::ObjectId::from_storage_path("attachments/objects/abc").unwrap()); + + let records = producer.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].shared_resource_id, "bigtable_objectstore"); + } + + #[test] + fn an_unusable_transport_disables_reporting_instead_of_failing() { + let factory = ChangeStreamFactory::new(&CostTrackerConfig::Kafka( + objectstore_inventory_tracker::kafka::KafkaConfig { + topic: "shared-resources-inventory".into(), + bootstrap_servers: vec!["127.0.0.1:9092".into()], + override_params: [("not.a.real.property".to_owned(), "1".to_owned())].into(), + }, + )); + + assert!(!reports(&factory.build(Some(&config())))); + } +} diff --git a/objectstore-service/src/change_stream/mod.rs b/objectstore-service/src/change_stream/mod.rs index cefd10d8..a8ad38e0 100644 --- a/objectstore-service/src/change_stream/mod.rs +++ b/objectstore-service/src/change_stream/mod.rs @@ -25,6 +25,9 @@ pub use factory::ChangeStreamFactory; #[cfg(feature = "storage-cogs")] pub use factory::CostTrackerConfig; +#[cfg(all(test, feature = "storage-cogs"))] +pub(crate) use factory::dummy_factory; + /// How long a backend waits for reported records to be handed off during shutdown. pub const FLUSH_TIMEOUT: Duration = Duration::from_secs(2);