Skip to content

Commit bffd3b7

Browse files
prontpront
committed
more changes
1 parent ebcefef commit bffd3b7

File tree

24 files changed

+116
-119
lines changed

24 files changed

+116
-119
lines changed

‎lib/vector-lookup/src/lookup_v2/mod.rs

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,13 @@ impl<'a> ValuePath<'a> for &'a ConfigValuePath {
3838
}
3939
}
4040

41+
#[cfg(any(test, feature = "test"))]
42+
impl From<&str> for ConfigValuePath {
43+
fn from(path: &str) -> Self {
44+
ConfigValuePath::try_from(path.to_string()).unwrap()
45+
}
46+
}
47+
4148
/// A wrapper around `OwnedTargetPath` that allows it to be used in Vector config
4249
/// with prefix default to `PathPrefix::Event`
4350
#[configurable_component]

‎src/codecs/encoding/config.rs

Lines changed: 3 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -172,10 +172,7 @@ mod test {
172172
transformer.only_fields(),
173173
&Some(vec![ConfigValuePath(parse_value_path("a.b[0]").unwrap())])
174174
);
175-
assert_eq!(
176-
transformer.except_fields(),
177-
&Some(vec!["ignore_me".to_owned()])
178-
);
175+
assert_eq!(transformer.except_fields(), &Some(vec!["ignore_me".into()]));
179176
assert_eq!(transformer.timestamp_format(), &Some(TimestampFormat::Unix));
180177
}
181178

@@ -207,10 +204,7 @@ mod test {
207204
transformer.only_fields(),
208205
&Some(vec![ConfigValuePath(parse_value_path("a.b[0]").unwrap())])
209206
);
210-
assert_eq!(
211-
transformer.except_fields(),
212-
&Some(vec!["ignore_me".to_owned()])
213-
);
207+
assert_eq!(transformer.except_fields(), &Some(vec!["ignore_me".into()]));
214208
assert_eq!(transformer.timestamp_format(), &Some(TimestampFormat::Unix));
215209
}
216210

@@ -239,10 +233,7 @@ mod test {
239233
transformer.only_fields(),
240234
&Some(vec![ConfigValuePath(parse_value_path("a.b[0]").unwrap())])
241235
);
242-
assert_eq!(
243-
transformer.except_fields(),
244-
&Some(vec!["ignore_me".to_owned()])
245-
);
236+
assert_eq!(transformer.except_fields(), &Some(vec!["ignore_me".into()]));
246237
assert_eq!(transformer.timestamp_format(), &Some(TimestampFormat::Unix));
247238
}
248239
}

