vector/sinks/sematext/
metrics.rs

1#![expect(
2    clippy::let_underscore_must_use,
3    reason = "derivative's Debug derive with ignored fields expands to a must_use let binding"
4)]
5
6use std::{collections::HashMap, future::ready, task::Poll};
7
8use bytes::{Bytes, BytesMut};
9use derivative::Derivative;
10use futures::{FutureExt, SinkExt, future::BoxFuture, stream};
11use http::{StatusCode, Uri};
12use hyper::{Body, Request};
13use indoc::indoc;
14use tower::Service;
15use vector_lib::{
16    ByteSizeOf, EstimatedJsonEncodedSizeOf, configurable::configurable_component,
17    sensitive_string::SensitiveString,
18};
19
20use super::Region;
21use crate::{
22    Result,
23    config::{
24        AcknowledgementsConfig, DynValidatedSink, GenerateConfig, Input, SinkConfig, SinkContext,
25        ValidatedSink,
26    },
27    event::{
28        Event, KeyString,
29        metric::{Metric, MetricValue},
30    },
31    http::HttpClient,
32    internal_events::{SematextMetricsEncodeEventError, SematextMetricsInvalidMetricError},
33    sinks::{
34        Healthcheck, HealthcheckError, VectorSink,
35        influxdb::{Field, ProtocolVersion, encode_timestamp, encode_uri, influx_line_protocol},
36        util::{
37            BatchConfig, BatchSettings, EncodedEvent, HttpEndpoint, SinkBatchSettings,
38            TowerRequestConfig,
39            buffer::metrics::{MetricNormalize, MetricNormalizer, MetricSet, MetricsBuffer},
40            http::{HttpBatchService, HttpRetryLogic},
41        },
42    },
43    vector_version,
44};
45
46#[derive(Clone)]
47struct SematextMetricsService {
48    config: SematextMetricsConfig,
49    inner: HttpBatchService<BoxFuture<'static, Result<Request<Bytes>>>>,
50}
51
52#[derive(Clone, Copy, Debug, Default)]
53pub(crate) struct SematextMetricsDefaultBatchSettings;
54
55impl SinkBatchSettings for SematextMetricsDefaultBatchSettings {
56    const MAX_EVENTS: Option<usize> = Some(20);
57    const MAX_BYTES: Option<usize> = None;
58    const TIMEOUT_SECS: f64 = 1.0;
59}
60
61/// Configuration for the `sematext_metrics` sink.
62#[configurable_component(sink("sematext_metrics", "Publish metric events to Sematext."))]
63#[derive(Clone, Debug)]
64pub struct SematextMetricsConfig {
65    /// Sets the default namespace for any metrics sent.
66    ///
67    /// This namespace is only used if a metric has no existing namespace. When a namespace is
68    /// present, it is used as a prefix to the metric name, and separated with a period (`.`).
69    #[configurable(metadata(docs::examples = "service"))]
70    pub default_namespace: String,
71
72    #[serde(default = "super::default_region")]
73    #[configurable(derived)]
74    pub region: Region,
75
76    /// The endpoint to send data to.
77    ///
78    /// Setting this option overrides the `region` option.
79    #[configurable(metadata(docs::examples = "http://127.0.0.1"))]
80    #[configurable(metadata(docs::examples = "https://example.com"))]
81    pub endpoint: Option<String>,
82
83    /// The token that is used to write to Sematext.
84    #[configurable(metadata(docs::examples = "${SEMATEXT_TOKEN}"))]
85    #[configurable(metadata(docs::examples = "some-sematext-token"))]
86    pub token: SensitiveString,
87
88    #[configurable(derived)]
89    #[serde(default)]
90    pub(self) batch: BatchConfig<SematextMetricsDefaultBatchSettings>,
91
92    #[configurable(derived)]
93    #[serde(default)]
94    pub request: TowerRequestConfig,
95
96    #[configurable(derived)]
97    #[serde(
98        default,
99        deserialize_with = "crate::serde::bool_or_struct",
100        skip_serializing_if = "crate::serde::is_default"
101    )]
102    acknowledgements: AcknowledgementsConfig,
103}
104
105impl GenerateConfig for SematextMetricsConfig {
106    fn generate_config() -> serde_json::Value {
107        serde_yaml::from_str(indoc! {r#"
108            default_namespace: vector
109            token: ${SEMATEXT_TOKEN}
110        "#})
111        .unwrap()
112    }
113}
114
115async fn healthcheck(endpoint: String, client: HttpClient) -> Result<()> {
116    let uri = HttpEndpoint::parse(&endpoint)?
117        .append_path("health")?
118        .into_uri();
119
120    let request = Request::get(uri)
121        .body(Body::empty())
122        .map_err(|e| e.to_string())?;
123
124    let response = client.send(request).await?;
125
126    match response.status() {
127        StatusCode::OK => Ok(()),
128        StatusCode::NO_CONTENT => Ok(()),
129        other => Err(HealthcheckError::UnexpectedStatus { status: other }.into()),
130    }
131}
132
133// https://sematext.com/docs/monitoring/custom-metrics/
134const US_ENDPOINT: &str = "https://spm-receiver.sematext.com";
135const EU_ENDPOINT: &str = "https://spm-receiver.eu.sematext.com";
136
137#[async_trait::async_trait]
138#[typetag::serde(name = "sematext_metrics")]
139impl SinkConfig for SematextMetricsConfig {
140    fn input(&self) -> Input {
141        Input::metric()
142    }
143    fn acknowledgements(&self) -> &AcknowledgementsConfig {
144        &self.acknowledgements
145    }
146
147    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
148        Some(self)
149    }
150}
151
152#[derive(Clone, Derivative)]
153#[derivative(Debug)]
154pub struct ValidatedSematextMetrics {
155    endpoint: String,
156    uri: Uri,
157    #[derivative(Debug = "ignore")]
158    batch: BatchSettings<MetricsBuffer>,
159}
160
161#[async_trait::async_trait]
162impl ValidatedSink for SematextMetricsConfig {
163    type Validated = ValidatedSematextMetrics;
164
165    fn validate(&self) -> Result<ValidatedSematextMetrics> {
166        let endpoint = match (&self.endpoint, &self.region) {
167            (Some(endpoint), _) => endpoint.clone(),
168            (None, Region::Us) => US_ENDPOINT.to_owned(),
169            (None, Region::Eu) => EU_ENDPOINT.to_owned(),
170        };
171
172        let uri = write_uri(&HttpEndpoint::parse(&endpoint)?)?;
173        let batch = self.batch.into_batch_settings()?;
174
175        Ok(ValidatedSematextMetrics {
176            endpoint,
177            uri,
178            batch,
179        })
180    }
181
182    async fn build(
183        &self,
184        validated: &ValidatedSematextMetrics,
185        cx: SinkContext,
186    ) -> Result<(VectorSink, Healthcheck)> {
187        let ValidatedSematextMetrics { endpoint, .. } = validated;
188
189        let client = HttpClient::new(None, cx.proxy())?;
190
191        let healthcheck = healthcheck(endpoint.clone(), client.clone()).boxed();
192        let sink = SematextMetricsService::from_validated(self.clone(), validated, client)?;
193
194        Ok((sink, healthcheck))
195    }
196}
197
198fn write_uri(endpoint: &HttpEndpoint) -> Result<Uri> {
199    encode_uri(
200        endpoint,
201        "write",
202        &[
203            ("db", Some("metrics".into())),
204            ("v", Some(format!("vector-{}", vector_version()))),
205            ("precision", Some("ns".into())),
206        ],
207    )
208}
209
210impl SematextMetricsService {
211    fn from_validated(
212        config: SematextMetricsConfig,
213        validated: &ValidatedSematextMetrics,
214        client: HttpClient,
215    ) -> Result<VectorSink> {
216        let ValidatedSematextMetrics { uri, batch, .. } = validated;
217        let request = config.request.into_settings();
218        let http_service = HttpBatchService::new(client, create_build_request(uri.clone()));
219        let sematext_service = SematextMetricsService {
220            config,
221            inner: http_service,
222        };
223        let mut normalizer = MetricNormalizer::<SematextMetricNormalize>::default();
224
225        let sink = request
226            .batch_sink(
227                HttpRetryLogic::default(),
228                sematext_service,
229                MetricsBuffer::new(batch.size),
230                batch.timeout,
231            )
232            .with_flat_map(move |event: Event| {
233                stream::iter({
234                    let byte_size = event.size_of();
235                    let json_byte_size = event.estimated_json_encoded_size_of();
236                    normalizer
237                        .normalize(event.into_metric())
238                        .map(|item| Ok(EncodedEvent::new(item, byte_size, json_byte_size)))
239                })
240            })
241            .sink_map_err(|error| error!(message = "Fatal sematext metrics sink error.", %error, internal_log_rate_limit = false));
242
243        #[allow(deprecated)]
244        Ok(VectorSink::from_event_sink(sink))
245    }
246}
247
248impl Service<Vec<Metric>> for SematextMetricsService {
249    type Response = http::Response<bytes::Bytes>;
250    type Error = crate::Error;
251    type Future = BoxFuture<'static, std::result::Result<Self::Response, Self::Error>>;
252
253    fn poll_ready(
254        &mut self,
255        cx: &mut std::task::Context,
256    ) -> Poll<std::result::Result<(), Self::Error>> {
257        self.inner.poll_ready(cx)
258    }
259
260    fn call(&mut self, items: Vec<Metric>) -> Self::Future {
261        let input = encode_events(
262            self.config.token.inner(),
263            &self.config.default_namespace,
264            items,
265        );
266        let body = input.item;
267
268        self.inner.call(body)
269    }
270}
271
272#[derive(Default)]
273struct SematextMetricNormalize;
274
275impl MetricNormalize for SematextMetricNormalize {
276    fn normalize(&mut self, state: &mut MetricSet, metric: Metric) -> Option<Metric> {
277        match &metric.value() {
278            MetricValue::Gauge { .. } => state.make_absolute(metric),
279            MetricValue::Counter { .. } => state.make_incremental(metric),
280            _ => {
281                emit!(SematextMetricsInvalidMetricError { metric: &metric });
282                None
283            }
284        }
285    }
286}
287
288fn create_build_request(
289    uri: http::Uri,
290) -> impl Fn(Bytes) -> BoxFuture<'static, Result<Request<Bytes>>> + Sync + Send + 'static {
291    move |body| {
292        Box::pin(ready(
293            Request::post(uri.clone())
294                .header("Content-Type", "text/plain")
295                .body(body)
296                .map_err(Into::into),
297        ))
298    }
299}
300
301fn encode_events(
302    token: &str,
303    default_namespace: &str,
304    metrics: Vec<Metric>,
305) -> EncodedEvent<Bytes> {
306    let mut output = BytesMut::new();
307    let byte_size = metrics.size_of();
308    let json_byte_size = metrics.estimated_json_encoded_size_of();
309    for metric in metrics.into_iter() {
310        let (series, data, _metadata) = metric.into_parts();
311        let namespace = series
312            .name
313            .namespace
314            .unwrap_or_else(|| default_namespace.into());
315        let label = series.name.name;
316        let ts = encode_timestamp(data.time.timestamp);
317
318        // Authentication in Sematext is by inserting the token as a tag.
319        let mut tags = series.tags.unwrap_or_default();
320        tags.replace("token".into(), token.to_string());
321        let (metric_type, fields) = match data.value {
322            MetricValue::Counter { value } => ("counter", to_fields(label, value)),
323            MetricValue::Gauge { value } => ("gauge", to_fields(label, value)),
324            _ => unreachable!(), // handled by SematextMetricNormalize
325        };
326
327        tags.replace("metric_type".into(), metric_type.to_string());
328
329        if let Err(error) = influx_line_protocol(
330            ProtocolVersion::V1,
331            &namespace,
332            Some(tags),
333            Some(fields),
334            ts,
335            &mut output,
336        ) {
337            emit!(SematextMetricsEncodeEventError { error });
338        };
339    }
340
341    if !output.is_empty() {
342        output.truncate(output.len() - 1);
343    }
344    EncodedEvent::new(output.freeze(), byte_size, json_byte_size)
345}
346
347fn to_fields(label: String, value: f64) -> HashMap<KeyString, Field> {
348    let mut result = HashMap::new();
349    result.insert(label.into(), Field::Float(value));
350    result
351}
352
353#[cfg(test)]
354mod tests {
355    use chrono::{Timelike, Utc, offset::TimeZone};
356    use futures::StreamExt;
357    use indoc::indoc;
358    use vector_lib::metric_tags;
359
360    use super::*;
361    use crate::{
362        config::ValidatedSink,
363        event::{Event, metric::MetricKind},
364        sinks::util::test::{build_test_server, load_sink},
365        test_util::{
366            addr::next_addr,
367            components::{HTTP_SINK_TAGS, assert_sink_compliance},
368            test_generate_config,
369        },
370    };
371
372    #[test]
373    fn generate_config() {
374        test_generate_config::<SematextMetricsConfig>();
375    }
376
377    #[test]
378    fn prepares_valid_config() {
379        let config = SematextMetricsConfig {
380            default_namespace: "ns".to_string(),
381            region: Region::Us,
382            endpoint: Some("http://localhost:9999".to_string()),
383            token: "atoken".to_string().into(),
384            batch: Default::default(),
385            request: Default::default(),
386            acknowledgements: Default::default(),
387        };
388
389        let validated = config.validate().expect("preparation should succeed");
390        assert_eq!(validated.endpoint, "http://localhost:9999");
391        assert!(
392            validated
393                .uri
394                .to_string()
395                .starts_with("http://localhost:9999/write?db=metrics&v=vector-")
396        );
397    }
398
399    #[test]
400    fn rejects_non_absolute_endpoint() {
401        let config = |endpoint: &str| SematextMetricsConfig {
402            default_namespace: "ns".to_string(),
403            region: Region::Us,
404            endpoint: Some(endpoint.to_string()),
405            token: "atoken".to_string().into(),
406            batch: Default::default(),
407            request: Default::default(),
408            acknowledgements: Default::default(),
409        };
410
411        // Non-http(s) schemes are rejected by the HttpEndpoint type-level validation.
412        assert!(
413            config("ftp://spm-receiver.sematext.com")
414                .validate()
415                .is_err()
416        );
417        // Relative paths cannot be resolved to an absolute http(s) URL.
418        assert!(config("/write").validate().is_err());
419    }
420
421    #[test]
422    fn test_encode_counter_event() {
423        let events = vec![
424            Metric::new(
425                "pool.used",
426                MetricKind::Incremental,
427                MetricValue::Counter { value: 42.0 },
428            )
429            .with_namespace(Some("jvm"))
430            .with_timestamp(Some(
431                Utc.with_ymd_and_hms(2020, 8, 18, 21, 0, 0)
432                    .single()
433                    .expect("invalid timestamp"),
434            )),
435        ];
436
437        assert_eq!(
438            "jvm,metric_type=counter,token=aaa pool.used=42 1597784400000000000",
439            encode_events("aaa", "ns", events).item
440        );
441    }
442
443    #[test]
444    fn test_encode_counter_event_no_namespace() {
445        let events = vec![
446            Metric::new(
447                "used",
448                MetricKind::Incremental,
449                MetricValue::Counter { value: 42.0 },
450            )
451            .with_timestamp(Some(
452                Utc.with_ymd_and_hms(2020, 8, 18, 21, 0, 0)
453                    .single()
454                    .expect("invalid timestamp"),
455            )),
456        ];
457
458        assert_eq!(
459            "ns,metric_type=counter,token=aaa used=42 1597784400000000000",
460            encode_events("aaa", "ns", events).item
461        );
462    }
463
464    #[test]
465    fn test_encode_counter_multiple_events() {
466        let events = vec![
467            Metric::new(
468                "pool.used",
469                MetricKind::Incremental,
470                MetricValue::Counter { value: 42.0 },
471            )
472            .with_namespace(Some("jvm"))
473            .with_timestamp(Some(
474                Utc.with_ymd_and_hms(2020, 8, 18, 21, 0, 0)
475                    .single()
476                    .expect("invalid timestamp"),
477            )),
478            Metric::new(
479                "pool.committed",
480                MetricKind::Incremental,
481                MetricValue::Counter { value: 18874368.0 },
482            )
483            .with_namespace(Some("jvm"))
484            .with_timestamp(Some(
485                Utc.with_ymd_and_hms(2020, 8, 18, 21, 0, 0)
486                    .single()
487                    .and_then(|t| t.with_nanosecond(1))
488                    .expect("invalid timestamp"),
489            )),
490        ];
491
492        assert_eq!(
493            "jvm,metric_type=counter,token=aaa pool.used=42 1597784400000000000\n\
494             jvm,metric_type=counter,token=aaa pool.committed=18874368 1597784400000000001",
495            encode_events("aaa", "ns", events).item
496        );
497    }
498
499    #[tokio::test]
500    async fn smoke() {
501        assert_sink_compliance(&HTTP_SINK_TAGS, async {
502
503        let (mut config, cx) = load_sink::<SematextMetricsConfig>(indoc! {r#"
504            default_namespace = "ns"
505            token = "atoken"
506            batch.max_events = 1
507        "#})
508        .unwrap();
509
510        let (_guard, addr) = next_addr();
511        // Swap out the endpoint so we can force send it
512        // to our local server
513        let endpoint = format!("http://{addr}");
514        config.endpoint = Some(endpoint.clone());
515
516        let (sink, _) = SinkConfig::build(&config, cx).await.unwrap();
517
518        let (rx, _trigger, server) = build_test_server(addr);
519        tokio::spawn(server);
520
521        // Make our test metrics.
522        let metrics = vec![
523            ("os", "swap.size", 324292.0),
524            ("os", "network.tx", 42000.0),
525            ("os", "network.rx", 54293.0),
526            ("process", "count", 12.0),
527            ("process", "uptime", 32423.0),
528            ("process", "rss", 2342333.0),
529            ("jvm", "pool.used", 18874368.0),
530            ("jvm", "pool.committed", 18868584.0),
531            ("jvm", "pool.max", 18874368.0),
532        ];
533
534        let mut events = Vec::new();
535        for (i, (namespace, metric, val)) in metrics.iter().enumerate() {
536            let event = Event::from(
537                Metric::new(
538                    *metric,
539                    MetricKind::Incremental,
540                    MetricValue::Counter { value: *val },
541                )
542                .with_namespace(Some(*namespace))
543                .with_tags(Some(metric_tags!("os.host" => "somehost")))
544                    .with_timestamp(Some(Utc.with_ymd_and_hms(2020, 8, 18, 21, 0, 0).single()
545                                         .and_then(|t| t.with_nanosecond(i as u32))
546                                         .expect("invalid timestamp"))),
547            );
548            events.push(event);
549        }
550
551        sink.run_events(events).await.unwrap();
552
553        let output = rx.take(metrics.len()).collect::<Vec<_>>().await;
554        assert_eq!("os,metric_type=counter,os.host=somehost,token=atoken swap.size=324292 1597784400000000000", output[0].1);
555        assert_eq!("os,metric_type=counter,os.host=somehost,token=atoken network.tx=42000 1597784400000000001", output[1].1);
556        assert_eq!("os,metric_type=counter,os.host=somehost,token=atoken network.rx=54293 1597784400000000002", output[2].1);
557        assert_eq!("process,metric_type=counter,os.host=somehost,token=atoken count=12 1597784400000000003", output[3].1);
558        assert_eq!("process,metric_type=counter,os.host=somehost,token=atoken uptime=32423 1597784400000000004", output[4].1);
559        assert_eq!("process,metric_type=counter,os.host=somehost,token=atoken rss=2342333 1597784400000000005", output[5].1);
560        assert_eq!("jvm,metric_type=counter,os.host=somehost,token=atoken pool.used=18874368 1597784400000000006", output[6].1);
561        assert_eq!("jvm,metric_type=counter,os.host=somehost,token=atoken pool.committed=18868584 1597784400000000007", output[7].1);
562        assert_eq!("jvm,metric_type=counter,os.host=somehost,token=atoken pool.max=18874368 1597784400000000008", output[8].1);
563        }).await;
564    }
565}
vector/sinks/sematext/metrics.rs

vector/sinks/sematext/
metrics.rs
