From 5d64b3c2ad39c990e20309c6df4b36d698636326 Mon Sep 17 00:00:00 2001 From: Janmejay Singh Date: Mon, 24 Aug 2026 23:38:45 +0530 Subject: [PATCH 1/4] Feat[OBE-11872] - Support decompression of data in SCol --- Cargo.lock | 4 ++++ Cargo.toml | 2 +- lib/observo/private | 2 +- lib/observo/scol/Cargo.toml | 4 ++++ src/sources/scol/mod.rs | 2 +- 5 files changed, 11 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index da7c93293..040a17d98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10534,6 +10534,7 @@ dependencies = [ "arc-swap", "assert_matches", "async-channel 2.5.0", + "async-compression", "async-recursion", "async-trait", "aws-sdk-s3", @@ -10546,6 +10547,7 @@ dependencies = [ "chrono", "cron", "dashmap 5.5.3", + "flate2", "futures 0.3.31", "futures-util", "http 0.2.9", @@ -10580,6 +10582,7 @@ dependencies = [ "serde_with", "smallvec", "snafu 0.7.5", + "snap", "stream-cancel", "tempfile", "test-case", @@ -10599,6 +10602,7 @@ dependencies = [ "vector-core", "vector-lib", "vrl", + "zstd 0.13.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 5930cb3aa..bc6be071d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -205,7 +205,7 @@ regex = { version = "1.11.1", default-features = false, features = ["std", "perf tokio-util = { version = "0.7", default-features = false, features = ["io", "time"] } bytesize = { version = "1.3.0", default-features = false } bytes = { version = "1.9.0", default-features = false, features = ["serde"] } -async-compression = { version = "0.4.18", default-features = false, features = ["tokio", "gzip", "zstd"] } +async-compression = { version = "0.4.18", default-features = false, features = ["tokio", "gzip", "zstd", "zlib", "deflate"] } object_store = { version = "0.10.2", features = ["gcp"], default-features = false } once_cell = { version = "1.20.2", default-features = false } smallvec = { version = "1", default-features = false, features = ["union", "serde"] } diff --git a/lib/observo/private b/lib/observo/private index bbdd4686b..051834733 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit bbdd4686bf16dc530637b621333bcf2cd1106598 +Subproject commit 05183473350a6a788d5fcaebb57a90026b363212 diff --git a/lib/observo/scol/Cargo.toml b/lib/observo/scol/Cargo.toml index 1378c130a..7283ff5d3 100644 --- a/lib/observo/scol/Cargo.toml +++ b/lib/observo/scol/Cargo.toml @@ -66,6 +66,8 @@ reqwest = { version = "0.12", default-features = false } # core). Pull hyper 1 under an alias so retry classification can downcast `object_store::Error` # sources to the *matching* `hyper::Error` type (see objstore::is_retriable). hyper1 = { package = "hyper", version = "1", default-features = false } +async-compression = { workspace = true } +snap.workspace = true # Optional dependencies for testing (below) testcontainers = { version = "0.19", optional = true } @@ -96,6 +98,8 @@ test-case = { version = "3.3" } tokio = { version = "1.43.0", features = ["test-util"] } scol = { path = ".", features = ["test-scenarios"] } chkpts = { path = "../chkpts" } +flate2 = { workspace = true } +zstd = { workspace = true } [features] test-scenarios = ["vector-lib/test", "dep:textwrap", "dep:rstest", "dep:oxide-auth", diff --git a/src/sources/scol/mod.rs b/src/sources/scol/mod.rs index 1e12ef700..e822b59c3 100644 --- a/src/sources/scol/mod.rs +++ b/src/sources/scol/mod.rs @@ -23,7 +23,7 @@ impl SourceConfig for Config { let chkptr = cx.checkpoint_accessor().await; let src = self .clone() - .build_source(cx.out, cx.shutdown, chkptr, lns) + .build_source(cx.out, cx.shutdown, chkptr, lns, cx.globals.limits.compression) .map(|r| match r { Ok(_) => Ok(()), Err(e) => { From ebecf5c28ccf926e1f714b0e4884487eae9ebdad Mon Sep 17 00:00:00 2001 From: Janmejay Singh Date: Tue, 25 Aug 2026 15:28:59 +0530 Subject: [PATCH 2/4] Feat[OBE-11872] - Bump private --- Cargo.lock | 1 + Cargo.toml | 1 + lib/observo/private | 2 +- lib/observo/scol/Cargo.toml | 1 + 4 files changed, 4 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 040a17d98..5723d3bdb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10580,6 +10580,7 @@ dependencies = [ "serde_json", "serde_urlencoded", "serde_with", + "serde_yaml 0.9.34+deprecated", "smallvec", "snafu 0.7.5", "snap", diff --git a/Cargo.toml b/Cargo.toml index bc6be071d..a05be1d16 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -178,6 +178,7 @@ rand = { version = "0.8.5", default-features = false, features = ["small_rng"] } serde_json = { version = "=1.0.140", default-features = false, features = ["raw_value", "std"] } serde = { version = "1.0.215", default-features = false, features = ["alloc", "derive", "rc"] } serde_with = { version = "3.12.0", default-features = false, features = ["macros", "std"] } +serde_yaml = { version = "0.9.34", default-features = false } snafu = { version = "0.7.5", default-features = false, features = ["futures", "std"] } rusqlite = { version = "0.37.0", features = [ "bundled" ] } tokio = { version = "1.43.0", default-features = false, features = ["full"] } diff --git a/lib/observo/private b/lib/observo/private index 051834733..dcecfe883 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit 05183473350a6a788d5fcaebb57a90026b363212 +Subproject commit dcecfe883aa96169e8d499e049849b34e5554711 diff --git a/lib/observo/scol/Cargo.toml b/lib/observo/scol/Cargo.toml index 7283ff5d3..24547e4e1 100644 --- a/lib/observo/scol/Cargo.toml +++ b/lib/observo/scol/Cargo.toml @@ -94,6 +94,7 @@ regex = "1" serde_urlencoded = "0.7" tempfile = "3.15.0" tracing-test = { version = "0.2.5", features = [ "no-env-filter" ] } +serde_yaml = { workspace = true } test-case = { version = "3.3" } tokio = { version = "1.43.0", features = ["test-util"] } scol = { path = ".", features = ["test-scenarios"] } From 61eef6d93199ae70f236655709c668ec031ddd63 Mon Sep 17 00:00:00 2001 From: Janmejay Singh Date: Wed, 26 Aug 2026 23:04:54 +0530 Subject: [PATCH 3/4] Feat[OBE-11941] - Add support for checkpoint compare-and-set on condition: absent --- lib/observo/private | 2 +- lib/vector-common/src/chkpts.rs | 20 +++++++++++++++++++- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/lib/observo/private b/lib/observo/private index dcecfe883..bc3a899cd 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit dcecfe883aa96169e8d499e049849b34e5554711 +Subproject commit bc3a899cdab3bafd2a486acae092ca188d8affd5 diff --git a/lib/vector-common/src/chkpts.rs b/lib/vector-common/src/chkpts.rs index ab507fdc6..b1db07346 100644 --- a/lib/vector-common/src/chkpts.rs +++ b/lib/vector-common/src/chkpts.rs @@ -39,6 +39,24 @@ pub enum ChkptErr { Unknown(crate::Error) } +/// Precondition guarding a `compare_and_set`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SetCond<'a> { + /// The checkpoint must exist and its latest value must equal this. + Old(Cow<'a, str>), + /// The checkpoint must not exist. + Absent, +} + +impl SetCond<'_> { + pub fn into_owned(self) -> SetCond<'static> { + match self { + SetCond::Old(v) => SetCond::Old(Cow::Owned(v.into_owned())), + SetCond::Absent => SetCond::Absent, + } + } +} + impl Display for ChkptErr { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{self:?}") @@ -58,7 +76,7 @@ impl Error for ChkptErr { pub trait Accessor: Send + dyn_clone::DynClone + Sync { async fn get(&self, id: Cow<'_, str>) -> Result; async fn set(&self, id: Cow<'_, str>, value: Cow<'_, str>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>; - async fn compare_and_set(&self, id: Cow<'_, str>, value: Cow<'_, str>, if_old: Cow<'_, str>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>; + async fn compare_and_set(&self, id: Cow<'_, str>, value: Cow<'_, str>, cond: SetCond<'_>, ctx: Cow<'_, str>) -> Result<(), ChkptErr>; async fn del(&self, id: Cow<'_, str>) -> Result<(), ChkptErr>; async fn compare_and_del(&self, id: Cow<'_, str>, if_old: Cow<'_, str>) -> Result<(), ChkptErr>; async fn del_range(&self, from: Cow<'_, str>, to: Cow<'_, str>) -> Result; From 550545a63efc7bfc8473fb6fae6c9262eb101be8 Mon Sep 17 00:00:00 2001 From: Janmejay Singh Date: Thu, 27 Aug 2026 00:24:30 +0530 Subject: [PATCH 4/4] Feat[OBE-11872] - Bump version, private --- Cargo.lock | 2 +- Cargo.toml | 2 +- lib/observo/private | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5723d3bdb..2ed1c15b2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13042,7 +13042,7 @@ dependencies = [ [[package]] name = "vector" -version = "0.44.12" +version = "0.44.13" dependencies = [ "apache-avro", "approx", diff --git a/Cargo.toml b/Cargo.toml index a05be1d16..84e998cb7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "vector" -version = "0.44.12" +version = "0.44.13" authors = ["Vector Contributors "] edition = "2021" description = "A lightweight and ultra-fast tool for building observability pipelines" diff --git a/lib/observo/private b/lib/observo/private index bc3a899cd..bd4c652d5 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit bc3a899cdab3bafd2a486acae092ca188d8affd5 +Subproject commit bd4c652d59771376c8c7180d82b486c8a2f033c3