‎src/codecs/encoding/transformer.rs

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -51,10 +51,10 @@ impl<'de> Deserialize<'de> for Transformer {
5151
Self::new(
5252
inner
5353
.only_fields
54-
.map(|v| v.iter().map(|p| ConfigValuePath { 0: p.clone() }).collect()),
54+
.map(|v| v.iter().map(|p| ConfigValuePath(p.clone())).collect()),
5555
inner
5656
.except_fields
57-
.map(|v| v.iter().map(|p| ConfigValuePath { 0: p.clone() }).collect()),
57+
.map(|v| v.iter().map(|p| ConfigValuePath(p.clone())).collect()),
5858
inner.timestamp_format,
5959
)
6060
.map_err(serde::de::Error::custom)
@@ -156,18 +156,18 @@ impl Transformer {
156156

157157
fn apply_except_fields(&self, log: &mut LogEvent) {
158158
if let Some(except_fields) = self.except_fields.as_ref() {
159-
let service_path = log
160-
.metadata()
161-
.schema_definition()
162-
.meaning_path(meaning::SERVICE);
163-
164159
for field in except_fields {
165-
let value = log.remove((PathPrefix::Event, field));
160+
let value_path = &field.0;
161+
let value = log.remove((PathPrefix::Event, value_path));
166162

163+
let service_path = log
164+
.metadata()
165+
.schema_definition()
166+
.meaning_path(meaning::SERVICE);
167167
// If we are removing the service field we need to store this in a `dropped_fields` list as we may need to
168168
// refer to this later when emitting metrics.
169-
if let Some(v) = value {
170-
if matches!(service_path, Some(target_path) if target_path.path == field.0) {
169+
if let (Some(v), Some(service_path)) = (value, service_path) {
170+
if service_path.path == *value_path {
171171
log.metadata_mut()
172172
.add_dropped_field(meaning::SERVICE.to_string(), v);
173173
}
@@ -274,7 +274,7 @@ mod tests {
274274
#[test]
275275
fn deserialize_and_transform_except() {
276276
let transformer: Transformer =
277-
toml::from_str(r#"except_fields = ["a.b.c", "b", "c[0].y", "d\\.z", "e"]"#).unwrap();
277+
toml::from_str(r#"except_fields = ["a.b.c", "b", "c[0].y", "d.z", "e"]"#).unwrap();
278278
let mut log = LogEvent::default();
279279
{
280280
log.insert("a", 1);
@@ -285,7 +285,7 @@ mod tests {
285285
log.insert("b[1].x", 1);
286286
log.insert("c[0].x", 1);
287287
log.insert("c[0].y", 1);
288-
log.insert("d\\.z", 1);
288+
log.insert("d.z", 1);
289289
log.insert("e.a", 1);
290290
log.insert("e.b", 1);
291291
}
@@ -295,7 +295,7 @@ mod tests {
295295
assert!(!event.as_mut_log().contains("b"));
296296
assert!(!event.as_mut_log().contains("b[1].x"));
297297
assert!(!event.as_mut_log().contains("c[0].y"));
298-
assert!(!event.as_mut_log().contains("d\\.z"));
298+
assert!(!event.as_mut_log().contains("d.z"));
299299
assert!(!event.as_mut_log().contains("e.a"));
300300

301301
assert!(event.as_mut_log().contains("a.b.d"));

‎src/internal_events/kafka.rs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use metrics::{counter, gauge};
22
use vector_core::{internal_event::InternalEvent, update_counter};
3+
use vrl::path::OwnedTargetPath;
34

45
use vector_common::{
56
internal_event::{error_stage, error_type},
@@ -161,7 +162,7 @@ impl InternalEvent for KafkaStatisticsReceived<'_> {
161162
}
162163

163164
pub struct KafkaHeaderExtractionError<'a> {
164-
pub header_field: &'a str,
165+
pub header_field: &'a OwnedTargetPath,
165166
}
166167

167168
impl InternalEvent for KafkaHeaderExtractionError<'_> {
@@ -171,7 +172,7 @@ impl InternalEvent for KafkaHeaderExtractionError<'_> {
171172
error_code = "extracting_header",
172173
error_type = error_type::PARSER_FAILED,
173174
stage = error_stage::RECEIVING,
174-
header_field = self.header_field,
175+
header_field = self.header_field.to_string(),
175176
internal_log_rate_limit = true,
176177
);
177178
counter!(

‎src/sinks/elasticsearch/tests.rs

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@ use crate::{
1212
},
1313
template::Template,
1414
};
15-
use lookup::owned_value_path;
1615

1716
// helper to unwrap template strings for tests only
1817
fn parse_template(input: &str) -> Template {
@@ -53,7 +52,7 @@ async fn sets_create_action_when_configured() {
5352
.request_builder
5453
.encoder
5554
.encode_input(
56-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
55+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
5756
&mut encoded,
5857
)
5958
.unwrap();
@@ -129,7 +128,7 @@ async fn encode_datastream_mode() {
129128
.request_builder
130129
.encoder
131130
.encode_input(
132-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
131+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
133132
&mut encoded,
134133
)
135134
.unwrap();
@@ -187,7 +186,7 @@ async fn encode_datastream_mode_no_routing() {
187186
.request_builder
188187
.encoder
189188
.encode_input(
190-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
189+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
191190
&mut encoded,
192191
)
193192
.unwrap();
@@ -223,7 +222,7 @@ async fn handle_metrics() {
223222
es.request_builder
224223
.encoder
225224
.encode_input(
226-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
225+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
227226
&mut encoded,
228227
)
229228
.unwrap();
@@ -339,7 +338,7 @@ async fn encode_datastream_mode_no_sync() {
339338
.request_builder
340339
.encoder
341340
.encode_input(
342-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
341+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
343342
&mut encoded,
344343
)
345344
.unwrap();
@@ -358,12 +357,8 @@ async fn allows_using_except_fields() {
358357
index: parse_template("{{ idx }}"),
359358
..Default::default()
360359
},
361-
encoding: Transformer::new(
362-
None,
363-
Some(vec!["idx".to_string(), "timestamp".to_string()]),
364-
None,
365-
)
366-
.unwrap(),
360+
encoding: Transformer::new(None, Some(vec!["idx".into(), "timestamp".into()]), None)
361+
.unwrap(),
367362
endpoints: vec![String::from("https://example.com")],
368363
api_version: ElasticsearchApiVersion::V6,
369364
..Default::default()
@@ -379,7 +374,7 @@ async fn allows_using_except_fields() {
379374
.request_builder
380375
.encoder
381376
.encode_input(
382-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
377+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
383378
&mut encoded,
384379
)
385380
.unwrap();
@@ -398,7 +393,7 @@ async fn allows_using_only_fields() {
398393
index: parse_template("{{ idx }}"),
399394
..Default::default()
400395
},
401-
encoding: Transformer::new(Some(vec![owned_value_path!("foo")]), None, None).unwrap(),
396+
encoding: Transformer::new(Some(vec!["foo".into()]), None, None).unwrap(),
402397
endpoints: vec![String::from("https://example.com")],
403398
api_version: ElasticsearchApiVersion::V6,
404399
..Default::default()
@@ -414,7 +409,7 @@ async fn allows_using_only_fields() {
414409
.request_builder
415410
.encoder
416411
.encode_input(
417-
vec![process_log(log, &es.mode, &None, &config.encoding).unwrap()],
412+
vec![process_log(log, &es.mode, None, &config.encoding).unwrap()],
418413
&mut encoded,
419414
)
420415
.unwrap();
@@ -540,7 +535,7 @@ async fn datastream_index_name() {
540535
),
541536
);
542537

543-
let processed_event = process_log(log, &es.mode, &None, &config.encoding).unwrap();
538+
let processed_event = process_log(log, &es.mode, None, &config.encoding).unwrap();
544539
assert_eq!(processed_event.index, test_case.want, "{test_case:?}");
545540
}
546541
}

‎src/sinks/humio/metrics.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ use codecs::JsonSerializerConfig;
33
use futures::StreamExt;
44
use futures_util::stream::BoxStream;
55
use indoc::indoc;
6-
use lookup::lookup_v2::OptionalValuePath;
6+
use lookup::lookup_v2::{ConfigValuePath, OptionalValuePath};
77
use vector_common::sensitive_string::SensitiveString;
88
use vector_config::configurable_component;
99
use vector_core::sink::StreamSink;
@@ -97,7 +97,7 @@ pub struct HumioMetricsConfig {
9797
///
9898
/// [humio_data_format]: https://docs.humio.com/integrations/data-shippers/hec/#format-of-data
9999
#[serde(default)]
100-
indexed_fields: Vec<String>,
100+
indexed_fields: Vec<ConfigValuePath>,
101101

102102
/// Optional name of the repository to ingest into.
103103
///

‎src/sinks/kafka/config.rs

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ use std::{collections::HashMap, time::Duration};
22

33
use codecs::JsonSerializerConfig;
44
use futures::FutureExt;
5+
use lookup::lookup_v2::ConfigTargetPath;
56
use rdkafka::ClientConfig;
67
use serde_with::serde_as;
78
use vector_config::configurable_component;
@@ -53,7 +54,9 @@ pub struct KafkaSinkConfig {
5354
/// no key.
5455
#[configurable(metadata(docs::advanced))]
5556
#[configurable(metadata(docs::examples = "user_id"))]
56-
pub key_field: Option<String>,
57+
#[configurable(metadata(docs::examples = ".my_topic"))]
58+
#[configurable(metadata(docs::examples = "%my_topic"))]
59+
pub key_field: Option<ConfigTargetPath>,
5760

5861
#[configurable(derived)]
5962
pub encoding: EncodingConfig,
@@ -108,7 +111,7 @@ pub struct KafkaSinkConfig {
108111
#[configurable(metadata(docs::advanced))]
109112
#[serde(alias = "headers_field")] // accidentally released as `headers_field` in 0.18
110113
#[configurable(metadata(docs::examples = "headers"))]
111-
pub headers_key: Option<String>,
114+
pub headers_key: Option<ConfigTargetPath>,
112115

113116
#[configurable(derived)]
114117
#[serde(
@@ -249,7 +252,7 @@ impl GenerateConfig for KafkaSinkConfig {
249252
toml::Value::try_from(Self {
250253
bootstrap_servers: "10.14.22.123:9092,10.14.23.332:9092".to_owned(),
251254
topic: Template::try_from("topic-1234".to_owned()).unwrap(),
252-
key_field: Some("user_id".to_owned()),
255+
key_field: Some(ConfigTargetPath::try_from("user_id".to_owned()).unwrap()),
253256
encoding: JsonSerializerConfig::default().into(),
254257
batch: Default::default(),
255258
compression: KafkaCompression::None,

‎src/sinks/kafka/request_builder.rs

Lines changed: 15 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ use std::num::NonZeroUsize;
33
use bytes::{Bytes, BytesMut};
44
use rdkafka::message::{Header, OwnedHeaders};
55
use tokio_util::codec::Encoder as _;
6+
use vrl::path::OwnedTargetPath;
67

78
use crate::{
89
codecs::{Encoder, Transformer},
@@ -16,8 +17,8 @@ use crate::{
1617
};
1718

1819
pub struct KafkaRequestBuilder {
19-
pub key_field: Option<String>,
20-
pub headers_key: Option<String>,
20+
pub key_field: Option<OwnedTargetPath>,
21+
pub headers_key: Option<OwnedTargetPath>,
2122
pub topic_template: Template,
2223
pub transformer: Transformer,
2324
pub encoder: Encoder<()>,
@@ -39,9 +40,9 @@ impl KafkaRequestBuilder {
3940

4041
let metadata = KafkaRequestMetadata {
4142
finalizers: event.take_finalizers(),
42-
key: get_key(&event, &self.key_field),
43+
key: get_key(&event, self.key_field.as_ref()),
4344
timestamp_millis: get_timestamp_millis(&event),
44-
headers: get_headers(&event, &self.headers_key),
45+
headers: get_headers(&event, self.headers_key.as_ref()),
4546
topic,
4647
};
4748
self.transformer.transform(&mut event);
@@ -64,14 +65,12 @@ impl KafkaRequestBuilder {
6465
}
6566
}
6667

67-
fn get_key(event: &Event, key_field: &Option<String>) -> Option<Bytes> {
68-
key_field.as_ref().and_then(|key_field| match event {
69-
Event::Log(log) => log
70-
.get(key_field.as_str())
71-
.map(|value| value.coerce_to_bytes()),
68+
fn get_key(event: &Event, key_field: Option<&OwnedTargetPath>) -> Option<Bytes> {
69+
key_field.and_then(|key_field| match event {
70+
Event::Log(log) => log.get(key_field).map(|value| value.coerce_to_bytes()),
7271
Event::Metric(metric) => metric
7372
.tags()
74-
.and_then(|tags| tags.get(key_field))
73+
.and_then(|tags| tags.get(key_field.to_string().as_str()))
7574
.map(|value| value.to_owned().into()),
7675
_ => None,
7776
})
@@ -86,10 +85,10 @@ fn get_timestamp_millis(event: &Event) -> Option<i64> {
8685
.map(|ts| ts.timestamp_millis())
8786
}
8887

89-
fn get_headers(event: &Event, headers_key: &Option<String>) -> Option<OwnedHeaders> {
90-
headers_key.as_ref().and_then(|headers_key| {
88+
fn get_headers(event: &Event, headers_key: Option<&OwnedTargetPath>) -> Option<OwnedHeaders> {
89+
headers_key.and_then(|headers_key| {
9190
if let Event::Log(log) = event {
92-
if let Some(headers) = log.get(headers_key.as_str()) {
91+
if let Some(headers) = log.get(headers_key) {
9392
match headers {
9493
Value::Object(headers_map) => {
9594
let mut owned_headers = OwnedHeaders::new_with_capacity(headers_map.len());
@@ -131,15 +130,15 @@ mod tests {
131130

132131
#[test]
133132
fn kafka_get_headers() {
134-
let headers_key = "headers";
133+
let headers_key = OwnedTargetPath::try_from("headers".to_string()).unwrap();
135134
let mut header_values = BTreeMap::new();
136135
header_values.insert("a-key".to_string(), Value::Bytes(Bytes::from("a-value")));
137136
header_values.insert("b-key".to_string(), Value::Bytes(Bytes::from("b-value")));
138137

139138
let mut event = Event::Log(LogEvent::from("hello"));
140-
event.as_mut_log().insert(headers_key, header_values);
139+
event.as_mut_log().insert(&headers_key, header_values);
141140

142-
let headers = get_headers(&event, &Some(headers_key.to_string())).unwrap();
141+
let headers = get_headers(&event, Some(&headers_key)).unwrap();
143142
assert_eq!(headers.get(0).key, "a-key");
144143
assert_eq!(headers.get(0).value.unwrap(), "a-value".as_bytes());
145144
assert_eq!(headers.get(1).key, "b-key");

0 commit comments

Comments
 (0)