refactor(value): Preserve known UTF-8 as strings

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 {

lib/opentelemetry-proto/src/common.rs

@@ -8,7 +8,7 @@ use super::proto::common::v1::{AnyValue, ArrayValue, KeyValue, any_value::Value
 impl From<PBValue> 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),

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()

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,

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

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(),
                                 ),

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<LogEvent> 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::<DatadogSearchConfig>();

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])

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),
                     ]);

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`.

src/sinks/datadog/logs/sink.rs

@@ -1,6 +1,5 @@
 use std::{collections::VecDeque, fmt::Debug, io, sync::Arc};
 
-use bytes::Bytes;
 use itertools::Itertools;
 use snafu::Snafu;
 use tracing::Instrument;
@@ -496,12 +495,17 @@ fn set_truncated_message(
     body_len: usize,
     conforms_as_agent: bool,
 ) {
-    let marker = TRUNCATION_MARKER.as_bytes();
-    let mut truncated = Vec::with_capacity(body_len + marker.len());
-    truncated.extend_from_slice(&message.as_bytes()[..body_len]);
-    truncated.extend_from_slice(marker);
+    // `select_message_body_len` advances by whole characters; clamp for callers and the slice.
+    let body_len = message.floor_char_boundary(body_len);
+    let mut truncated = String::with_capacity(body_len + TRUNCATION_MARKER.len());
+    #[expect(
+        clippy::string_slice,
+        reason = "floor_char_boundary guarantees a valid char boundary"
+    )]
+    truncated.push_str(&message[..body_len]);
+    truncated.push_str(TRUNCATION_MARKER);
     *message_value_mut(log, conforms_as_agent).expect("the message was previously found") =
-        Value::Bytes(Bytes::from(truncated));
+        Value::from(truncated);
 }
 
 fn ensure_truncated_tag(log: &mut LogEvent, encoded_size: usize) -> io::Result<usize> {

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<MessageBufferingConfig> {
             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
     }

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

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.

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,

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<semver::VersionReq> =
     LazyLock::new(|| semver::VersionReq::parse("~2").unwrap());
@@ -920,22 +921,22 @@ 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(
         AwsS3Config::NAME,
         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

src/sources/datadog_agent/logs.rs

@@ -218,7 +218,7 @@ fn parse_ddtags(ddtags_raw: &Bytes) -> Value {
     let ddtags: Vec<Value> = 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() {