diff --git a/lib/codecs/src/decoding/format/gelf.rs b/lib/codecs/src/decoding/format/gelf.rs index cb292f722b2e3..95dd632139da0 100644 --- a/lib/codecs/src/decoding/format/gelf.rs +++ b/lib/codecs/src/decoding/format/gelf.rs @@ -316,30 +316,28 @@ mod tests { assert_eq!( log.get(event_path!(VERSION)), - Some(&Value::Bytes(Bytes::from_static(b"1.1"))) + Some(&Value::from_static_str("1.1")) ); assert_eq!( log.get(event_path!(HOST)), - Some(&Value::Bytes(Bytes::from_static(b"example.org"))) + Some(&Value::from_static_str("example.org")) ); assert_eq!( log.get(log_schema().message_key_target_path().unwrap()), - Some(&Value::Bytes(Bytes::from_static( - b"A short message that helps you identify what is going on" - ))) + Some(&Value::from_static_str( + "A short message that helps you identify what is going on" + )) ); assert_eq!( log.get(event_path!(FULL_MESSAGE)), - Some(&Value::Bytes(Bytes::from_static( - b"Backtrace here\n\nmore stuff" - ))) + Some(&Value::from_static_str("Backtrace here\n\nmore stuff")) ); let dt = DateTime::from_timestamp(1_385_053_862, 307_200_000).expect("invalid timestamp"); assert_eq!(log.get(event_path!(TIMESTAMP)), Some(&Value::Timestamp(dt))); assert_eq!(log.get(event_path!(LEVEL)), Some(&Value::Integer(1))); assert_eq!( log.get(event_path!(FACILITY)), - Some(&Value::Bytes(Bytes::from_static(b"foo"))) + Some(&Value::from_static_str("foo")) ); assert_eq!( log.get(event_path!(LINE)), @@ -347,7 +345,7 @@ mod tests { ); assert_eq!( log.get(event_path!(FILE)), - Some(&Value::Bytes(Bytes::from_static(b"/tmp/bar"))) + Some(&Value::from_static_str("/tmp/bar")) ); assert_eq!( log.get(event_path!(add_on_int_in)), @@ -357,7 +355,7 @@ mod tests { ); assert_eq!( log.get(event_path!(add_on_str_in)), - Some(&Value::Bytes(Bytes::from_static(b"A Space Odyssey"))) + Some(&Value::from_static_str("A Space Odyssey")) ); } @@ -486,7 +484,7 @@ mod tests { assert_eq!( log.get(event_path!(VERSION)), - Some(&Value::Bytes(Bytes::from_static(b"1.0"))) + Some(&Value::from_static_str("1.0")) ); assert_eq!( diff --git a/lib/codecs/src/decoding/format/syslog.rs b/lib/codecs/src/decoding/format/syslog.rs index 982bcabdd42b2..4559353017338 100644 --- a/lib/codecs/src/decoding/format/syslog.rs +++ b/lib/codecs/src/decoding/format/syslog.rs @@ -297,7 +297,7 @@ impl Deserializer for SyslogDeserializer { syslog_loose::parse_message_with_year_exact(line, resolve_year, Variant::Either)?; let log = if let (Some(source), LogNamespace::Vector) = (self.source, log_namespace) { - let mut log = LogEvent::from(Value::Bytes(Bytes::from(parsed.msg.to_string()))); + let mut log = LogEvent::from(Value::from(parsed.msg)); insert_metadata_fields_from_syslog(&mut log, source, parsed, log_namespace); log } else { diff --git a/lib/opentelemetry-proto/src/common.rs b/lib/opentelemetry-proto/src/common.rs index e948aea127ab7..255f63e964a4d 100644 --- a/lib/opentelemetry-proto/src/common.rs +++ b/lib/opentelemetry-proto/src/common.rs @@ -8,7 +8,7 @@ use super::proto::common::v1::{AnyValue, ArrayValue, KeyValue, any_value::Value impl From for Value { fn from(av: PBValue) -> Self { match av { - PBValue::StringValue(v) => Value::Bytes(Bytes::from(v)), + PBValue::StringValue(v) => Value::from(v), PBValue::BoolValue(v) => Value::Boolean(v), PBValue::IntValue(v) => Value::Integer(v), PBValue::DoubleValue(v) => NotNan::new(v).map_or(Value::Null, Value::Float), diff --git a/lib/opentelemetry-proto/src/logs.rs b/lib/opentelemetry-proto/src/logs.rs index d3e2b9b44d6da..08d86d3bdcecf 100644 --- a/lib/opentelemetry-proto/src/logs.rs +++ b/lib/opentelemetry-proto/src/logs.rs @@ -1,4 +1,3 @@ -use bytes::Bytes; use chrono::{DateTime, TimeZone, Utc}; use vector_core::{ config::{LegacyKey, LogNamespace, log_schema}, @@ -147,7 +146,7 @@ impl ResourceLog { &mut log, Some(LegacyKey::Overwrite(path!(TRACE_ID_KEY))), path!(TRACE_ID_KEY), - Bytes::from(to_hex(&self.log_record.trace_id)), + Value::from(to_hex(&self.log_record.trace_id)), ); } if !self.log_record.span_id.is_empty() { @@ -156,7 +155,7 @@ impl ResourceLog { &mut log, Some(LegacyKey::Overwrite(path!(SPAN_ID_KEY))), path!(SPAN_ID_KEY), - Bytes::from(to_hex(&self.log_record.span_id)), + Value::from(to_hex(&self.log_record.span_id)), ); } if !self.log_record.severity_text.is_empty() { @@ -240,7 +239,7 @@ impl ResourceLog { &mut log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(SOURCE_NAME.as_bytes()), + Value::from_static_str(SOURCE_NAME), ); if log_namespace == LogNamespace::Vector { log.metadata_mut() diff --git a/lib/vector-core/src/config/mod.rs b/lib/vector-core/src/config/mod.rs index 541f7a4348011..cabc43c4efe57 100644 --- a/lib/vector-core/src/config/mod.rs +++ b/lib/vector-core/src/config/mod.rs @@ -1,7 +1,6 @@ use std::{collections::HashMap, fmt, num::NonZeroUsize, sync::Arc}; use bitmask_enum::bitmask; -use bytes::Bytes; use chrono::{DateTime, Utc}; mod global_options; @@ -501,7 +500,7 @@ impl LogNamespace { log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(source_name.as_bytes()), + Value::from_static_str(source_name), ); self.insert_vector_metadata( log, diff --git a/lib/vector-core/src/event/log_event.rs b/lib/vector-core/src/event/log_event.rs index aec715d7739ad..a0fc47c0ec09d 100644 --- a/lib/vector-core/src/event/log_event.rs +++ b/lib/vector-core/src/event/log_event.rs @@ -792,9 +792,9 @@ impl From<&tracing::Event<'_>> for LogEvent { log.insert( &TRACING_TARGET_PATHS.kind, if meta.is_event() { - Value::Bytes("event".to_string().into()) + Value::from_static_str("event") } else if meta.is_span() { - Value::Bytes("span".to_string().into()) + Value::from_static_str("span") } else { Value::Null }, @@ -803,7 +803,7 @@ impl From<&tracing::Event<'_>> for LogEvent { log.insert( &TRACING_TARGET_PATHS.module_path, meta.module_path() - .map_or(Value::Null, |mp| Value::Bytes(mp.to_string().into())), + .map_or(Value::Null, |mp| Value::from(mp.to_string())), ); log.insert(&TRACING_TARGET_PATHS.target, meta.target().to_string()); log diff --git a/lib/vector-vrl/metrics/src/common.rs b/lib/vector-vrl/metrics/src/common.rs index 73938ac426155..76f1254d170ff 100644 --- a/lib/vector-vrl/metrics/src/common.rs +++ b/lib/vector-vrl/metrics/src/common.rs @@ -150,7 +150,7 @@ pub(crate) fn metric_into_vrl(value: &Metric) -> Value { Value::Array( v.iter() .filter_map(|v| { - v.map(ToString::to_string).map(Into::into).map(Value::Bytes) + v.map(ToString::to_string).map(Value::from) }) .collect(), ), diff --git a/src/conditions/datadog_search.rs b/src/conditions/datadog_search.rs index 77109b92fb0b5..c4461aed0eb84 100644 --- a/src/conditions/datadog_search.rs +++ b/src/conditions/datadog_search.rs @@ -1,6 +1,5 @@ use std::{borrow::Cow, str::FromStr}; -use bytes::Bytes; use vector_lib::{ configurable::configurable_component, event::{Event, EventRef, LogEvent, Value}, @@ -140,17 +139,13 @@ impl Filter for EventFilter { Field::Reserved(field) if field == "tags" => { let to_match = to_match.to_owned(); - array_match_multiple(vec!["ddtags", "tags"], move |values| { - values.contains(&Value::Bytes(Bytes::copy_from_slice(to_match.as_bytes()))) - }) + any_string_match_multiple(vec!["ddtags", "tags"], move |value| value == to_match) } // Individual tags are compared by element key:value. Field::Tag(tag) => { - let value_bytes = Value::Bytes(format!("{tag}:{to_match}").into()); + let to_match = format!("{tag}:{to_match}"); - array_match_multiple(vec!["ddtags", "tags"], move |values| { - values.contains(&value_bytes) - }) + any_string_match_multiple(vec!["ddtags", "tags"], move |value| value == to_match) } // A literal "source" field should string match in "source" and "ddsource" fields (OR condition). Field::Reserved(field) if field == "source" => { @@ -1648,6 +1643,28 @@ mod test { test_filter(EventFilter, vector_lib::event::Event::into_log); } + #[test] + fn tag_equality_matches_byte_values() { + for query in ["tags:foo", "env:prod"] { + let config: DatadogSearchConfig = query.parse().unwrap(); + let runner = DatadogSearchRunner::try_from(&config).unwrap(); + let mut log = LogEvent::default(); + log.insert( + vrl::event_path!("tags"), + Value::Array(vec![Value::Bytes( + if query == "tags:foo" { + "foo" + } else { + "env:prod" + } + .into(), + )]), + ); + + assert!(runner.matches(&Event::Log(log)), "query: {query}"); + } + } + #[test] fn generate_config() { crate::test_util::test_generate_config::(); diff --git a/src/enrichment_tables/memory/bloom_table.rs b/src/enrichment_tables/memory/bloom_table.rs index 5fcd9177d7c41..8391e48bfc4d0 100644 --- a/src/enrichment_tables/memory/bloom_table.rs +++ b/src/enrichment_tables/memory/bloom_table.rs @@ -7,7 +7,6 @@ use std::{ use async_trait::async_trait; use bloomy::{BloomFilter, bloom}; -use bytes::Bytes; use futures::{ Stream, StreamExt, stream::{self, BoxStream}, @@ -164,10 +163,7 @@ impl Table for BloomMemoryTable { include_key_metric_tag: self.config.internal_metrics.include_key_tag }); let result = ObjectMap::from([ - ( - KeyString::from("key"), - Value::Bytes(Bytes::copy_from_slice(key.as_bytes())), - ), + (KeyString::from("key"), Value::from(key)), (KeyString::from("value"), Value::Null), ]); Ok(vec![result]) diff --git a/src/enrichment_tables/memory/cuckoo_table.rs b/src/enrichment_tables/memory/cuckoo_table.rs index 6b374004a7e12..bef4e1029ca13 100644 --- a/src/enrichment_tables/memory/cuckoo_table.rs +++ b/src/enrichment_tables/memory/cuckoo_table.rs @@ -14,7 +14,6 @@ use std::{ }; use async_trait::async_trait; -use bytes::Bytes; use cuckoo_clock::{ CuckooFilter, ExportableRandomState, InsertValues, LookupValues, config::{CounterConfig, CuckooConfiguration, LruAgingStrategy, LruConfig, TtlConfig}, @@ -646,16 +645,10 @@ impl Table for CuckooMemoryTable { include_key_metric_tag: self.config.internal_metrics.include_key_tag }); let mut result = ObjectMap::from([ - ( - KeyString::from("key"), - Value::Bytes(Bytes::copy_from_slice(key.as_bytes())), - ), + (KeyString::from("key"), Value::from(key)), ( KeyString::from("fingerprint"), - Value::Bytes(Bytes::from(format!( - "{:X}", - associated_data.get_fingerprint() - ))), + Value::from(format!("{:X}", associated_data.get_fingerprint())), ), (KeyString::from("value"), Value::Null), ]); diff --git a/src/enrichment_tables/memory/table.rs b/src/enrichment_tables/memory/table.rs index be7e6e16f3eea..6a90c00c77678 100644 --- a/src/enrichment_tables/memory/table.rs +++ b/src/enrichment_tables/memory/table.rs @@ -8,7 +8,6 @@ use std::{ }; use async_trait::async_trait; -use bytes::Bytes; use evmap::{ shallow_copy::CopyValue, {self}, @@ -70,10 +69,7 @@ impl MemoryEntry { .ttl .saturating_sub(now.duration_since(*self.update_time).as_secs()); Ok(ObjectMap::from([ - ( - KeyString::from("key"), - Value::Bytes(Bytes::copy_from_slice(key.as_bytes())), - ), + (KeyString::from("key"), Value::from(key)), ( KeyString::from("value"), // Unreachable in normal operation: `value` was serialized by `handle_value`. diff --git a/src/sinks/websocket_server/buffering.rs b/src/sinks/websocket_server/buffering.rs index 3b94e88bf0953..39074da453db3 100644 --- a/src/sinks/websocket_server/buffering.rs +++ b/src/sinks/websocket_server/buffering.rs @@ -1,6 +1,5 @@ use std::{collections::VecDeque, net::SocketAddr, num::NonZeroUsize}; -use bytes::Bytes; use derivative::Derivative; use tokio_tungstenite::tungstenite::{Message, handshake::server::Request}; use url::Url; @@ -11,7 +10,7 @@ use vector_lib::{ event::{Event, MaybeAsLogMut}, lookup::lookup_v2::ConfigValuePath, }; -use vrl::prelude::VrlValueConvert; +use vrl::{prelude::VrlValueConvert, value::Value}; use crate::serde::default_decoding; @@ -239,7 +238,7 @@ impl WsMessageBufferConfig for Option { let mut buffer = [0; 36]; let uuid = message_id.hyphenated().encode_lower(&mut buffer); log.value_mut() - .insert(message_id_path, Bytes::copy_from_slice(uuid.as_bytes())); + .insert(message_id_path, Value::from(uuid as &str)); } message_id } diff --git a/src/sources/amqp.rs b/src/sources/amqp.rs index 0f374fa59d0f1..179f2dbc5f5c2 100644 --- a/src/sources/amqp.rs +++ b/src/sources/amqp.rs @@ -22,7 +22,7 @@ use vector_lib::{ internal_event::{CountByteSize, EventsReceived, InternalEventHandle as _}, lookup::{lookup_v2::OptionalValuePath, metadata_path, owned_value_path, path}, }; -use vrl::value::Kind; +use vrl::value::{Kind, Value}; use crate::{ SourceSender, @@ -286,7 +286,7 @@ fn populate_log_event( log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(AmqpSourceConfig::NAME.as_bytes()), + Value::from_static_str(AmqpSourceConfig::NAME), ); // This handles the transition from the original timestamp logic. Originally the diff --git a/src/sources/aws_kinesis_firehose/handlers.rs b/src/sources/aws_kinesis_firehose/handlers.rs index ce8b3eac00343..a53fcbc96d658 100644 --- a/src/sources/aws_kinesis_firehose/handlers.rs +++ b/src/sources/aws_kinesis_firehose/handlers.rs @@ -105,7 +105,7 @@ pub(super) async fn firehose( log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(AwsKinesisFirehoseConfig::NAME.as_bytes()), + Value::from_static_str(AwsKinesisFirehoseConfig::NAME), ); // This handles the transition from the original timestamp logic. Originally the // `timestamp_key` was always populated by the `request.timestamp` time. diff --git a/src/sources/aws_kinesis_firehose/mod.rs b/src/sources/aws_kinesis_firehose/mod.rs index 8744bc3ffa5c9..61e876743730f 100644 --- a/src/sources/aws_kinesis_firehose/mod.rs +++ b/src/sources/aws_kinesis_firehose/mod.rs @@ -607,7 +607,7 @@ mod tests { assert_event_data_eq!( events, vec![log_event! { - "source_type" => Bytes::from("aws_kinesis_firehose"), + "source_type" => "aws_kinesis_firehose", "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms "message" => Bytes::from(expected), "request_id" => REQUEST_ID, @@ -792,7 +792,7 @@ mod tests { assert_event_data_eq!( events, vec![log_event! { - "source_type" => Bytes::from("aws_kinesis_firehose"), + "source_type" => "aws_kinesis_firehose", "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms "message"=> RECORD, "request_id" => REQUEST_ID, @@ -839,7 +839,7 @@ mod tests { assert_event_data_eq!( events, vec![log_event! { - "source_type" => Bytes::from("aws_kinesis_firehose"), + "source_type" => "aws_kinesis_firehose", "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms "message"=> RECORD, "request_id" => REQUEST_ID, @@ -975,7 +975,7 @@ mod tests { assert_event_data_eq!( events, vec![log_event! { - "source_type" => Bytes::from("aws_kinesis_firehose"), + "source_type" => "aws_kinesis_firehose", "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms "message"=> RECORD, "request_id" => REQUEST_ID, @@ -1309,7 +1309,7 @@ mod tests { assert_event_data_eq!( events, vec![log_event! { - "source_type" => Bytes::from("aws_kinesis_firehose"), + "source_type" => "aws_kinesis_firehose", "timestamp" => timestamp.trunc_subsecs(3), // AWS sends timestamps as ms "message"=> Bytes::from(expected), "request_id" => REQUEST_ID, diff --git a/src/sources/aws_s3/sqs.rs b/src/sources/aws_s3/sqs.rs index 9bca963f4d5c0..3db30f722d660 100644 --- a/src/sources/aws_s3/sqs.rs +++ b/src/sources/aws_s3/sqs.rs @@ -59,6 +59,7 @@ use crate::{ sources::aws_s3::AwsS3Config, tls::TlsConfig, }; +use vrl::value::Value; static SUPPORTED_S3_EVENT_VERSION: LazyLock = LazyLock::new(|| semver::VersionReq::parse("~2").unwrap()); @@ -920,7 +921,7 @@ fn handle_single_log( log, Some(LegacyKey::Overwrite(path!("bucket"))), path!("bucket"), - Bytes::from(s3_event.s3.bucket.name.as_bytes().to_vec()), + Value::from(s3_event.s3.bucket.name.clone()), ); log_namespace.insert_source_metadata( @@ -928,14 +929,14 @@ fn handle_single_log( log, Some(LegacyKey::Overwrite(path!("object"))), path!("object"), - Bytes::from(s3_event.s3.object.key.as_bytes().to_vec()), + Value::from(s3_event.s3.object.key.clone()), ); log_namespace.insert_source_metadata( AwsS3Config::NAME, log, Some(LegacyKey::Overwrite(path!("region"))), path!("region"), - Bytes::from(s3_event.aws_region.as_bytes().to_vec()), + Value::from(s3_event.aws_region.clone()), ); if let Some(metadata) = metadata { @@ -954,7 +955,7 @@ fn handle_single_log( log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(AwsS3Config::NAME.as_bytes()), + Value::from_static_str(AwsS3Config::NAME), ); // This handles the transition from the original timestamp logic. Originally the diff --git a/src/sources/datadog_agent/logs.rs b/src/sources/datadog_agent/logs.rs index ee6a663ea655e..96880e33ddf6e 100644 --- a/src/sources/datadog_agent/logs.rs +++ b/src/sources/datadog_agent/logs.rs @@ -218,7 +218,7 @@ fn parse_ddtags(ddtags_raw: &Bytes) -> Value { let ddtags: Vec = ddtags_str .split(',') .filter(|kv| !kv.is_empty()) - .map(|kv| Value::Bytes(Bytes::from(kv.trim().to_string()))) + .map(|kv| Value::from(kv.trim())) .collect(); if ddtags.is_empty() && !ddtags_str.is_empty() { diff --git a/src/sources/datadog_agent/traces.rs b/src/sources/datadog_agent/traces.rs index 456af32166be5..43d326a76575f 100644 --- a/src/sources/datadog_agent/traces.rs +++ b/src/sources/datadog_agent/traces.rs @@ -141,7 +141,7 @@ fn handle_dd_trace_payload_v1( } trace_event.insert( &source.log_schema_source_type_key, - Bytes::from("datadog_agent"), + Value::from_static_str("datadog_agent"), ); trace_event.insert(event_path!("payload_version"), "v2".to_string()); trace_event.insert(&source.log_schema_host_key, hostname.clone()); diff --git a/src/sources/docker_logs/mod.rs b/src/sources/docker_logs/mod.rs index 337aa8f9628c9..886c3b65b2586 100644 --- a/src/sources/docker_logs/mod.rs +++ b/src/sources/docker_logs/mod.rs @@ -1182,7 +1182,7 @@ impl ContainerLogInfo { &mut log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(DockerLogsConfig::NAME.as_bytes()), + Value::from_static_str(DockerLogsConfig::NAME), ); // This handles the transition from the original timestamp logic. Originally the diff --git a/src/sources/file.rs b/src/sources/file.rs index 9f059e6aebd98..81a1e8977f2aa 100644 --- a/src/sources/file.rs +++ b/src/sources/file.rs @@ -23,7 +23,7 @@ use vector_lib::{ finalizer::OrderedFinalizer, lookup::{OwnedValuePath, lookup_v2::OptionalValuePath, owned_value_path, path}, }; -use vrl::value::Kind; +use vrl::value::{Kind, Value}; use super::util::{EncodingConfig, MultilineConfig}; use crate::{ @@ -766,7 +766,7 @@ fn create_event( &mut event, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(FileConfig::NAME.as_bytes()), + Value::from_static_str(FileConfig::NAME), ); log_namespace.insert_vector_metadata( &mut event, diff --git a/src/sources/fluent/message.rs b/src/sources/fluent/message.rs index a096fabbd2c32..0a2832848362e 100644 --- a/src/sources/fluent/message.rs +++ b/src/sources/fluent/message.rs @@ -150,7 +150,7 @@ impl From for Value { .map(Value::Integer) // unwrap large numbers to string similar to how // `From for Value` handles it - .unwrap_or_else(|| Value::Bytes(i.to_string().into())), + .unwrap_or_else(|| Value::from(i.to_string())), rmpv::Value::F32(f) => { // serde_json converts NaN to Null, so we model that behavior here since this is non-fallible NotNan::new(f as f64) @@ -161,7 +161,13 @@ impl From for Value { // serde_json converts NaN to Null, so we model that behavior here since this is non-fallible NotNan::new(f).map(Value::Float).unwrap_or(Value::Null) } - rmpv::Value::String(s) => Value::Bytes(s.into_bytes().into()), + rmpv::Value::String(s) => { + if s.is_str() { + Value::from(s.into_str().expect("string is valid UTF-8")) + } else { + Value::Bytes(s.into_bytes().into()) + } + } rmpv::Value::Binary(bytes) => Value::Bytes(bytes.into()), rmpv::Value::Array(values) => Value::Array( values @@ -250,7 +256,7 @@ mod test { fn from_u64(input: u64) -> () { if input > i64::MAX as u64 { assert_eq!(Value::from(FluentValue(rmpv::Value::Integer(rmpv::Integer::from(input)))), - Value::Bytes(input.to_string().into())) + Value::from(input.to_string())) } else { assert_eq!(Value::from(FluentValue(rmpv::Value::Integer(rmpv::Integer::from(input)))), Value::Integer(input as i64)) @@ -282,11 +288,51 @@ mod test { quickcheck! { fn from_string(input: String) -> () { - assert_eq!(Value::from(FluentValue(rmpv::Value::String(rmpv::Utf8String::from(input.clone())))), - Value::Bytes(input.into_bytes().into())) + let value = Value::from(FluentValue(rmpv::Value::String( + rmpv::Utf8String::from(input.clone()), + ))); + assert!(matches!(value, Value::String(_))); + assert_eq!(value, Value::from(input)) } } + #[test] + fn from_string_preserves_contents_and_nested_strings() { + let actual = Value::from(FluentValue(rmpv::Value::Array(vec![ + rmpv::Value::String("hello\n\\world".into()), + rmpv::Value::Map(vec![( + rmpv::Value::String("key".into()), + rmpv::Value::String("value".into()), + )]), + ]))); + + let expected = Value::Array(vec![ + Value::from("hello\n\\world"), + Value::Object(ObjectMap::from([("key".into(), Value::from("value"))])), + ]); + assert_eq!(actual, expected); + assert!(matches!( + actual, + Value::Array(values) + if matches!(values.first(), Some(Value::String(_))) + && matches!( + values.get(1), + Some(Value::Object(object)) + if matches!(object.get("key"), Some(Value::String(_))) + ) + )); + } + + #[test] + fn from_invalid_string_preserves_bytes() { + let mut encoded = &b"\xa1\xff"[..]; + let value = rmpv::decode::read_value(&mut encoded).unwrap(); + let actual = Value::from(FluentValue(value)); + + assert_eq!(actual, Value::from(&[0xff][..])); + assert!(matches!(actual, Value::Bytes(_))); + } + quickcheck! { fn from_binary(input: Vec) -> () { assert_eq!(Value::from(FluentValue(rmpv::Value::Binary(input.clone()))), diff --git a/src/sources/fluent/mod.rs b/src/sources/fluent/mod.rs index a48c68f65dbaa..4bd402f9248ac 100644 --- a/src/sources/fluent/mod.rs +++ b/src/sources/fluent/mod.rs @@ -809,7 +809,7 @@ impl From> for LogEvent { &mut log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(FluentConfig::NAME.as_bytes()), + Value::from_static_str(FluentConfig::NAME), ); match log_namespace { diff --git a/src/sources/gcp_pubsub.rs b/src/sources/gcp_pubsub.rs index 08d07ee39fbf5..2d0a27ae03fe6 100644 --- a/src/sources/gcp_pubsub.rs +++ b/src/sources/gcp_pubsub.rs @@ -667,7 +667,7 @@ impl PubsubSource { message .attributes .into_iter() - .map(|(key, value)| (key.into(), Value::Bytes(value.into()))) + .map(|(key, value)| (key.into(), Value::from(value))) .collect(), ); let log_namespace = self.log_namespace; diff --git a/src/sources/http_client/client.rs b/src/sources/http_client/client.rs index 2406e4df02cc4..9363a58f54c0f 100644 --- a/src/sources/http_client/client.rs +++ b/src/sources/http_client/client.rs @@ -24,6 +24,7 @@ use vector_lib::{ use vrl::{ compiler::{CompileConfig, Function, Program, runtime::Runtime}, prelude::TypeState, + value::Value, }; use crate::{ @@ -579,7 +580,7 @@ impl http_client::HttpClientContext for HttpClientContext { } Event::Trace(trace) => { trace.maybe_insert(log_schema().source_type_key_target_path(), || { - Bytes::from(HttpClientConfig::NAME).into() + Value::from_static_str(HttpClientConfig::NAME) }); } } diff --git a/src/sources/journald.rs b/src/sources/journald.rs index fb0b0bcc6d345..2d5a20e684060 100644 --- a/src/sources/journald.rs +++ b/src/sources/journald.rs @@ -916,7 +916,7 @@ fn create_log_event_from_record( LogNamespace::Vector => { let message_value = record .remove(MESSAGE) - .map(|msg| Value::Bytes(Bytes::from(msg))) + .map(Value::from) .unwrap_or(Value::Null); let mut log = LogEvent::from(message_value).with_batch_notifier_option(batch); diff --git a/src/sources/kubernetes_logs/mod.rs b/src/sources/kubernetes_logs/mod.rs index 964869310bbba..4b1c2432f6afe 100644 --- a/src/sources/kubernetes_logs/mod.rs +++ b/src/sources/kubernetes_logs/mod.rs @@ -40,7 +40,7 @@ use vector_lib::{ internal_event::{ByteSize, BytesReceived, InternalEventHandle as _, Protocol}, lookup::{OwnedTargetPath, lookup_v2::OptionalTargetPath, owned_value_path, path}, }; -use vrl::value::{Kind, kind::Collection}; +use vrl::value::{Kind, Value, kind::Collection}; use crate::{ SourceSender, @@ -1119,7 +1119,7 @@ fn create_event( &mut log, log_schema().source_type_key(), path!("source_type"), - Bytes::from(Config::NAME), + Value::from_static_str(Config::NAME), ); match (log_namespace, ingestion_timestamp_field) { // When using LogNamespace::Vector always set the ingest_timestamp. diff --git a/src/sources/logstash.rs b/src/sources/logstash.rs index f50562c3c220a..a8239f441a425 100644 --- a/src/sources/logstash.rs +++ b/src/sources/logstash.rs @@ -222,7 +222,7 @@ impl TcpSource for LogstashSource { log, log_schema().source_type_key(), path!("source_type"), - Bytes::from_static(LogstashConfig::NAME.as_bytes()), + Value::from_static_str(LogstashConfig::NAME), ); let log_timestamp = log.get(event_path!("@timestamp")).and_then(|timestamp| { diff --git a/src/sources/redis/mod.rs b/src/sources/redis/mod.rs index 9ea029de8ee66..2035361085822 100644 --- a/src/sources/redis/mod.rs +++ b/src/sources/redis/mod.rs @@ -1,4 +1,3 @@ -use bytes::Bytes; use chrono::Utc; use futures::StreamExt; use snafu::{ResultExt, Snafu}; @@ -15,7 +14,7 @@ use vector_lib::{ }, lookup::{OwnedValuePath, lookup_v2::OptionalValuePath, owned_value_path, path}, }; -use vrl::value::Kind; +use vrl::value::{Kind, Value}; use crate::{ config::{GenerateConfig, SourceConfig, SourceContext, SourceOutput, log_schema}, @@ -250,7 +249,7 @@ impl InputHandler { log, log_schema().source_type_key(), path!("source_type"), - Bytes::from(RedisSourceConfig::NAME), + Value::from_static_str(RedisSourceConfig::NAME), ); self.log_namespace.insert_vector_metadata( log, diff --git a/src/sources/splunk_hec/mod.rs b/src/sources/splunk_hec/mod.rs index 2e85ad66c5577..1ab2a2973ea1a 100644 --- a/src/sources/splunk_hec/mod.rs +++ b/src/sources/splunk_hec/mod.rs @@ -1522,7 +1522,7 @@ fn decode_payload( log, log_schema().source_type_key(), lookup::path!("source_type"), - Bytes::from_static(SplunkConfig::NAME.as_bytes()), + Value::from_static_str(SplunkConfig::NAME), ); match log_namespace { LogNamespace::Vector => { diff --git a/src/sources/util/message_decoding.rs b/src/sources/util/message_decoding.rs index 0c5c35448f778..5ef2fc115824a 100644 --- a/src/sources/util/message_decoding.rs +++ b/src/sources/util/message_decoding.rs @@ -1,6 +1,6 @@ use std::iter; -use bytes::{Bytes, BytesMut}; +use bytes::BytesMut; use chrono::{DateTime, Utc}; use tokio_util::codec::Decoder as _; use vector_lib::{ @@ -16,6 +16,7 @@ use crate::{ event::{BatchNotifier, Event}, }; use vector_lib::codecs::Decoder; +use vrl::value::Value; pub fn decode_message<'a>( mut decoder: Decoder, @@ -41,7 +42,7 @@ pub fn decode_message<'a>( log, schema.source_type_key(), path!("source_type"), - Bytes::from(source_type), + Value::from_static_str(source_type), ); match log_namespace { LogNamespace::Vector => { diff --git a/src/sources/windows_event_log/integration_tests.rs b/src/sources/windows_event_log/integration_tests.rs index c47e7ee85b523..7f454ea4dd4dc 100644 --- a/src/sources/windows_event_log/integration_tests.rs +++ b/src/sources/windows_event_log/integration_tests.rs @@ -844,7 +844,7 @@ async fn test_event_data_format_coercion() { let log = events[0].as_log(); if let Some(eid) = log.get(event_path!("event_id")) { assert!( - matches!(eid, vrl::value::Value::Bytes(_)), + matches!(eid, vrl::value::Value::String(_)), "event_data_format set event_id to String but got {eid:?}. \ Check apply_custom_formatting in parser." ); diff --git a/src/sources/windows_event_log/parser.rs b/src/sources/windows_event_log/parser.rs index dbe89ad1d1c6a..19e06a7189169 100644 --- a/src/sources/windows_event_log/parser.rs +++ b/src/sources/windows_event_log/parser.rs @@ -70,12 +70,12 @@ impl EventLogParser { .cloned() .unwrap_or_else(|| self.extract_message_from_event_data(event)); - log_event.try_insert(message_key, Value::Bytes(message.into())); + log_event.try_insert(message_key, Value::from(message)); } // Set source/host if let Some(host_key) = log_schema.host_key_target_path() { - log_event.try_insert(host_key, Value::Bytes(event.computer.clone().into())); + log_event.try_insert(host_key, Value::from(event.computer.clone())); } // Set Windows-specific fields @@ -104,11 +104,11 @@ impl EventLogParser { .cloned() .unwrap_or_else(|| self.extract_message_from_event_data(event)); - log_event.try_insert(message_key, Value::Bytes(message.into())); + log_event.try_insert(message_key, Value::from(message)); } if let Some(host_key) = log_schema.host_key_target_path() { - log_event.try_insert(host_key, Value::Bytes(event.computer.clone().into())); + log_event.try_insert(host_key, Value::from(event.computer.clone())); } // Set Windows-specific fields at root level @@ -133,47 +133,35 @@ impl EventLogParser { Value::Integer(event.record_id as i64), ); - log_event.insert( - event_path!("level"), - Value::Bytes(event.level_name().into()), - ); + log_event.insert(event_path!("level"), Value::from(event.level_name())); log_event.insert( event_path!("level_value"), Value::Integer(event.level as i64), ); - log_event.insert( - event_path!("channel"), - Value::Bytes(event.channel.clone().into()), - ); + log_event.insert(event_path!("channel"), Value::from(event.channel.clone())); log_event.insert( event_path!("provider_name"), - Value::Bytes(event.provider_name.clone().into()), + Value::from(event.provider_name.clone()), ); if let Some(ref provider_guid) = event.provider_guid { log_event.insert( event_path!("provider_guid"), - Value::Bytes(provider_guid.clone().into()), + Value::from(provider_guid.clone()), ); } - log_event.insert( - event_path!("computer"), - Value::Bytes(event.computer.clone().into()), - ); + log_event.insert(event_path!("computer"), Value::from(event.computer.clone())); if let Some(ref user_id) = event.user_id { - log_event.insert(event_path!("user_id"), Value::Bytes(user_id.clone().into())); + log_event.insert(event_path!("user_id"), Value::from(user_id.clone())); } if let Some(ref user_name) = event.user_name { - log_event.insert( - event_path!("user_name"), - Value::Bytes(user_name.clone().into()), - ); + log_event.insert(event_path!("user_name"), Value::from(user_name.clone())); } log_event.insert( @@ -190,10 +178,7 @@ impl EventLogParser { log_event.insert(event_path!("task"), Value::Integer(event.task as i64)); if let Some(ref task_name) = event.task_name { - log_event.insert( - event_path!("task_name"), - Value::Bytes(task_name.clone().into()), - ); + log_event.insert(event_path!("task_name"), Value::from(task_name.clone())); } } @@ -201,40 +186,34 @@ impl EventLogParser { log_event.insert(event_path!("opcode"), Value::Integer(event.opcode as i64)); if let Some(ref opcode_name) = event.opcode_name { - log_event.insert( - event_path!("opcode_name"), - Value::Bytes(opcode_name.clone().into()), - ); + log_event.insert(event_path!("opcode_name"), Value::from(opcode_name.clone())); } } if event.keywords != 0 { log_event.insert( event_path!("keywords"), - Value::Bytes(format!("0x{:016X}", event.keywords).into()), + Value::from(format!("0x{:016X}", event.keywords)), ); if !event.keyword_names.is_empty() { let kw_values: Vec = event .keyword_names .iter() - .map(|s| Value::Bytes(s.clone().into())) + .map(|s| Value::from(s.clone())) .collect(); log_event.insert(event_path!("keyword_names"), Value::Array(kw_values)); } } if let Some(ref activity_id) = event.activity_id { - log_event.insert( - event_path!("activity_id"), - Value::Bytes(activity_id.clone().into()), - ); + log_event.insert(event_path!("activity_id"), Value::from(activity_id.clone())); } if let Some(ref related_activity_id) = event.related_activity_id { log_event.insert( event_path!("related_activity_id"), - Value::Bytes(related_activity_id.clone().into()), + Value::from(related_activity_id.clone()), ); } @@ -252,17 +231,14 @@ impl EventLogParser { let string_inserts: Vec = event .string_inserts .iter() - .map(|s| Value::Bytes(s.clone().into())) + .map(|s| Value::from(s.clone())) .collect(); log_event.insert(event_path!("string_inserts"), Value::Array(string_inserts)); } // Include raw XML if requested if self.config.include_xml && !event.raw_xml.is_empty() { - log_event.insert( - event_path!("xml"), - Value::Bytes(event.raw_xml.clone().into()), - ); + log_event.insert(event_path!("xml"), Value::from(event.raw_xml.clone())); } // Include event data if configured @@ -297,29 +273,29 @@ impl EventLogParser { /// Use `event_data_format` config entries to opt in to typed coercion /// for specific fields. fn coerce_field_value(&self, key: &str, value: &str) -> Value { - let as_bytes = || Value::Bytes(value.to_string().into()); + let as_string = || Value::from(value); if let Some(fmt) = self.config.event_data_format.get(key) { return match fmt { EventDataFormat::Integer => value .parse::() .map(Value::Integer) - .unwrap_or_else(|_| as_bytes()), + .unwrap_or_else(|_| as_string()), EventDataFormat::Float => value .parse::() .ok() .and_then(|f| ordered_float::NotNan::new(f).ok()) .map(Value::Float) - .unwrap_or_else(as_bytes), + .unwrap_or_else(as_string), EventDataFormat::Boolean => { let lower = value.to_lowercase(); Value::Boolean(matches!(lower.as_str(), "true" | "1" | "yes" | "on")) } - EventDataFormat::String | EventDataFormat::Auto => as_bytes(), + EventDataFormat::String | EventDataFormat::Auto => as_string(), }; } - as_bytes() + as_string() } fn extract_message_from_event_data(&self, event: &WindowsEvent) -> String { @@ -412,7 +388,7 @@ impl EventLogParser { format: &EventDataFormat, ) -> Result { match format { - EventDataFormat::String => Ok(Value::Bytes(value.to_string().into())), + EventDataFormat::String => Ok(Value::from(value.to_string())), EventDataFormat::Integer => { let int_value = match value { Value::Integer(i) => *i, @@ -555,31 +531,31 @@ mod tests { ); assert_eq!( log_event.get(event_path!("level")).unwrap(), - &Value::Bytes("Information".into()) + &Value::from("Information") ); assert_eq!( log_event.get(event_path!("channel")).unwrap(), - &Value::Bytes("TestChannel".into()) + &Value::from("TestChannel") ); assert_eq!( log_event.get(event_path!("provider_name")).unwrap(), - &Value::Bytes("TestProvider".into()) + &Value::from("TestProvider") ); assert_eq!( log_event.get(event_path!("computer")).unwrap(), - &Value::Bytes("TEST-PC".into()) + &Value::from("TEST-PC") ); // Enriched fields from the new resolution methods // opcode=2 -> "Stop" assert_eq!( log_event.get(event_path!("opcode_name")).unwrap(), - &Value::Bytes("Stop".into()) + &Value::from("Stop") ); // keywords=0x8000000000000000 -> ["Classic"] assert_eq!( log_event.get(event_path!("keyword_names")).unwrap(), - &Value::Array(vec![Value::Bytes("Classic".into())]) + &Value::Array(vec![Value::from("Classic")]) ); // task=1 with provider "TestProvider" has no known mapping assert!(log_event.get(event_path!("task_name")).is_none()); @@ -625,7 +601,7 @@ mod tests { // Level 0 should map to "Information" (not "Unknown") assert_eq!( log_event.get(event_path!("level")).unwrap(), - &Value::Bytes("Information".into()) + &Value::from("Information") ); assert_eq!( log_event.get(event_path!("level_value")).unwrap(), @@ -635,13 +611,13 @@ mod tests { // Task 12544 -> "Logon" assert_eq!( log_event.get(event_path!("task_name")).unwrap(), - &Value::Bytes("Logon".into()) + &Value::from("Logon") ); // keywords=Audit Success assert_eq!( log_event.get(event_path!("keyword_names")).unwrap(), - &Value::Array(vec![Value::Bytes("Audit Success".into())]) + &Value::Array(vec![Value::from("Audit Success")]) ); // opcode=0 is not emitted since the condition is `if event.opcode != 0` @@ -662,7 +638,7 @@ mod tests { assert!(log_event.get(event_path!("xml")).is_some()); assert_eq!( log_event.get(event_path!("xml")).unwrap(), - &Value::Bytes(event.raw_xml.into()) + &Value::from(event.raw_xml) ); } @@ -677,8 +653,8 @@ mod tests { let log_event = parser.parse_event(event.clone()).unwrap(); if let Some(Value::Object(event_data)) = log_event.get(event_path!("event_data")) { - assert_eq!(event_data.get("key1"), Some(&Value::Bytes("value1".into()))); - assert_eq!(event_data.get("key2"), Some(&Value::Bytes("value2".into()))); + assert_eq!(event_data.get("key1"), Some(&Value::from("value1"))); + assert_eq!(event_data.get("key2"), Some(&Value::from("value2"))); } else { panic!("event_data should be present"); } @@ -699,7 +675,7 @@ mod tests { // event_id should be converted to string assert_eq!( log_event.get(event_path!("event_id")).unwrap(), - &Value::Bytes("1000".into()) + &Value::from("1000") ); } @@ -764,17 +740,17 @@ mod tests { let result = parser .format_value(&value, &EventDataFormat::String) .unwrap(); - assert_eq!(result, Value::Bytes("123".into())); + assert_eq!(result, Value::from("123")); // Test integer conversion - let value = Value::Bytes("456".into()); + let value = Value::from("456"); let result = parser .format_value(&value, &EventDataFormat::Integer) .unwrap(); assert_eq!(result, Value::Integer(456)); // Test float conversion - let value = Value::Bytes("123.45".into()); + let value = Value::from("123.45"); let result = parser .format_value(&value, &EventDataFormat::Float) .unwrap(); @@ -785,7 +761,7 @@ mod tests { } // Test boolean conversion - let value = Value::Bytes("true".into()); + let value = Value::from("true"); let result = parser .format_value(&value, &EventDataFormat::Boolean) .unwrap(); diff --git a/src/sources/windows_event_log/tests.rs b/src/sources/windows_event_log/tests.rs index 1d17cc4f77982..4c3c8dc19a147 100644 --- a/src/sources/windows_event_log/tests.rs +++ b/src/sources/windows_event_log/tests.rs @@ -310,7 +310,7 @@ mod parser_tests { ); assert_eq!( log_event.get(event_path!("level")), - Some(&Value::Bytes("Information".into())) + Some(&Value::from("Information")) ); assert_eq!( log_event.get(event_path!("level_value")), @@ -318,15 +318,15 @@ mod parser_tests { ); assert_eq!( log_event.get(event_path!("channel")), - Some(&Value::Bytes("Security".into())) + Some(&Value::from("Security")) ); assert_eq!( log_event.get(event_path!("provider_name")), - Some(&Value::Bytes("Microsoft-Windows-Security-Auditing".into())) + Some(&Value::from("Microsoft-Windows-Security-Auditing")) ); assert_eq!( log_event.get(event_path!("computer")), - Some(&Value::Bytes("WIN-SERVER-01".into())) + Some(&Value::from("WIN-SERVER-01")) ); assert_eq!( log_event.get(event_path!("process_id")), @@ -350,8 +350,8 @@ mod parser_tests { // XML should be included assert!(log_event.get(event_path!("xml")).is_some()); - if let Some(Value::Bytes(xml_bytes)) = log_event.get(event_path!("xml")) { - let xml_string = String::from_utf8_lossy(xml_bytes); + if let Some(xml_value) = log_event.get(event_path!("xml")) { + let xml_string = xml_value.to_string_lossy(); assert!(xml_string.contains("4624<")); } @@ -371,9 +371,9 @@ mod parser_tests { if let Some(Value::Object(event_data)) = log_event.get(event_path!("event_data")) { assert_eq!( event_data.get("TargetUserName"), - Some(&Value::Bytes("admin".into())) + Some(&Value::from("admin")) ); - assert_eq!(event_data.get("LogonType"), Some(&Value::Bytes("2".into()))); + assert_eq!(event_data.get("LogonType"), Some(&Value::from("2"))); } else { panic!("event_data should be present and be an object"); } @@ -397,7 +397,7 @@ mod parser_tests { // event_id should be formatted as string assert_eq!( log_event.get(event_path!("event_id")), - Some(&Value::Bytes("4624".into())) + Some(&Value::from("4624")) ); // process_id should be formatted as float @@ -1582,10 +1582,10 @@ mod truncation_tests { let log_event = parser.parse_event(event).unwrap(); - if let Some(Value::Bytes(xml)) = log_event.get(event_path!("xml")) { + if let Some(xml_value) = log_event.get(event_path!("xml")) { // XML should be truncated or limited assert!( - xml.len() <= 40000, + xml_value.as_bytes().map_or(0, |xml| xml.len()) <= 40000, "XML should be handled without memory issues" ); }