vector/sinks/pulsar/
util.rs

1use std::collections::HashMap;
2
3use bytes::Bytes;
4use vector_lib::{event::Event, lookup::lookup_v2::OptionalTargetPath};
5use vrl::value::{KeyString, Value};
6
7use crate::{
8    internal_events::{PulsarPropertyExtractionError, TemplateRenderingError},
9    sinks::pulsar::{config::PulsarSinkConfig, sink::PulsarEvent},
10    template::ConfinedTemplate,
11};
12
13/// Transforms an event into a Pulsar event by rendering the required template fields.
14/// Returns None if there is an error whilst rendering.
15pub(super) fn make_pulsar_event(
16    topic: &ConfinedTemplate,
17    config: &PulsarSinkConfig,
18    event: Event,
19) -> Option<PulsarEvent> {
20    let topic = topic
21        .render_string(&event)
22        .map_err(|error| {
23            emit!(TemplateRenderingError {
24                field: Some("topic"),
25                drop_event: true,
26                error,
27            });
28        })
29        .ok()?;
30    let key = get_key(&event, &config.partition_key_field);
31    let timestamp_millis = get_timestamp_millis(&event);
32    let properties = get_properties(&event, &config.properties_key);
33    Some(PulsarEvent {
34        event,
35        topic,
36        key,
37        timestamp_millis,
38        properties,
39    })
40}
41
42fn get_key(event: &Event, partition_key_field: &Option<OptionalTargetPath>) -> Option<Bytes> {
43    partition_key_field
44        .as_ref()
45        .and_then(|partition_key_field| match event {
46            Event::Log(log) => partition_key_field
47                .path
48                .as_ref()
49                .and_then(|path| log.get(path).map(|value| value.coerce_to_bytes())),
50            Event::Metric(metric) => partition_key_field
51                .path
52                .as_ref()
53                .and_then(|path| metric.tags().and_then(|tags| tags.get(&path.to_string())))
54                .map(|value| value.to_owned().into()),
55            _ => None,
56        })
57}
58
59fn get_timestamp_millis(event: &Event) -> Option<i64> {
60    match &event {
61        Event::Log(log) => log.get_timestamp().and_then(|v| v.as_timestamp()).copied(),
62        Event::Metric(metric) => metric.timestamp(),
63        _ => None,
64    }
65    .map(|ts| ts.timestamp_millis())
66}
67
68pub(super) fn get_properties(
69    event: &Event,
70    properties_key: &Option<OptionalTargetPath>,
71) -> Option<HashMap<KeyString, Bytes>> {
72    properties_key.as_ref().and_then(|properties_key| {
73        properties_key.path.as_ref().and_then(|path| {
74            event.maybe_as_log().and_then(|log| {
75                log.get(path).and_then(|properties| match properties {
76                    Value::Object(headers_map) => {
77                        let mut property_map = HashMap::new();
78                        for (key, value) in headers_map {
79                            if let Value::Bytes(value_bytes) = value {
80                                property_map.insert(key.clone(), value_bytes.clone());
81                            } else {
82                                emit!(PulsarPropertyExtractionError {
83                                    property_field: path
84                                });
85                            }
86                        }
87                        Some(property_map)
88                    }
89                    _ => {
90                        emit!(PulsarPropertyExtractionError {
91                            property_field: path
92                        });
93                        None
94                    }
95                })
96            })
97        })
98    })
99}
vector/sinks/pulsar/util.rs

vector/sinks/pulsar/
util.rs
