From 28100c237a35a02fa8e393510f183c791c19aa5f Mon Sep 17 00:00:00 2001 From: Matt Hammerly Date: Mon, 10 Aug 2026 22:40:29 -0700 Subject: [PATCH] feat(cogs): report backend changes to a change stream --- objectstore-service/docs/architecture.md | 60 +++- objectstore-service/src/backend/bigtable.rs | 282 +++++++++++++++++- objectstore-service/src/backend/common.rs | 25 ++ objectstore-service/src/backend/gcs.rs | 195 +++++++++++- objectstore-service/src/backend/tiered.rs | 25 +- .../src/change_stream/factory.rs | 7 +- objectstore-service/src/change_stream/mod.rs | 3 + 7 files changed, 536 insertions(+), 61 deletions(-) diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 0b745d46..9b898535 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -108,24 +108,66 @@ 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 +[`ChangeStreamConfig`](change_stream::ChangeStreamConfig) naming at least one +listener that the service has a matching sink for. + +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, a listener such as [`KafkaStream`](change_stream::kafka::KafkaStream) +uses [`InventoryTracker`](objectstore_inventory_tracker::InventoryTracker) to +publish change events. Each listener 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 284a9dea..ce59e91f 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/objectstore-service/src/backend/bigtable.rs @@ -505,6 +505,29 @@ fn object_mutations(mut metadata: Metadata, payload: Vec) -> Result<[v2::Mut ]) } +/// 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_bytes(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. /// /// Tombstone-specific metadata evolves independently of object [`Metadata`]. Only fields @@ -763,6 +786,21 @@ impl BigTableBackend { }) } + /// Report a write operation to the [`ChangeStream`]. + /// + /// `mutations` are the ones just applied, so the reported size stays in step with + /// whatever columns a row actually writes. + fn report_write( + &self, + id: &ObjectId, + path: &[u8], + mutations: &[v2::Mutation], + expires_at: Option, + ) { + self.change_stream + .write(id, row_bytes(path, mutations), expires_at); + } + /// Reads a single row by key, returning parsed row data. /// /// Returns `None` if the row is absent or has expired. @@ -849,6 +887,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 +903,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 +935,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 +1003,10 @@ impl Backend for BigTableBackend { payload.push(chunk); } - self.put_row(path, metadata.clone(), payload.into_bytes().into(), "put") - .await?; + // Inline `put_row()` because we need the mutations to compute their size. + let mutations = object_mutations(metadata.clone(), payload.into_bytes().into())?; + self.mutate(path.clone(), mutations.clone(), "put").await?; + self.report_write(id, &path, &mutations, metadata.time_expires); Ok(()) } @@ -975,6 +1037,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(()) } @@ -1009,6 +1072,7 @@ impl HighVolumeBackend for BigTableBackend { .await?; if write_succeeded { + self.report_write(id, &path, &mutations, metadata.time_expires); return Ok(None); } @@ -1117,6 +1181,7 @@ impl HighVolumeBackend for BigTableBackend { .await?; if write_succeeded { + self.change_stream.delete(id); return Ok(None); } @@ -1163,14 +1228,40 @@ 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; + (object_mutations(m, p.to_vec())?.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.report_write(id, &path, &mutations, expires_at), + // We deleted something + (true, None) => self.change_stream.delete(id), + } + + Ok(written) } } @@ -1330,6 +1421,10 @@ mod tests { use std::collections::BTreeMap; use anyhow::Result; + use objectstore_inventory_tracker::OpType; + use objectstore_inventory_tracker::test_utils::DummyProducer; + + use crate::change_stream::{ListenerConfigs, kafka}; use objectstore_types::scope::{Scope, Scopes}; use super::*; @@ -1356,6 +1451,21 @@ mod tests { BigTableBackend::new(test_config(), &ChangeStreamFactory::default()).await } + async fn create_test_backend_with_change_stream() -> Result<(BigTableBackend, DummyProducer)> { + let (streams, producer) = crate::change_stream::dummy_factory(); + let config = BigTableConfig { + change_stream: Some(ChangeStreamConfig { + shared_resource_id: "bigtable_objectstore".into(), + listeners: ListenerConfigs { + kafka: Some(kafka::ListenerConfig { sample_rate: 1.0 }), + }, + }), + ..test_config() + }; + + Ok((BigTableBackend::new(config, &streams).await?, producer)) + } + fn make_id() -> ObjectId { ObjectId::random(ObjectContext { usecase: "testing".into(), @@ -2393,4 +2503,152 @@ mod tests { Ok(()) } + + #[test] + fn row_bytes_counts_the_key_and_every_cell() { + let path = b"attachments/org.1/objects/abc"; + let mutations = object_mutations(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(); + + assert_eq!( + row_bytes(path, &mutations), + (path.len() + 10 + metadata_len) as u64 + ); + } + + #[test] + fn row_bytes_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_bytes(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)) + ); + } + + #[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(()) + } + + #[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(()) + } + + #[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(()) + } + + 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/common.rs b/objectstore-service/src/backend/common.rs index 45b9b4a0..2e8e3a55 100644 --- a/objectstore-service/src/backend/common.rs +++ b/objectstore-service/src/backend/common.rs @@ -1,7 +1,10 @@ //! Shared trait definition and types for all backends. use std::fmt; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use futures_util::{Stream, StreamExt}; use objectstore_types::metadata::{ExpirationPolicy, Metadata}; use objectstore_types::range::{ByteRange, ContentRange}; @@ -29,6 +32,28 @@ pub type MetadataResponse = Option; /// Backend response for delete operations. pub type DeleteResponse = (); +/// Wraps a stream to count the total bytes yielded by successful chunks. +/// +/// Returns the shared counter and the wrapped stream. The counter is incremented +/// as the stream is consumed, so read it only after the stream is exhausted. +pub(crate) fn counting_stream( + stream: S, +) -> (Arc, impl Stream>) +where + S: Stream>, +{ + let counter = Arc::new(AtomicU64::new(0)); + + ( + counter.clone(), + stream.inspect(move |res| { + if let Ok(chunk) = res { + counter.fetch_add(chunk.len() as u64, Ordering::Relaxed); + } + }), + ) +} + /// Trait implemented by all storage backends. #[async_trait::async_trait] pub trait Backend: fmt::Debug + Send + Sync + 'static { diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 4fc37a8e..54533ecd 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -4,6 +4,7 @@ use std::borrow::Cow; use std::collections::BTreeMap; use std::future::Future; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::SystemTime; use std::{fmt, io}; @@ -20,7 +21,7 @@ use serde::{Deserialize, Serialize}; use super::extensions::{ResponseExt, SendTraced}; use crate::backend::common::{ self, Backend, DeleteResponse, GetResponse, MetadataResponse, MultipartUploadBackend, - PutResponse, + PutResponse, counting_stream, }; use crate::change_stream::{ ChangeStream, ChangeStreamConfig, ChangeStreamFactory, flush_change_stream, @@ -590,8 +591,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 +647,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 +675,7 @@ impl GcsBackend { custom_time: SystemTime, generation: &str, metageneration: &str, - ) -> Result<()> { + ) -> Result { #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct CustomTimeRequest { @@ -684,14 +701,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), } }) @@ -728,6 +745,11 @@ impl Backend for GcsBackend { objectstore_log::debug!("Writing to GCS backend"); let gcs_metadata = GcsObject::from_metadata(metadata); + // The payload arrives as a stream with no declared length, so the stored size is + // only known once the upload has drained. Safe from double-counting because this + // request is not retried. + let (stored_size, stream) = counting_stream(stream); + // NB: Ensure the order of these fields and that a content-type is attached to them. Both // are required by the GCS API. let metadata_json = serde_json::to_string(&gcs_metadata).map_err(|cause| Error::Serde { @@ -744,7 +766,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), @@ -768,6 +790,10 @@ impl Backend for GcsBackend { .drain_body() .await; + let stored_size = stored_size.load(Ordering::Acquire); + self.change_stream + .write(id, stored_size, metadata.time_expires); + Ok(()) } @@ -776,7 +802,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 +868,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))] @@ -871,7 +897,12 @@ impl Backend for GcsBackend { Ok(()) }) - .await + .await?; + + // Reported after the retry loop so a retried delete is reported once. + self.change_stream.delete(id); + + Ok(()) } async fn join(&self) { @@ -1229,6 +1260,13 @@ mod tests { use anyhow::Result; use objectstore_types::scope::{Scope, Scopes}; + use objectstore_inventory_tracker::OpType; + use objectstore_inventory_tracker::test_utils::DummyProducer; + + use crate::change_stream::{ListenerConfigs, kafka}; + + use crate::stream::ClientError; + use super::*; use crate::id::ObjectContext; use crate::multipart::CompletedPart; @@ -1251,6 +1289,21 @@ mod tests { GcsBackend::new(test_config(), &ChangeStreamFactory::default()).await } + async fn create_test_backend_with_change_stream() -> Result<(GcsBackend, DummyProducer)> { + let (streams, producer) = crate::change_stream::dummy_factory(); + let config = GcsConfig { + change_stream: Some(ChangeStreamConfig { + shared_resource_id: "gcs_objectstore".into(), + listeners: ListenerConfigs { + kafka: Some(kafka::ListenerConfig { 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 +2135,114 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn change_stream_reports_the_streamed_payload_size() -> 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)); + assert!(records[0].expiration_time.is_some()); + + Ok(()) + } + + #[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(()) + } + + #[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(()) + } + + #[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/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index 9887fe24..789f20c6 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -98,12 +98,12 @@ //! persisted. use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::Ordering; use std::time::{Duration, SystemTime}; use base64::Engine as _; use bytes::Bytes; -use futures_util::{Stream, StreamExt}; +use futures_util::StreamExt; use objectstore_types::metadata::Metadata; use objectstore_types::range::ByteRange; use sentry::{Hub, SentryFutureExt}; @@ -113,6 +113,7 @@ use crate::backend::changelog::{Change, ChangeGuard, ChangeLog, ChangeManager, C use crate::backend::common::{ Backend, DeleteResponse, GetResponse, HighVolumeBackend, MetadataResponse, MultipartUploadBackend, PutResponse, TieredGet, TieredMetadata, TieredWrite, Tombstone, + counting_stream, }; use crate::backend::{HighVolumeStorageConfig, MultipartUploadStorageConfig}; use crate::error::{Error, Result}; @@ -556,26 +557,6 @@ impl std::fmt::Display for BackendChoice { } } -/// Wraps a stream to count the total bytes yielded by successful chunks. -/// -/// Returns the shared counter and the wrapped stream. The counter is incremented -/// as the stream is consumed, so read it only after the stream is exhausted. -fn counting_stream(stream: S) -> (Arc, impl Stream>) -where - S: Stream>, -{ - let counter = Arc::new(AtomicU64::new(0)); - - ( - counter.clone(), - stream.inspect(move |res| { - if let Ok(chunk) = res { - counter.fetch_add(chunk.len() as u64, Ordering::Relaxed); - } - }), - ) -} - /// The multipart upload state for TieredStorage. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] struct TieredUploadId { diff --git a/objectstore-service/src/change_stream/factory.rs b/objectstore-service/src/change_stream/factory.rs index f498cd8b..0d71fac3 100644 --- a/objectstore-service/src/change_stream/factory.rs +++ b/objectstore-service/src/change_stream/factory.rs @@ -8,7 +8,7 @@ use objectstore_inventory_tracker::SharedProducer; use objectstore_inventory_tracker::test_utils; use super::kafka::KafkaStream; -use super::{ChangeStream, ChangeStreamConfig, NoopStream, SinkConfigs, kafka}; +use super::{ChangeStream, ChangeStreamConfig, ListenerConfigs, NoopStream, SinkConfigs, kafka}; /// Builds the [`ChangeStream`] a backend reports to. /// @@ -35,7 +35,10 @@ impl ChangeStreamFactory { }; let shared_resource_id = &config.shared_resource_id; - let Some(listener) = &config.listeners.kafka else { + // Destructured so that adding a listener type has to be handled here. + let ListenerConfigs { kafka } = &config.listeners; + + let Some(listener) = kafka else { objectstore_log::warn!( shared_resource_id, "change stream has no listeners; this backend will report nothing" diff --git a/objectstore-service/src/change_stream/mod.rs b/objectstore-service/src/change_stream/mod.rs index 6024ab36..8588f868 100644 --- a/objectstore-service/src/change_stream/mod.rs +++ b/objectstore-service/src/change_stream/mod.rs @@ -25,6 +25,9 @@ mod factory; pub use factory::ChangeStreamFactory; +#[cfg(test)] +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);