-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(codec,sources): influxdb line protcol decoder
Signed-off-by: Michael Hoffmann <[email protected]>
- Loading branch information
1 parent
cebe628
commit 7d57649
Showing
6 changed files
with
259 additions
and
3 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,217 @@ | ||
| use std::borrow::Cow; | ||
|
|
||
| use bytes::Bytes; | ||
| use chrono::NaiveDateTime; | ||
| use derivative::Derivative; | ||
| use influxdb_line_protocol::{FieldValue, ParsedLine}; | ||
| use smallvec::SmallVec; | ||
| use vector_config::configurable_component; | ||
| use vector_core::config::LogNamespace; | ||
| use vector_core::event::{Event, Metric, MetricKind, MetricTags, MetricValue}; | ||
| use vector_core::{config::DataType, schema}; | ||
| use vrl::value::kind::Collection; | ||
| use vrl::value::Kind; | ||
|
|
||
| use crate::decoding::format::default_lossy; | ||
|
|
||
| use super::Deserializer; | ||
|
|
||
| /// Config used to build a `InfluxdbDeserializer`. | ||
| #[configurable_component] | ||
| #[derive(Debug, Clone, Default)] | ||
| pub struct InfluxdbDeserializerConfig { | ||
| /// Influxdb-specific decoding options. | ||
| #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] | ||
| pub influxdb: InfluxdbDeserializerOptions, | ||
| } | ||
|
|
||
| impl InfluxdbDeserializerConfig { | ||
| /// new constructs a new InfluxdbDeserializerConfig | ||
| pub fn new(options: InfluxdbDeserializerOptions) -> Self { | ||
| Self { influxdb: options } | ||
| } | ||
|
|
||
| /// build constructs a new InfluxdbDeserializer | ||
| pub fn build(&self) -> InfluxdbDeserializer { | ||
| Into::<InfluxdbDeserializer>::into(self) | ||
| } | ||
|
|
||
| /// The output type produced by the deserializer. | ||
| pub fn output_type(&self) -> DataType { | ||
| DataType::Metric | ||
| } | ||
|
|
||
| /// The schema produced by the deserializer. | ||
| pub fn schema_definition(&self, log_namespace: LogNamespace) -> schema::Definition { | ||
| schema::Definition::new_with_default_metadata( | ||
| Kind::object(Collection::empty()), | ||
| [log_namespace], | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| /// Influxdb-specific decoding options. | ||
| #[configurable_component] | ||
| #[derive(Debug, Clone, PartialEq, Eq, Derivative)] | ||
| #[derivative(Default)] | ||
| pub struct InfluxdbDeserializerOptions { | ||
| /// Determines whether or not to replace invalid UTF-8 sequences instead of failing. | ||
| /// | ||
| /// When true, invalid UTF-8 sequences are replaced with the [`U+FFFD REPLACEMENT CHARACTER`][U+FFFD]. | ||
| /// | ||
| /// [U+FFFD]: https://en.wikipedia.org/wiki/Specials_(Unicode_block)#Replacement_character | ||
| #[serde( | ||
| default = "default_lossy", | ||
| skip_serializing_if = "vector_core::serde::is_default" | ||
| )] | ||
| #[derivative(Default(value = "default_lossy()"))] | ||
| pub lossy: bool, | ||
| } | ||
|
|
||
| /// Deserializer for the influxdb line protocol | ||
| #[derive(Debug, Clone, Derivative)] | ||
| #[derivative(Default)] | ||
| pub struct InfluxdbDeserializer { | ||
| #[derivative(Default(value = "default_lossy()"))] | ||
| lossy: bool, | ||
| } | ||
|
|
||
| impl InfluxdbDeserializer { | ||
| /// new constructs a new InfluxdbDeserializer | ||
| pub fn new(lossy: bool) -> Self { | ||
| Self { lossy } | ||
| } | ||
| } | ||
|
|
||
| impl Deserializer for InfluxdbDeserializer { | ||
| fn parse( | ||
| &self, | ||
| bytes: Bytes, | ||
| _log_namespace: LogNamespace, | ||
| ) -> vector_common::Result<SmallVec<[Event; 1]>> { | ||
| let line: Cow<str> = match self.lossy { | ||
| true => String::from_utf8_lossy(&bytes), | ||
| false => Cow::from(std::str::from_utf8(&bytes)?), | ||
| }; | ||
| let parsed_line = influxdb_line_protocol::parse_lines(&line); | ||
|
|
||
| let res = parsed_line | ||
| .collect::<Result<Vec<_>, _>>()? | ||
| .iter() | ||
| .flat_map(|line| { | ||
| let ParsedLine { | ||
| series, | ||
| field_set, | ||
| timestamp, | ||
| } = line; | ||
|
|
||
| field_set | ||
| .iter() | ||
| .filter_map(|f| { | ||
| let measurement = series.measurement.clone(); | ||
| let tags = series.tag_set.as_ref().clone(); | ||
| let val = match f.1 { | ||
| FieldValue::I64(v) => v as f64, | ||
| FieldValue::U64(v) => v as f64, | ||
| FieldValue::F64(v) => v, | ||
| FieldValue::Boolean(v) => { | ||
| if v { | ||
| 1.0 | ||
| } else { | ||
| 0.0 | ||
| } | ||
| } | ||
| // String values cannot be modelled in our schema | ||
| FieldValue::String(_) => return None, | ||
| }; | ||
| Some(Event::Metric( | ||
| Metric::new( | ||
| format!("{0}_{1}", measurement, f.0), | ||
| MetricKind::Absolute, | ||
| MetricValue::Gauge { value: val }, | ||
| ) | ||
| .with_tags(tags.and_then(|ts| { | ||
| Some(MetricTags::from_iter( | ||
| ts.iter().map(|t| (t.0.to_string(), t.1.to_string())), | ||
| )) | ||
| })) | ||
| .with_timestamp(timestamp.and_then(|t| { | ||
| NaiveDateTime::from_timestamp_micros(t).map(|t| t.and_utc()) | ||
| })), | ||
| )) | ||
| }) | ||
| .collect::<Vec<_>>() | ||
| }) | ||
| .collect(); | ||
|
|
||
| Ok(res) | ||
| } | ||
| } | ||
|
|
||
| impl From<&InfluxdbDeserializerConfig> for InfluxdbDeserializer { | ||
| fn from(config: &InfluxdbDeserializerConfig) -> Self { | ||
| Self { | ||
| lossy: config.influxdb.lossy, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use bytes::Bytes; | ||
| use chrono::NaiveDateTime; | ||
| use vector_core::{ | ||
| config::LogNamespace, | ||
| event::{Metric, MetricKind, MetricTags, MetricValue}, | ||
| }; | ||
|
|
||
| use crate::decoding::format::{Deserializer, InfluxdbDeserializer}; | ||
|
|
||
| #[test] | ||
| fn deserialize_success() { | ||
| let deser = InfluxdbDeserializer::new(true); | ||
| let buffer = Bytes::from( | ||
| "cpu,host=A,region=west usage_system=64i,usage_user=10i 1590488773254420000", | ||
| ); | ||
| let events = deser.parse(buffer, LogNamespace::default()).unwrap(); | ||
| assert_eq!(events.len(), 2); | ||
|
|
||
| assert_eq!( | ||
| events[0].as_metric(), | ||
| &Metric::new( | ||
| "cpu_usage_system", | ||
| MetricKind::Absolute, | ||
| MetricValue::Gauge { value: 64. }, | ||
| ) | ||
| .with_tags(Some(MetricTags::from_iter([ | ||
| ("host".to_string(), "A".to_string()), | ||
| ("region".to_string(), "west".to_string()), | ||
| ]))) | ||
| .with_timestamp( | ||
| NaiveDateTime::from_timestamp_micros(1590488773254420000).map(|t| t.and_utc()) | ||
| ) | ||
| ); | ||
| assert_eq!( | ||
| events[1].as_metric(), | ||
| &Metric::new( | ||
| "cpu_usage_user", | ||
| MetricKind::Absolute, | ||
| MetricValue::Gauge { value: 10. }, | ||
| ) | ||
| .with_tags(Some(MetricTags::from_iter([ | ||
| ("host".to_string(), "A".to_string()), | ||
| ("region".to_string(), "west".to_string()), | ||
| ]))) | ||
| .with_timestamp( | ||
| NaiveDateTime::from_timestamp_micros(1590488773254420000).map(|t| t.and_utc()) | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| #[test] | ||
| fn deserialize_error() { | ||
| let deser = InfluxdbDeserializer::new(true); | ||
| let buffer = Bytes::from("some invalid string"); | ||
| assert!(deser.parse(buffer, LogNamespace::default()).is_err()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters