diff --git a/Cargo.lock b/Cargo.lock index 687885d2b..8733f2b74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -66,7 +66,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "rand 0.10.1", - "sha1", + "sha1 0.11.0", "smallvec", "tokio", "tokio-util", @@ -777,9 +777,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] name = "aws-config" -version = "1.8.14" +version = "1.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a8fc176d53d6fe85017f230405e3255cedb4a02221cb55ed6d76dccbbb099b2" +checksum = "701418aa459dac33e50a0f8e818e5662a16bc018a6ac7423659b70f3799d67a8" dependencies = [ "aws-credential-types", "aws-runtime", @@ -791,13 +791,14 @@ dependencies = [ "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", "aws-types", "bytes", "fastrand 2.3.0", "hex", "http 1.4.0", - "ring", + "sha1 0.10.7", "time", "tokio", "tracing", @@ -807,9 +808,9 @@ dependencies = [ [[package]] name = "aws-credential-types" -version = "1.2.13" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d203b0bf2626dcba8665f5cd0871d7c2c0930223d6b6be9097592fea21242d0" +checksum = "e93964ffdaf57857f544be3666a5f57570bb699e934700f11b49708f61bb556e" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api", @@ -859,9 +860,9 @@ dependencies = [ [[package]] name = "aws-runtime" -version = "1.7.1" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ede2ddc593e6c8acc6ce3358c28d6677a6dc49b65ba4b37a2befe14a11297e75" +checksum = "a6b50a43f3ccdf331521c6d6c68b7cc9668b6e09d439ebda9569df5722324d76" dependencies = [ "aws-credential-types", "aws-sigv4", @@ -884,10 +885,11 @@ dependencies = [ [[package]] name = "aws-sdk-sso" -version = "1.95.0" +version = "1.104.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00c5ff27c6ba2cbd95e6e26e2e736676fdf6bcf96495b187733f521cfe4ce448" +checksum = "b53416d16c278234845392e38d93bd4481d2f09daa0f005a2277f0aa91f59c22" dependencies = [ + "arc-swap", "aws-credential-types", "aws-runtime", "aws-smithy-async", @@ -896,6 +898,7 @@ dependencies = [ "aws-smithy-observability", "aws-smithy-runtime", "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", "aws-types", "bytes", @@ -908,10 +911,11 @@ dependencies = [ [[package]] name = "aws-sdk-ssooidc" -version = "1.97.0" +version = "1.106.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d186f1e5a3694a188e5a0640b3115ccc6e084d104e16fd6ba968dca072ffef8" +checksum = "cc9b706c3305ed0285d5b1b696c747aa34950f830fb03e3e6c76890f99b9f188" dependencies = [ + "arc-swap", "aws-credential-types", "aws-runtime", "aws-smithy-async", @@ -920,6 +924,7 @@ dependencies = [ "aws-smithy-observability", "aws-smithy-runtime", "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", "aws-types", "bytes", @@ -932,10 +937,11 @@ dependencies = [ [[package]] name = "aws-sdk-sts" -version = "1.99.0" +version = "1.109.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9acba7c62f3d4e2408fa998a3a8caacd8b9a5b5549cf36e2372fbdae329d5449" +checksum = "32d214cdfa5bbe17f117e76a7643fadf32a5234fb597322ef8b1fb4b2f17dbbd" dependencies = [ + "arc-swap", "aws-credential-types", "aws-runtime", "aws-smithy-async", @@ -945,6 +951,7 @@ dependencies = [ "aws-smithy-query", "aws-smithy-runtime", "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", "aws-smithy-xml", "aws-types", @@ -957,9 +964,9 @@ dependencies = [ [[package]] name = "aws-sigv4" -version = "1.4.1" +version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37411f8e0f4bea0c3ca0958ce7f18f6439db24d555dbd809787262cd00926aa9" +checksum = "723c2234ad7511ceef63eab016b7ba6ff7c55590fefb96fa8467af014a07309f" dependencies = [ "aws-credential-types", "aws-smithy-http", @@ -972,16 +979,16 @@ dependencies = [ "http 0.2.12", "http 1.4.0", "percent-encoding", - "sha2", + "sha2 0.11.0", "time", "tracing", ] [[package]] name = "aws-smithy-async" -version = "1.2.13" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5cc50d0f63e714784b84223abd7abbc8577de8c35d699e0edd19f0a88a08ae13" +checksum = "f02e407fb3b54891734224b9ffac8a71fdd35f542500fa1af95754a6b2beb316" dependencies = [ "futures-util", "pin-project-lite", @@ -990,9 +997,9 @@ dependencies = [ [[package]] name = "aws-smithy-http" -version = "0.63.5" +version = "0.64.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d619373d490ad70966994801bc126846afaa0d1ee920697a031f0cf63f2568e7" +checksum = "37843d9add67c3aff5856f409c6dc315d3cdff60f9c0cb5b670dab1e9920306d" dependencies = [ "aws-smithy-runtime-api", "aws-smithy-types", @@ -1011,9 +1018,9 @@ dependencies = [ [[package]] name = "aws-smithy-http-client" -version = "1.1.11" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00ccbb08c10f6bcf912f398188e42ee2eab5f1767ce215a02a73bc5df1bbdd95" +checksum = "635d23afda0a6ab48d666c4d447c4873e8d1e83518a2be2093122397e50b838e" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api", @@ -1035,43 +1042,49 @@ dependencies = [ [[package]] name = "aws-smithy-json" -version = "0.62.4" +version = "0.63.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "27b3a779093e18cad88bbae08dc4261e1d95018c4c5b9356a52bcae7c0b6e9bb" +checksum = "3dc65a121adb4b33729919fcfa14fa36fb33c1555a8f06bb0e2188dbfdc1d9ef" dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", ] [[package]] name = "aws-smithy-observability" -version = "0.2.5" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d3f39d5bb871aaf461d59144557f16d5927a5248a983a40654d9cf3b9ba183b" +checksum = "8e86338c869539a581bf161247762a6e87f92c5c075060057b5ed6d06632ed0c" dependencies = [ "aws-smithy-runtime-api", ] [[package]] name = "aws-smithy-query" -version = "0.60.14" +version = "0.62.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f76a580e3d8f8961e5d48763214025a2af65c2fa4cd1fb7f270a0e107a71b0" +checksum = "512346c7212ab7436df2d77a16d976a468ae44a418835511d2a69269810aaf62" dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", + "aws-smithy-xml", "urlencoding", ] [[package]] name = "aws-smithy-runtime" -version = "1.10.2" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22ccf7f6eba8b2dcf8ce9b74806c6c185659c311665c4bf8d6e71ebd454db6bf" +checksum = "bea94a9ff8464016338c851e24b472d7131c388c88898a502e781815b2ee6045" dependencies = [ "aws-smithy-async", "aws-smithy-http", "aws-smithy-http-client", "aws-smithy-observability", "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", "bytes", "fastrand 2.3.0", @@ -1088,11 +1101,12 @@ dependencies = [ [[package]] name = "aws-smithy-runtime-api" -version = "1.11.5" +version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b4af6e5def28be846479bbeac55aa4603d6f7986fc5da4601ba324dd5d377516" +checksum = "22ed1ebe6e0a95ea84570225f5a8208dec4b8f77e61a9b0d6f51773fcb4612f0" dependencies = [ "aws-smithy-async", + "aws-smithy-runtime-api-macros", "aws-smithy-types", "bytes", "http 0.2.12", @@ -1103,11 +1117,33 @@ dependencies = [ "zeroize", ] +[[package]] +name = "aws-smithy-runtime-api-macros" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "221eaa237ddf1ca79b60d1372aad77e47f9c0ea5b3ce5099da8c61d027dc77b3" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "aws-smithy-schema" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56e0a4e53127a632224e43633b0fe045fa9e1e3cfc68b9830f1115e103f910" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "http 1.4.0", +] + [[package]] name = "aws-smithy-types" -version = "1.4.5" +version = "1.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ca2734c16913a45343b37313605d84e7d8b34a4611598ce1d25b35860a2bed3" +checksum = "d6dc683efb34b9e755675b37fedbe0103141e5b6df7bdc9eb6967756a8c167d8" dependencies = [ "base64-simd", "bytes", @@ -1131,22 +1167,26 @@ dependencies = [ [[package]] name = "aws-smithy-xml" -version = "0.60.14" +version = "0.62.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b53543b4b86ed43f051644f704a98c7291b3618b67adf057ee77a366fa52fcaa" +checksum = "ce84f71c72fee2cbbadde6e7d082f5fb466e3a84733855295fa7aafd1b31b7d8" dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", "xmlparser", ] [[package]] name = "aws-types" -version = "1.3.13" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0470cc047657c6e286346bdf10a8719d26efd6a91626992e0e64481e44323e96" +checksum = "eec1cd5469f328c782dc3e33d4153cf118a54e33cbb3356d60d16f89883e1f94" dependencies = [ "aws-credential-types", "aws-smithy-async", "aws-smithy-runtime-api", + "aws-smithy-schema", "aws-smithy-types", "rustc_version", "tracing", @@ -1245,6 +1285,24 @@ dependencies = [ "num-traits", ] +[[package]] +name = "bindgen" +version = "0.72.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" +dependencies = [ + "bitflags 2.10.0", + "cexpr", + "clang-sys", + "itertools 0.12.1", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex", + "syn", +] + [[package]] name = "biscuit" version = "0.7.0" @@ -1435,6 +1493,15 @@ dependencies = [ "shlex", ] +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -1507,6 +1574,17 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" +[[package]] +name = "clang-sys" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +dependencies = [ + "glob", + "libc", + "libloading", +] + [[package]] name = "clap" version = "4.5.53" @@ -1564,6 +1642,12 @@ dependencies = [ "cc", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "colorchoice" version = "1.0.4" @@ -1841,6 +1925,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "ctutils" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" +dependencies = [ + "cmov", +] + [[package]] name = "curl-sys" version = "0.4.90+curl-8.21.0" @@ -1853,7 +1946,7 @@ dependencies = [ "openssl-sys", "pkg-config", "vcpkg", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2278,7 +2371,7 @@ dependencies = [ "num-traits", "rand 0.9.4", "regex", - "sha2", + "sha2 0.10.9", "unicode-segmentation", "uuid", ] @@ -2693,6 +2786,7 @@ dependencies = [ "block-buffer 0.12.0", "const-oid", "crypto-common 0.2.1", + "ctutils", ] [[package]] @@ -3165,11 +3259,11 @@ checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" [[package]] name = "hmac" -version = "0.12.1" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" dependencies = [ - "digest 0.10.7", + "digest 0.11.2", ] [[package]] @@ -3777,6 +3871,16 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libloading" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +dependencies = [ + "cfg-if", + "windows-link 0.2.1", +] + [[package]] name = "liblzma" version = "0.4.6" @@ -3951,6 +4055,12 @@ dependencies = [ "unicase", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -3979,6 +4089,16 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e94e1e6445d314f972ff7395df2de295fe51b71821694f0b0e1e79c4f12c8577" +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "ntapi" version = "0.4.2" @@ -4494,7 +4614,7 @@ dependencies = [ "serde_json", "serde_repr", "sha1_smol", - "sha2", + "sha2 0.10.9", "static-files 0.2.5", "sysinfo 0.34.2", "temp-dir", @@ -4727,9 +4847,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.104" +version = "1.0.107" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9695f8df41bb4f3d222c95a67532365f569318332d03d5f3f67f37b20e6ebdf0" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" dependencies = [ "unicode-ident", ] @@ -5107,9 +5227,9 @@ dependencies = [ [[package]] name = "rdkafka" -version = "0.37.0" +version = "0.39.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "14b52c81ac3cac39c9639b95c20452076e74b8d9a71bc6fc4d83407af2ea6fff" +checksum = "d7956f9ac12b5712e50372d9749a3102f4810a8d42481c5eae3748d36d585bcf" dependencies = [ "futures-channel", "futures-util", @@ -5126,9 +5246,9 @@ dependencies = [ [[package]] name = "rdkafka-sys" -version = "4.9.0+2.10.0" +version = "4.10.0+2.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5230dca48bc354d718269f3e4353280e188b610f7af7e2fcf54b7a79d5802872" +checksum = "e234cf318915c1059d4921ef7f75616b5219b10b46e9f3a511a15eb4b56a3f77" dependencies = [ "cmake", "curl-sys", @@ -5138,6 +5258,7 @@ dependencies = [ "openssl-sys", "pkg-config", "sasl2-sys", + "zstd-sys", ] [[package]] @@ -5640,6 +5761,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + [[package]] name = "sha1" version = "0.11.0" @@ -5668,6 +5800,17 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.2", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -7587,6 +7730,7 @@ version = "2.0.16+zstd.1.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" dependencies = [ + "bindgen", "cc", "pkg-config", ] diff --git a/Cargo.toml b/Cargo.toml index 1ec1391d9..d9115c69c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -50,16 +50,16 @@ tower-http = { version = "0.6.1", features = ["cors"] } url = "2.4.0" # Connectors dependencies -rdkafka = { version = "0.37", optional = true, features = [ +rdkafka = { version = "0.39.0", optional = true, features = [ "cmake-build", "tracing", "libz-static", ] } -sasl2-sys = { version = "0.1.22", optional = true, features = ["vendored"] } +sasl2-sys = { version = "0.1.22+2.1.28", optional = true, features = ["vendored"] } # AWS MSK IAM authentication for Kafka (SASL/OAUTHBEARER token signing) aws-msk-iam-sasl-signer = { version = "1.0.1", optional = true } -aws-types = { version = "1", optional = true } -aws-config = { version = "1", optional = true } +aws-types = { version = "1.4.0", optional = true } +aws-config = { version = "1.9.0", optional = true } # Authentication and Security argon2 = "0.5.0" @@ -242,6 +242,7 @@ kafka = [ "rdkafka", "rdkafka/ssl-vendored", "rdkafka/ssl", + "rdkafka/zstd", "rdkafka/sasl", "rdkafka/curl", "rdkafka/curl-static", diff --git a/src/connectors/kafka/processor.rs b/src/connectors/kafka/processor.rs index 014e2c5d5..a892691b2 100644 --- a/src/connectors/kafka/processor.rs +++ b/src/connectors/kafka/processor.rs @@ -19,13 +19,15 @@ use crate::{ connectors::common::processor::Processor, event::{ - Event as ParseableEvent, USER_AGENT_KEY, + USER_AGENT_KEY, format::{EventFormat, LogSourceEntry, json}, }, handlers::TelemetryType, + metadata::SchemaVersion, parseable::PARSEABLE, storage::StreamType, }; +use arrow_schema::Field; use async_trait::async_trait; use chrono::Utc; use futures_util::StreamExt; @@ -42,19 +44,14 @@ use super::{ConsumerRecord, StreamConsumer, TopicPartition, config::BufferConfig pub struct ParseableSinkProcessor; impl ParseableSinkProcessor { - async fn build_event_from_chunk( + /// Ensures the destination stream exists and fetches the metadata needed + /// to build an event from a chunk. Stays on the (small, I/O-focused) + /// kafka tokio runtime since it does async catalog/metadata lookups. + async fn prepare_stream( &self, - records: &[ConsumerRecord], - ) -> anyhow::Result { - let stream_name = records - .first() - .map(|r| r.topic.as_str()) - .unwrap_or_default(); - let tenant_id = if let Some(r) = records.first() { - &r.tenant_id - } else { - &None - }; + stream_name: &str, + tenant_id: &Option, + ) -> anyhow::Result { let log_source_entry = LogSourceEntry::default(); PARSEABLE .create_stream_if_not_exists( @@ -70,43 +67,68 @@ impl ParseableSinkProcessor { .await?; let stream = PARSEABLE.get_stream(stream_name, tenant_id)?; - let schema = stream.get_schema_raw(); - let time_partition = stream.get_time_partition(); - let custom_partition = stream.get_custom_partition(); - let static_schema_flag = stream.get_static_schema_flag(); - let schema_version = stream.get_schema_version(); - let infer_timestamp = stream.get_infer_timestamp(); - - let mut json_vec = Vec::with_capacity(records.len()); - let mut total_payload_size = 0u64; - - for record in records.iter().filter_map(|r| r.payload.as_ref()) { - total_payload_size += record.len() as u64; - if let Ok(value) = serde_json::from_slice::(record) { - json_vec.push(value); - } + Ok(StreamMeta { + schema: stream.get_schema_raw(), + time_partition: stream.get_time_partition(), + custom_partition: stream.get_custom_partition(), + static_schema_flag: stream.get_static_schema_flag(), + schema_version: stream.get_schema_version(), + infer_timestamp: stream.get_infer_timestamp(), + }) + } +} + +struct StreamMeta { + schema: HashMap>, + time_partition: Option, + custom_partition: Option, + static_schema_flag: bool, + schema_version: SchemaVersion, + infer_timestamp: bool, +} + +/// JSON parse + Arrow record-batch encode for a chunk of records. CPU-bound, +/// so it must run off the kafka connector's dedicated tokio runtime (sized to +/// `partition_listener_concurrency`, default 2 — shared by every partition's +/// poll/commit/recv), same reasoning as the rayon offload used for otel and +/// prometheus remote-write ingestion. +fn build_and_process_event( + records: Vec, + stream_name: String, + tenant_id: Option, + meta: StreamMeta, +) -> anyhow::Result<()> { + let mut json_vec = Vec::with_capacity(records.len()); + let mut total_payload_size = 0u64; + + for record in records.iter().filter_map(|r| r.payload.as_ref()) { + total_payload_size += record.len() as u64; + if let Ok(value) = serde_json::from_slice::(record) { + json_vec.push(value); } + } - let mut p_custom_fields = HashMap::new(); - p_custom_fields.insert(USER_AGENT_KEY.to_string(), "kafka".to_string()); + let mut p_custom_fields = HashMap::new(); + p_custom_fields.insert(USER_AGENT_KEY.to_string(), "kafka".to_string()); - let p_event = json::Event::new(Value::Array(json_vec), Utc::now()).into_event( - stream_name.to_string(), + json::Event::new(Value::Array(json_vec), Utc::now()) + .into_event( + stream_name, total_payload_size, - &schema, - static_schema_flag, - custom_partition.as_ref(), - time_partition.as_ref(), - schema_version, + &meta.schema, + meta.static_schema_flag, + meta.custom_partition.as_ref(), + meta.time_partition.as_ref(), + meta.schema_version, StreamType::UserDefined, &p_custom_fields, TelemetryType::Logs, - tenant_id, - infer_timestamp, - )?; + &tenant_id, + meta.infer_timestamp, + )? + .process()?; - Ok(p_event) - } + Ok(()) } #[async_trait] @@ -115,7 +137,17 @@ impl Processor, ()> for ParseableSinkProcessor { let len = records.len(); debug!("Processing {len} records"); - self.build_event_from_chunk(&records).await?.process()?; + let stream_name = records.first().map(|r| r.topic.clone()).unwrap_or_default(); + let tenant_id = records.first().and_then(|r| r.tenant_id.clone()); + + let meta = self.prepare_stream(&stream_name, &tenant_id).await?; + + let (tx, rx) = tokio::sync::oneshot::channel(); + rayon::spawn(move || { + let res = build_and_process_event(records, stream_name, tenant_id, meta); + let _ = tx.send(res); + }); + rx.await??; debug!("Processed {len} records"); Ok(()) diff --git a/src/handlers/http/users/filters.rs b/src/handlers/http/users/filters.rs index 92d5f52bb..22976fd9b 100644 --- a/src/handlers/http/users/filters.rs +++ b/src/handlers/http/users/filters.rs @@ -205,9 +205,9 @@ mod tests { validate_stream_name("team1/serviceA").expect_err("path-like names must be rejected"); assert!(matches!( err, - super::FiltersError::InvalidStreamName( - StreamNameValidationError::NameSpecialChar { c: '/' } - ) + super::FiltersError::InvalidStreamName(StreamNameValidationError::NameSpecialChar { + c: '/' + }) )); } }