vector/sinks/splunk_hec/metrics/
sink.rs

1use std::{fmt, sync::Arc};
2
3use serde::Serialize;
4use vector_lib::event::{Metric, MetricValue};
5use vrl::path::OwnedValuePath;
6
7use super::request_builder::HecMetricsRequestBuilder;
8use crate::{
9    internal_events::{SplunkInvalidMetricReceivedError, TemplateRenderingError},
10    sinks::{
11        prelude::*,
12        splunk_hec::common::request::HecRequest,
13        util::{encode_namespace, processed_event::ProcessedEvent},
14    },
15};
16
17pub struct HecMetricsSink<S> {
18    pub service: S,
19    pub batch_settings: BatcherSettings,
20    pub request_builder: HecMetricsRequestBuilder,
21    pub sourcetype: Option<Template>,
22    pub source: Option<Template>,
23    pub index: Option<Template>,
24    pub host_key: Option<OwnedValuePath>,
25    pub default_namespace: Option<String>,
26}
27
28impl<S> HecMetricsSink<S>
29where
30    S: Service<HecRequest> + Send + 'static,
31    S::Future: Send + 'static,
32    S::Response: DriverResponse + Send + 'static,
33    S::Error: fmt::Debug + Into<crate::Error> + Send,
34{
35    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
36        let sourcetype = self.sourcetype.as_ref();
37        let source = self.source.as_ref();
38        let index = self.index.as_ref();
39        let host_key = self.host_key.as_ref();
40        let default_namespace = self.default_namespace.as_deref();
41        let batch_settings = self.batch_settings;
42
43        input
44            .map(|event| (event.size_of(), event.into_metric()))
45            .filter_map(move |(event_byte_size, metric)| {
46                future::ready(process_metric(
47                    metric,
48                    event_byte_size,
49                    sourcetype,
50                    source,
51                    index,
52                    host_key,
53                    default_namespace,
54                ))
55            })
56            .batched_partitioned(EventPartitioner, batch_settings.timeout, |_| {
57                batch_settings.as_byte_size_config()
58            })
59            .request_builder(
60                default_request_builder_concurrency_limit(),
61                self.request_builder,
62            )
63            .filter_map(|request| async move {
64                match request {
65                    Err(e) => {
66                        error!("Failed to build HEC Metrics request: {:?}.", e);
67                        None
68                    }
69                    Ok(req) => Some(req),
70                }
71            })
72            .into_driver(self.service)
73            .run()
74            .await
75    }
76}
77
78#[async_trait]
79impl<S> StreamSink<Event> for HecMetricsSink<S>
80where
81    S: Service<HecRequest> + Send + 'static,
82    S::Future: Send + 'static,
83    S::Response: DriverResponse + Send + 'static,
84    S::Error: fmt::Debug + Into<crate::Error> + Send,
85{
86    async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
87        self.run_inner(input).await
88    }
89}
90
91#[derive(Default)]
92struct EventPartitioner;
93
94impl Partitioner for EventPartitioner {
95    type Item = HecProcessedEvent;
96    type Key = Option<Arc<str>>;
97
98    fn partition(&self, item: &Self::Item) -> Self::Key {
99        item.event.metadata().splunk_hec_token()
100    }
101}
102
103#[derive(Serialize)]
104pub struct HecMetricsProcessedEventMetadata {
105    pub event_byte_size: usize,
106    pub sourcetype: Option<String>,
107    pub source: Option<String>,
108    pub index: Option<String>,
109    pub host: Option<String>,
110    pub metric_name: String,
111    pub metric_value: f64,
112    pub templated_field_keys: Vec<String>,
113}
114
115impl ByteSizeOf for HecMetricsProcessedEventMetadata {
116    fn allocated_bytes(&self) -> usize {
117        self.sourcetype.allocated_bytes()
118            + self.source.allocated_bytes()
119            + self.index.allocated_bytes()
120            + self.host.allocated_bytes()
121            + self.metric_name.allocated_bytes()
122            + self.templated_field_keys.allocated_bytes()
123    }
124}
125
126impl HecMetricsProcessedEventMetadata {
127    fn extract_metric_name(metric: &Metric, default_namespace: Option<&str>) -> String {
128        encode_namespace(metric.namespace().or(default_namespace), '.', metric.name())
129    }
130
131    fn extract_metric_value(metric: &Metric) -> Option<f64> {
132        match *metric.value() {
133            MetricValue::Counter { value } => Some(value),
134            MetricValue::Gauge { value } => Some(value),
135            _ => {
136                emit!(SplunkInvalidMetricReceivedError {
137                    value: metric.value(),
138                    kind: &metric.kind(),
139                    error: "Metric kind not supported.".into(),
140                });
141                None
142            }
143        }
144    }
145}
146
147pub type HecProcessedEvent = ProcessedEvent<Metric, HecMetricsProcessedEventMetadata>;
148
149pub fn process_metric(
150    metric: Metric,
151    event_byte_size: usize,
152    sourcetype: Option<&Template>,
153    source: Option<&Template>,
154    index: Option<&Template>,
155    host_key: Option<&OwnedValuePath>,
156    default_namespace: Option<&str>,
157) -> Option<HecProcessedEvent> {
158    let templated_field_keys = [index.as_ref(), source.as_ref(), sourcetype.as_ref()]
159        .iter()
160        .flatten()
161        .filter_map(|t| t.get_fields())
162        .flatten()
163        .map(|f| f.replace("tags.", ""))
164        .collect::<Vec<_>>();
165    let metric_name =
166        HecMetricsProcessedEventMetadata::extract_metric_name(&metric, default_namespace);
167    let metric_value = HecMetricsProcessedEventMetadata::extract_metric_value(&metric)?;
168
169    // Render routing templates; Confined errors are intentional security drops.
170    // Returns Some(Some(value)) on success, Some(None) on non-fatal error
171    // (field omitted), or None to signal the whole event must be dropped.
172    let render = |tpl: Option<&Template>, field: &'static str| -> Option<Option<String>> {
173        let Some(t) = tpl else { return Some(None) };
174        match t.render_string(&metric) {
175            Ok(s) => Some(Some(s)),
176            Err(error @ crate::template::TemplateRenderingError::Confined { .. }) => {
177                emit!(TemplateRenderingError {
178                    error,
179                    field: Some(field),
180                    drop_event: true,
181                });
182                None
183            }
184            Err(error) => {
185                emit!(TemplateRenderingError {
186                    error,
187                    field: Some(field),
188                    drop_event: false,
189                });
190                Some(None)
191            }
192        }
193    };
194
195    let sourcetype = render(sourcetype, "sourcetype")?;
196    let source = render(source, "source")?;
197    let index = render(index, "index")?;
198    let host = host_key.and_then(|key| metric.tag_value(key.to_string().as_str()));
199
200    let metadata = HecMetricsProcessedEventMetadata {
201        event_byte_size,
202        sourcetype,
203        source,
204        index,
205        host,
206        metric_name,
207        metric_value,
208        templated_field_keys,
209    };
210
211    Some(HecProcessedEvent {
212        event: metric,
213        metadata,
214    })
215}
216
217impl EventCount for HecProcessedEvent {
218    fn event_count(&self) -> usize {
219        // A HecProcessedEvent is mapped one-to-one with an event.
220        1
221    }
222}
vector/sinks/splunk_hec/metrics/sink.rs

vector/sinks/splunk_hec/metrics/
sink.rs
