diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 10f602a4..b5ee5632 100644 --- a/.github/workflows/build.yml +++ b/.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 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 20e5b103..3afdba09 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/Cargo.lock b/Cargo.lock index b6c731a5..f64cda66 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2946,6 +2946,7 @@ dependencies = [ "gcp_auth", "humantime", "humantime-serde", + "objectstore-inventory-tracker", "objectstore-log", "objectstore-metrics", "objectstore-types", diff --git a/Dockerfile.cross b/Dockerfile.cross index f4ee4cf9..7ffe2e99 100644 --- a/Dockerfile.cross +++ b/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 diff --git a/bigtable-bench/src/main.rs b/bigtable-bench/src/main.rs index d693c93c..d08baa72 100644 --- a/bigtable-bench/src/main.rs +++ b/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")?; diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index 1a1dc8c1..111e2292 100644 --- a/objectstore-server/src/state.rs +++ b/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) diff --git a/objectstore-service/Cargo.toml b/objectstore-service/Cargo.toml index df6b771a..b946a728 100644 --- a/objectstore-service/Cargo.toml +++ b/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 } diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs index 016272c9..f1a54119 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/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, + + /// 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, } /// 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, } 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 { + pub async fn new( + config: BigTableConfig, + streams: &ChangeStreamFactory, + ) -> anyhow::Result { 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::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::new(test_config(), &ChangeStreamFactory::default()).await } fn make_id() -> ObjectId { diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index a5b370cb..6ea1cffe 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/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, } /// Default endpoint used to access the GCS JSON API. @@ -443,12 +460,21 @@ pub struct GcsBackend { endpoint: Url, bucket: String, token_provider: Option, + + /// Records each write/update/delete performed by the backend. May be + /// [`NoopStream`](crate::change_stream::NoopStream). + change_stream: Arc, } impl GcsBackend { /// Creates an authenticated GCS JSON API backend bound to the bucket in `config`. - pub async fn new(config: GcsConfig) -> anyhow::Result { - let GcsConfig { endpoint, bucket } = config; + pub async fn new(config: GcsConfig, streams: &ChangeStreamFactory) -> anyhow::Result { + 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::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::new(test_config(), &ChangeStreamFactory::default()).await } fn make_id() -> ObjectId { diff --git a/objectstore-service/src/backend/mod.rs b/objectstore-service/src/backend/mod.rs index 4d265139..a9c62d28 100644 --- a/objectstore-service/src/backend/mod.rs +++ b/objectstore-service/src/backend/mod.rs @@ -11,6 +11,8 @@ use anyhow::Result; use serde::{Deserialize, Serialize}; +use crate::change_stream::ChangeStreamFactory; + pub mod bigtable; pub mod changelog; pub mod common; @@ -58,29 +60,38 @@ pub enum StorageConfig { } /// Constructs a type-erased [`Backend`](common::Backend) from the given [`StorageConfig`]. -pub async fn from_config(config: StorageConfig) -> Result> { +/// +/// Backends that configured a [`ChangeStream`](crate::change_stream::ChangeStream) get one +/// from `streams`; the rest report nothing. +pub async fn from_config( + config: StorageConfig, + streams: &ChangeStreamFactory, +) -> Result> { Ok(match config { StorageConfig::Tiered(c) => { - let hv = hv_from_config(c.high_volume).await?; - let lt = lt_from_config(c.long_term).await?; + let hv = hv_from_config(c.high_volume, streams).await?; + let lt = lt_from_config(c.long_term, streams).await?; let log = Box::new(changelog::NoopChangeLog); Box::new(tiered::TieredStorage::new(hv, lt, log)) } // All non-Tiered variants are handled by from_leaf_config. A wildcard // is intentional here: any new leaf variant should fall through to // from_leaf_config, which will handle it or produce a compile error. - _ => from_leaf_config(config).await?, + _ => from_leaf_config(config, streams).await?, }) } -async fn from_leaf_config(config: StorageConfig) -> Result> { +async fn from_leaf_config( + config: StorageConfig, + streams: &ChangeStreamFactory, +) -> Result> { Ok(match config { StorageConfig::FileSystem(c) => Box::new(local_fs::LocalFsBackend::new(c)), StorageConfig::S3Compatible(c) => { Box::new(s3_compatible::S3CompatibleBackend::without_token(c)) } - StorageConfig::Gcs(c) => Box::new(gcs::GcsBackend::new(c).await?), - StorageConfig::BigTable(c) => Box::new(bigtable::BigTableBackend::new(c).await?), + StorageConfig::Gcs(c) => Box::new(gcs::GcsBackend::new(c, streams).await?), + StorageConfig::BigTable(c) => Box::new(bigtable::BigTableBackend::new(c, streams).await?), StorageConfig::Tiered(_) => anyhow::bail!("nested tiered storage is not supported"), }) } @@ -101,9 +112,12 @@ pub enum HighVolumeStorageConfig { /// Constructs a type-erased [`common::HighVolumeBackend`] from the given config. async fn hv_from_config( config: HighVolumeStorageConfig, + streams: &ChangeStreamFactory, ) -> Result> { Ok(match config { - HighVolumeStorageConfig::BigTable(c) => Box::new(bigtable::BigTableBackend::new(c).await?), + HighVolumeStorageConfig::BigTable(c) => { + Box::new(bigtable::BigTableBackend::new(c, streams).await?) + } }) } @@ -125,9 +139,10 @@ pub enum MultipartUploadStorageConfig { /// Constructs a type-erased [`common::MultipartUploadBackend`] from the given config. async fn lt_from_config( config: MultipartUploadStorageConfig, + streams: &ChangeStreamFactory, ) -> Result> { Ok(match config { MultipartUploadStorageConfig::FileSystem(c) => Box::new(local_fs::LocalFsBackend::new(c)), - MultipartUploadStorageConfig::Gcs(c) => Box::new(gcs::GcsBackend::new(c).await?), + MultipartUploadStorageConfig::Gcs(c) => Box::new(gcs::GcsBackend::new(c, streams).await?), }) } diff --git a/objectstore-service/src/change_stream/cost_tracker.rs b/objectstore-service/src/change_stream/cost_tracker.rs new file mode 100644 index 00000000..51dab0e8 --- /dev/null +++ b/objectstore-service/src/change_stream/cost_tracker.rs @@ -0,0 +1,263 @@ +//! The change stream implementation that reports through an [`InventoryTracker`]. + +use std::fmt; +use std::time::{Duration, SystemTime}; + +use objectstore_inventory_tracker::{BoxError, InventoryTracker, Producer}; + +use super::{ChangeStream, CostTrackerStreamConfig, SCOPE_ORGANIZATION, SCOPE_PROJECT, scope_id}; +use crate::id::ObjectId; + +/// Reports through an [`InventoryTracker`], which hashes each [`ObjectId`] both to +/// anonymize it and to decide whether it is sampled. See [`objectstore_inventory_tracker`] +/// for the record format. +/// +/// Logs, counts, and swallows errors returned by the [`InventoryTracker`]. +pub struct CostTrackerStream { + tracker: InventoryTracker

, +} + +impl CostTrackerStream

{ + /// Reports changes through `producer`, as described by `config`. + pub fn new(producer: P, config: &CostTrackerStreamConfig) -> Self { + Self { + tracker: InventoryTracker::new( + producer, + &config.shared_resource_id, + config.sample_rate, + ), + } + } + + /// Counts and logs a failed report, then returns. + /// + /// Note: success here doesn't necessarily mean that a message was emitted. The + /// tracker's sampling policy may have filtered an object out and chosen to emit + /// nothing. Or, a message may have been enqueued only to fail later. Those are + /// counted by the transport's delivery failure callback instead. + fn swallow(&self, op: &'static str, result: Result<(), P::Error>) + where + P::Error: Into, + { + if let Err(error) = result { + // Boxed because `SharedProducer` reports a `Box`, which std does + // not implement `Error` for. + let error: BoxError = error.into(); + // Records are dropped rather than retried + objectstore_metrics::count!( + "cost_tracker.dropped" += 1, + shared_resource_id = self.tracker.shared_resource_id().to_owned(), + op = op, + ); + objectstore_log::warn!(!!&*error, op, "failed to publish change stream record"); + } + } +} + +impl fmt::Debug for CostTrackerStream

{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CostTrackerStream") + .field("shared_resource_id", &self.tracker.shared_resource_id()) + .field("sample_rate", &self.tracker.sample_rate()) + .finish() + } +} + +#[async_trait::async_trait] +impl

ChangeStream for CostTrackerStream

+where + P: Producer + Clone + Send + Sync + 'static, + P::Error: Into + Send + 'static, +{ + fn write(&self, id: &ObjectId, size: u64, expires_at: Option) { + let result = self.tracker.write( + &id.as_storage_path().to_string(), + id.usecase(), + size, + SystemTime::now(), + expires_at, + scope_id(id, SCOPE_ORGANIZATION), + scope_id(id, SCOPE_PROJECT), + ); + self.swallow("write", result); + } + + fn update(&self, id: &ObjectId, expires_at: Option) { + let result = self.tracker.update( + &id.as_storage_path().to_string(), + id.usecase(), + SystemTime::now(), + expires_at, + scope_id(id, SCOPE_ORGANIZATION), + scope_id(id, SCOPE_PROJECT), + ); + self.swallow("update", result); + } + + fn delete(&self, id: &ObjectId) { + let result = self.tracker.delete( + &id.as_storage_path().to_string(), + id.usecase(), + SystemTime::now(), + ); + self.swallow("delete", result); + } + + async fn join(&self, timeout: Duration) { + self.swallow("join", self.tracker.join(timeout).await); + } +} + +#[cfg(test)] +mod tests { + use objectstore_inventory_tracker::OpType; + use objectstore_inventory_tracker::test_utils::DummyProducer; + + use super::*; + + fn object_id(path: &str) -> ObjectId { + ObjectId::from_storage_path(path).expect("valid storage path") + } + + fn stream(sample_rate: f64) -> (DummyProducer, CostTrackerStream) { + let producer = DummyProducer::default(); + let stream = CostTrackerStream::new( + producer.clone(), + &CostTrackerStreamConfig { + shared_resource_id: "bigtable_objectstore".into(), + sample_rate, + }, + ); + (producer, stream) + } + + #[test] + fn usecase_and_scopes_are_extracted_from_the_id() { + let (producer, stream) = stream(1.0); + let id = object_id("attachments/org.17/project.42/objects/abc"); + + stream.write(&id, 4096, None); + + let record = &producer.records()[0]; + assert_eq!(record.shared_resource_id, "bigtable_objectstore"); + assert_eq!(record.app_feature, "attachments"); + assert_eq!(record.organization_id, Some(17)); + assert_eq!(record.project_id, Some(42)); + assert_eq!(record.size, Some(4096)); + assert_eq!(record.op_type, OpType::Write); + } + + #[test] + fn the_storage_path_is_not_emitted() { + let (producer, stream) = stream(1.0); + let id = object_id("attachments/org.17/project.42/objects/abc"); + + stream.write(&id, 4096, None); + + let record_id = &producer.records()[0].record_id; + assert_ne!(record_id, &id.as_storage_path().to_string()); + assert!(!record_id.contains("attachments")); + } + + #[test] + fn missing_or_unparseable_scopes_are_reported_as_absent() { + let (producer, stream) = stream(1.0); + + for path in [ + "attachments/objects/abc", + "attachments/organization.17/objects/abc", + "attachments/org.not-a-number/project.42/objects/abc", + ] { + stream.write(&object_id(path), 1, None); + } + + let records = producer.records(); + assert_eq!( + records.len(), + 3, + "every object is reported regardless of scopes" + ); + assert_eq!(records[0].organization_id, None, "no scopes at all"); + assert_eq!( + records[1].organization_id, None, + "wrong scope key is not org" + ); + assert_eq!( + records[2].organization_id, None, + "unparseable org is absent" + ); + assert_eq!( + records[2].project_id, + Some(42), + "but project still resolves" + ); + for record in &records { + assert_eq!(record.app_feature, "attachments"); + } + } + + #[test] + fn every_operation_on_an_object_reports_the_same_record() { + let (producer, stream) = stream(1.0); + let id = object_id("attachments/org.1/project.2/objects/abc"); + + stream.write(&id, 10, None); + stream.update(&id, Some(SystemTime::now())); + stream.delete(&id); + + let records = producer.records(); + assert_eq!(records.len(), 3); + assert_eq!(records[0].record_id, records[1].record_id); + assert_eq!(records[1].record_id, records[2].record_id); + } + + #[test] + fn distinct_revisions_are_distinct_records() { + let (producer, stream) = stream(1.0); + + stream.write( + &object_id("attachments/org.1/project.2/objects/abc/0199aaaa"), + 1, + None, + ); + stream.write( + &object_id("attachments/org.1/project.2/objects/abc/0199bbbb"), + 1, + None, + ); + + let records = producer.records(); + assert_ne!(records[0].record_id, records[1].record_id); + } + + #[test] + fn update_omits_size_and_delete_omits_everything_optional() { + let (producer, stream) = stream(1.0); + let id = object_id("attachments/org.1/project.2/objects/abc"); + + stream.update(&id, Some(SystemTime::now())); + stream.delete(&id); + + let records = producer.records(); + assert_eq!(records[0].op_type, OpType::Update); + assert_eq!(records[0].size, None); + 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].expiration_time, None); + } + + #[test] + fn a_listener_sampled_at_zero_reports_nothing() { + let (producer, stream) = stream(0.0); + + for i in 0..100 { + let id = object_id(&format!("attachments/org.1/project.2/objects/{i}")); + stream.write(&id, 1, None); + stream.update(&id, None); + stream.delete(&id); + } + + assert!(producer.records().is_empty()); + } +} diff --git a/objectstore-service/src/change_stream/factory.rs b/objectstore-service/src/change_stream/factory.rs new file mode 100644 index 00000000..a42a2a47 --- /dev/null +++ b/objectstore-service/src/change_stream/factory.rs @@ -0,0 +1,108 @@ +//! Constructs the [`ChangeStream`] implementation(s) for a backend based on the available +//! service-wide sink config and per-backend stream config. + +use std::fmt; +use std::sync::Arc; + +#[cfg(feature = "storage-cogs")] +use objectstore_inventory_tracker::SharedProducer; +#[cfg(feature = "storage-cogs")] +use serde::{Deserialize, Serialize}; + +#[cfg(feature = "storage-cogs")] +use super::CostTrackerStream; +use super::{ChangeStream, CostTrackerStreamConfig, NoopStream}; + +/// Where every backend's change stream records are carried for cost tracking. +/// +/// Service-wide: a transport owns connections and a send queue worth sharing. +#[cfg(feature = "storage-cogs")] +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(tag = "type", rename_all = "lowercase")] +pub enum CostTrackerConfig { + /// Reports onto a Kafka topic. + Kafka(objectstore_inventory_tracker::kafka::KafkaConfig), +} + +/// Builds the [`ChangeStream`](s) a backend reports to. +/// +/// Without a usable transport every backend gets a [`NoopStream`]. +#[derive(Clone, Default)] +pub struct ChangeStreamFactory { + #[cfg(feature = "storage-cogs")] + producer: Option, +} + +impl ChangeStreamFactory { + /// Builds the transport described by `config`. + /// + /// Fails open: an unusable transport is logged, not fatal. + #[cfg(feature = "storage-cogs")] + pub fn new(config: &CostTrackerConfig) -> Self { + let CostTrackerConfig::Kafka(kafka) = config; + Self { + producer: build_kafka_producer(kafka), + } + } + + /// Builds the stream `config` asks for, or a [`NoopStream`] if it cannot be built. + #[cfg(feature = "storage-cogs")] + pub fn build(&self, config: Option<&CostTrackerStreamConfig>) -> Arc { + let Some(config) = config else { + return Arc::new(NoopStream); + }; + + let Some(producer) = self.producer.clone() else { + objectstore_log::error!( + shared_resource_id = config.shared_resource_id, + "backend configured a change stream but the service has no usable \ + transport; this backend will report nothing" + ); + return Arc::new(NoopStream); + }; + + Arc::new(CostTrackerStream::new(producer, config)) + } + + /// Reporting is not compiled in, so every backend reports nothing. + #[cfg(not(feature = "storage-cogs"))] + pub fn build(&self, _config: Option<&CostTrackerStreamConfig>) -> Arc { + Arc::new(NoopStream) + } +} + +impl fmt::Debug for ChangeStreamFactory { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut f = f.debug_struct("ChangeStreamFactory"); + #[cfg(feature = "storage-cogs")] + f.field("producer", &self.producer.is_some()); + f.finish() + } +} + +/// Creates the shared Kafka producer, or logs why there will be no reporting. +#[cfg(feature = "storage-cogs")] +fn build_kafka_producer( + config: &objectstore_inventory_tracker::kafka::KafkaConfig, +) -> Option { + use objectstore_inventory_tracker::kafka::KafkaProducer; + use objectstore_inventory_tracker::Producer as _; + + // Delivery is asynchronous, so an accepted record can still fail to arrive. Without + // this those show up only as a shortfall in the downstream data. + let on_delivery_failure = Box::new(|_: &_| { + objectstore_metrics::count!("cost_tracker.undelivered" += 1); + }); + + match KafkaProducer::try_new(config.clone(), Some(on_delivery_failure)) { + Ok(producer) => Some(producer.shared()), + Err(error) => { + objectstore_log::error!( + !!&error, + "failed to create the change stream kafka producer; \ + backends with a change stream will report nothing" + ); + None + } + } +} diff --git a/objectstore-service/src/change_stream/mod.rs b/objectstore-service/src/change_stream/mod.rs new file mode 100644 index 00000000..cefd10d8 --- /dev/null +++ b/objectstore-service/src/change_stream/mod.rs @@ -0,0 +1,124 @@ +//! The change stream each storage backend publishes. +//! +//! A backend describes its cost-tracking reporting with a [`CostTrackerStreamConfig`]; +//! the service describes where those records go with a [`CostTrackerConfig`], shared by +//! every backend. [`ChangeStreamFactory`] pairs the two into a [`ChangeStream`]. +//! +//! Behind the `storage-cogs` feature. Without it every backend gets a [`NoopStream`] and +//! the transport is left out of the binary. + +use std::fmt; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; + +use serde::{Deserialize, Serialize}; + +use crate::id::ObjectId; + +#[cfg(feature = "storage-cogs")] +mod cost_tracker; +mod factory; + +#[cfg(feature = "storage-cogs")] +pub use cost_tracker::CostTrackerStream; +pub use factory::ChangeStreamFactory; +#[cfg(feature = "storage-cogs")] +pub use factory::CostTrackerConfig; + +/// How long a backend waits for reported records to be handed off during shutdown. +pub const FLUSH_TIMEOUT: Duration = Duration::from_secs(2); + +/// Scope key holding the Sentry organization ID. +#[cfg(feature = "storage-cogs")] +const SCOPE_ORGANIZATION: &str = "org"; +/// Scope key holding the Sentry project ID. +#[cfg(feature = "storage-cogs")] +const SCOPE_PROJECT: &str = "project"; + +/// What a single backend reports for cost tracking, and how much of it. +/// +/// A backend without one reports nothing. +/// +/// # Example +/// +/// ```yaml +/// storage_cogs: +/// shared_resource_id: bigtable_objectstore +/// sample_rate: 1.0 +/// ``` +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct CostTrackerStreamConfig { + /// Identifies the storage backend resource. + /// + /// This is meant to correspond to a `shared_resource_id` label on a provisioned + /// storage resource so that change stream data can be joined with other data about + /// the storage resource. + pub shared_resource_id: String, + + /// Proportion of records to report, in `[0, 1]`. + /// + /// `1.0` reports every change. It can be lowered if the stream is under too much load + /// but beware: when the sample rate decreases, records that used to be tracked will + /// no longer be tracked. Stream consumers may have inconsistent state for them until + /// they expire. + #[serde(default = "default_sample_rate")] + pub sample_rate: f64, +} + +/// Reports everything by default. +fn default_sample_rate() -> f64 { + 1.0 +} + +/// Publishes the changes a single backend makes to the objects it stores. +/// +/// See [module docs](self). +#[async_trait::async_trait] +pub trait ChangeStream: fmt::Debug + Send + Sync + 'static { + /// Reports that `id` now occupies `size` bytes. Used for new writes and overwrites. + fn write(&self, id: &ObjectId, size: u64, expires_at: Option); + + /// Reports that `id`'s expiration moved, with its stored size unchanged. + fn update(&self, id: &ObjectId, expires_at: Option); + + /// Reports that `id` was deleted explicitly. Does not account for automatic GC. + fn delete(&self, id: &ObjectId); + + /// Blocks until reported records have been delivered, or `timeout` elapses. + /// + /// Call this during shutdown to drain the change stream queue. + /// Waits for reported records to be delivered, or until `timeout` elapses. + /// + /// Awaited from [`Backend::join`](crate::backend::common::Backend::join) so records + /// reported just before shutdown are not lost. + async fn join(&self, timeout: Duration); +} + +/// Drains `change_stream`, bounded by [`FLUSH_TIMEOUT`]. +/// +/// Backends call this from [`Backend::join`](crate::backend::common::Backend::join) so +/// records reported just before shutdown are not silently lost. +pub async fn flush_change_stream(change_stream: &Arc) { + change_stream.join(FLUSH_TIMEOUT).await; +} + +/// A [`ChangeStream`] that reports nothing. +#[derive(Clone, Copy, Debug, Default)] +pub struct NoopStream; + +#[async_trait::async_trait] +impl ChangeStream for NoopStream { + fn write(&self, _id: &ObjectId, _size: u64, _expires_at: Option) {} + + fn update(&self, _id: &ObjectId, _expires_at: Option) {} + + fn delete(&self, _id: &ObjectId) {} + + async fn join(&self, _timeout: Duration) {} +} + +/// Reads a scope value off `id` as an integer, if present and well-formed. +#[cfg(feature = "storage-cogs")] +fn scope_id(id: &ObjectId, scope: &str) -> Option { + id.scopes().get_value(scope)?.parse().ok() +} diff --git a/objectstore-service/src/lib.rs b/objectstore-service/src/lib.rs index e33a2f7c..a7a82b6d 100644 --- a/objectstore-service/src/lib.rs +++ b/objectstore-service/src/lib.rs @@ -3,6 +3,7 @@ #![warn(missing_debug_implementations)] pub mod backend; +pub mod change_stream; pub mod concurrency; pub mod error; mod gcp_auth; diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 0953fcd8..0684a317 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -380,6 +380,7 @@ mod tests { use crate::backend::in_memory::InMemoryBackend; use crate::backend::testing::{Hooks, TestBackend}; use crate::backend::tiered::TieredStorage; + use crate::change_stream::ChangeStreamFactory; use crate::error::Error; use crate::stream::{self, ClientStream}; @@ -436,9 +437,12 @@ mod tests { let config = GcsConfig { endpoint: Some("http://localhost:8087".into()), bucket: "test-bucket".into(), // aligned with the env var in devservices and CI + storage_cogs: None, }; - let backend = GcsBackend::new(config).await.unwrap(); + let backend = GcsBackend::new(config, &ChangeStreamFactory::default()) + .await + .unwrap(); let service = StorageService::new(Box::new(backend)); let key = service @@ -465,19 +469,31 @@ mod tests { instance_name: "objectstore".into(), table_name: "objectstore".into(), connections: None, + storage_cogs: None, }; let gcs_config = GcsConfig { endpoint: Some("http://localhost:8087".into()), bucket: "test-bucket".into(), + storage_cogs: None, }; - let high_volume = Box::new(BigTableBackend::new(bigtable_config).await.unwrap()); - let long_term = Box::new(GcsBackend::new(gcs_config.clone()).await.unwrap()); + let high_volume = Box::new( + BigTableBackend::new(bigtable_config, &ChangeStreamFactory::default()) + .await + .unwrap(), + ); + let long_term = Box::new( + GcsBackend::new(gcs_config.clone(), &ChangeStreamFactory::default()) + .await + .unwrap(), + ); let backend = TieredStorage::new(high_volume, long_term, Box::new(NoopChangeLog)); let service = StorageService::new(Box::new(backend)); // A separate GCS backend to directly inspect the long-term storage. - let gcs_backend = GcsBackend::new(gcs_config.clone()).await.unwrap(); + let gcs_backend = GcsBackend::new(gcs_config.clone(), &ChangeStreamFactory::default()) + .await + .unwrap(); // Insert a >1 MiB object with a key. This forces the long-term path: // the real payload goes to GCS, and a redirect tombstone is written to BigTable.