saluki_io/deser/codec/dogstatsd/
service_check.rs

1use nom::{
2    bytes::complete::tag,
3    character::complete::u8 as parse_u8,
4    combinator::all_consuming,
5    error::{Error, ErrorKind},
6    sequence::{preceded, separated_pair},
7    IResult, Parser as _,
8};
9use saluki_context::{origin::OriginTagCardinality, tags::RawTags};
10use saluki_core::data_model::event::service_check::*;
11use stringtheory::MetaString;
12
13use super::{helpers::*, DogstatsdCodecConfiguration};
14
15/// A DogStatsD service check packet.
16pub struct ServiceCheckPacket<'a> {
17    pub name: MetaString,
18    pub status: CheckStatus,
19    pub timestamp: Option<u64>,
20    pub hostname: Option<&'a str>,
21    pub message: Option<&'a str>,
22    pub tags: RawTags<'a>,
23    pub local_data: Option<&'a str>,
24    pub external_data: Option<&'a str>,
25    pub cardinality: Option<OriginTagCardinality>,
26}
27
28pub fn parse_dogstatsd_service_check<'a>(
29    input: &'a [u8], config: &DogstatsdCodecConfiguration,
30) -> IResult<&'a [u8], ServiceCheckPacket<'a>> {
31    let (remaining, (name, raw_check_status)) = preceded(
32        tag(SERVICE_CHECK_PREFIX),
33        separated_pair(ascii_alphanum_and_seps, tag("|"), parse_u8),
34    )
35    .parse(input)?;
36
37    let check_status =
38        CheckStatus::try_from(raw_check_status).map_err(|_| nom::Err::Error(Error::new(input, ErrorKind::Verify)))?;
39
40    let mut maybe_timestamp = None;
41    let mut maybe_hostname = None;
42    let mut maybe_tags = None;
43    let mut maybe_message = None;
44    let mut maybe_local_data = None;
45    let mut maybe_external_data = None;
46    let mut maybe_cardinality = None;
47
48    let remaining = if !remaining.is_empty() {
49        let (mut remaining, _) = tag("|")(remaining)?;
50        while let Some((chunk, tail)) = split_at_delimiter(remaining, b'|') {
51            if chunk.len() < 2 {
52                break;
53            }
54
55            match &chunk[..2] {
56                // Timestamp: client-provided timestamp for the event, relative to the Unix epoch, in seconds.
57                TIMESTAMP_PREFIX => {
58                    let (_, timestamp) = all_consuming(preceded(tag(TIMESTAMP_PREFIX), unix_timestamp)).parse(chunk)?;
59                    maybe_timestamp = Some(timestamp);
60                }
61                // Hostname: client-provided hostname for the host that this service check originated from.
62                HOSTNAME_PREFIX => {
63                    let (_, hostname) =
64                        all_consuming(preceded(tag(HOSTNAME_PREFIX), ascii_alphanum_and_seps)).parse(chunk)?;
65                    maybe_hostname = Some(hostname);
66                }
67                // Local Data: client-provided data used for resolving the entity ID that this service check originated from.
68                LOCAL_DATA_PREFIX => {
69                    let (_, local_data) = all_consuming(preceded(tag(LOCAL_DATA_PREFIX), local_data)).parse(chunk)?;
70                    maybe_local_data = Some(local_data);
71                }
72                // External Data: client-provided data used for resolving the entity ID that this service check originated from.
73                EXTERNAL_DATA_PREFIX => {
74                    let (_, external_data) =
75                        all_consuming(preceded(tag(EXTERNAL_DATA_PREFIX), external_data)).parse(chunk)?;
76                    maybe_external_data = Some(external_data);
77                }
78                // Tags: additional tags to be added to the service check.
79                _ if chunk.starts_with(TAGS_PREFIX) => {
80                    let (_, tags) = all_consuming(preceded(tag(TAGS_PREFIX), tags(config))).parse(chunk)?;
81                    maybe_tags = Some(tags);
82                }
83                // Message: A message describing the current state of the service check.
84                SERVICE_CHECK_MESSAGE_PREFIX => {
85                    let (_, message) = all_consuming(preceded(tag(SERVICE_CHECK_MESSAGE_PREFIX), utf8)).parse(chunk)?;
86                    maybe_message = Some(message);
87                }
88                // Cardinality: client-provided cardinality for the service check.
89                _ if chunk.starts_with(CARDINALITY_PREFIX) => {
90                    let (_, cardinality) = cardinality(chunk)?;
91                    maybe_cardinality = cardinality;
92                }
93                _ => {
94                    // We don't know what this is, so we just skip it.
95                    //
96                    // TODO: Should we throw an error, warn, or be silently permissive?
97                }
98            }
99            remaining = tail;
100        }
101        remaining
102    } else {
103        remaining
104    };
105
106    let tags = maybe_tags.unwrap_or_else(RawTags::empty);
107
108    let service_check_packet = ServiceCheckPacket {
109        name: name.into(),
110        status: check_status,
111        tags,
112        timestamp: maybe_timestamp,
113        hostname: maybe_hostname,
114        message: maybe_message,
115        local_data: maybe_local_data,
116        external_data: maybe_external_data,
117        cardinality: maybe_cardinality,
118    };
119    Ok((remaining, service_check_packet))
120}
121
122#[cfg(test)]
123mod tests {
124    use nom::IResult;
125    use saluki_context::{
126        origin::OriginTagCardinality,
127        tags::{SharedTagSet, TagSet},
128    };
129    use saluki_core::data_model::event::service_check::{CheckStatus, ServiceCheck};
130    use stringtheory::MetaString;
131
132    use super::{parse_dogstatsd_service_check, DogstatsdCodecConfiguration};
133
134    type NomResult<'input, T> = Result<T, nom::Err<nom::error::Error<&'input [u8]>>>;
135
136    fn parse_dsd_service_check(input: &[u8]) -> NomResult<'_, ServiceCheck> {
137        let default_config = DogstatsdCodecConfiguration::default();
138        parse_dsd_service_check_with_conf(input, &default_config)
139    }
140
141    fn parse_dsd_service_check_with_conf<'input>(
142        input: &'input [u8], config: &DogstatsdCodecConfiguration,
143    ) -> NomResult<'input, ServiceCheck> {
144        let (remaining, service_check) = parse_dsd_service_check_direct(input, config)?;
145        assert!(remaining.is_empty());
146
147        Ok(service_check)
148    }
149
150    fn parse_dsd_service_check_direct<'input>(
151        input: &'input [u8], config: &DogstatsdCodecConfiguration,
152    ) -> IResult<&'input [u8], ServiceCheck> {
153        let (remaining, packet) = parse_dogstatsd_service_check(input, config)?;
154        assert!(remaining.is_empty());
155
156        let mut service_check_tags = TagSet::default();
157        for tag in packet.tags.into_iter() {
158            service_check_tags.insert_tag(tag);
159        }
160
161        let service_check = ServiceCheck::new(packet.name, packet.status)
162            .with_timestamp(packet.timestamp)
163            .with_hostname(packet.hostname.map(|s| s.into()))
164            .with_tags(service_check_tags)
165            .with_message(packet.message.map(|s| s.into()));
166
167        Ok((remaining, service_check))
168    }
169
170    #[track_caller]
171    fn check_basic_service_check_eq(expected: ServiceCheck, actual: ServiceCheck) {
172        assert_eq!(expected.name(), actual.name());
173        assert_eq!(expected.status(), actual.status());
174        assert_eq!(expected.timestamp(), actual.timestamp());
175        assert_eq!(expected.hostname(), actual.hostname());
176        assert_eq!(expected.tags(), actual.tags());
177        assert_eq!(expected.message(), actual.message());
178        assert_eq!(expected.origin_tags(), actual.origin_tags());
179    }
180
181    #[test]
182    fn basic_service_checks() {
183        let name = "testsvc";
184        let sc_status = CheckStatus::Warning;
185        let raw = format!("_sc|{}|{}", name, sc_status.as_u8());
186        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
187        let expected = ServiceCheck::new(name, sc_status);
188        check_basic_service_check_eq(expected, actual);
189    }
190
191    #[test]
192    fn service_check_timestamp() {
193        let name = "testsvc";
194        let sc_status = CheckStatus::Warning;
195        let sc_timestamp = 1234567890;
196        let raw = format!("_sc|{}|{}|d:{}", name, sc_status.as_u8(), sc_timestamp);
197        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
198        let expected = ServiceCheck::new(name, sc_status).with_timestamp(sc_timestamp);
199        check_basic_service_check_eq(expected, actual);
200    }
201
202    #[test]
203    fn service_check_tags() {
204        let name = "testsvc";
205        let sc_status = CheckStatus::Warning;
206        let tags = ["tag1", "tag2"];
207        let raw = format!("_sc|{}|{}|#{}", name, sc_status.as_u8(), tags.join(","));
208        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
209        let shared_tag_set = SharedTagSet::from(TagSet::from_iter(tags.iter().map(|&s| s.into())));
210        let expected = ServiceCheck::new(name, sc_status).with_tags(shared_tag_set);
211        check_basic_service_check_eq(expected, actual);
212    }
213
214    #[test]
215    fn service_check_message() {
216        let name = "testsvc";
217        let sc_status = CheckStatus::Ok;
218        let sc_message = MetaString::from("service running properly");
219        let raw = format!("_sc|{}|{}|m:{}", name, sc_status.as_u8(), sc_message);
220        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
221        let expected = ServiceCheck::new(name, sc_status).with_message(sc_message);
222        check_basic_service_check_eq(expected, actual);
223    }
224
225    #[test]
226    fn service_check_fields_after_message() {
227        let name = "testsvc";
228        let sc_status = CheckStatus::Ok;
229        let sc_message = MetaString::from("service running properly");
230        let sc_local_data = "ci-1234567890";
231        let sc_external_data = "it-false,cn-redis,pu-810fe89d-da47-410b-8979-9154a40f8183";
232        let raw = format!(
233            "_sc|{}|{}|#tag1,tag2|m:{}|c:{}|e:{}",
234            name,
235            sc_status.as_u8(),
236            sc_message,
237            sc_local_data,
238            sc_external_data,
239        );
240        let result = parse_dsd_service_check(raw.as_bytes()).unwrap();
241        let tags = ["tag1", "tag2"];
242        let shared_tag_set = SharedTagSet::from(TagSet::from_iter(tags.iter().map(|&s| s.into())));
243        let expected = ServiceCheck::new(name, sc_status)
244            .with_message(sc_message)
245            .with_tags(shared_tag_set);
246        check_basic_service_check_eq(expected, result);
247
248        let config = DogstatsdCodecConfiguration::default();
249        let (_, packet) = parse_dogstatsd_service_check(raw.as_bytes(), &config).expect("should not fail to parse");
250        assert_eq!(packet.local_data, Some(sc_local_data));
251        assert_eq!(packet.external_data, Some(sc_external_data));
252    }
253
254    #[test]
255    fn service_check_multiple_extensions() {
256        let name = "testsvc";
257        let sc_status = CheckStatus::Unknown;
258        let sc_timestamp = 1234567890;
259        let sc_hostname = MetaString::from("myhost");
260        let sc_local_data = "abcdef123456";
261        let sc_external_data = "it-false,cn-redis,pu-810fe89d-da47-410b-8979-9154a40f8183";
262        let tags = ["tag1", "tag2"];
263        let sc_message = MetaString::from("service status unknown");
264        let sc_cardinality = "none";
265        let raw = format!(
266            "_sc|{}|{}|d:{}|h:{}|c:{}|e:{}|card:{}|#{}|m:{}",
267            name,
268            sc_status.as_u8(),
269            sc_timestamp,
270            sc_hostname,
271            sc_local_data,
272            sc_external_data,
273            sc_cardinality,
274            tags.join(","),
275            sc_message
276        );
277        let shared_tag_set = SharedTagSet::from(TagSet::from_iter(tags.iter().map(|&s| s.into())));
278        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
279        let expected = ServiceCheck::new(name, sc_status)
280            .with_timestamp(sc_timestamp)
281            .with_hostname(sc_hostname)
282            .with_tags(shared_tag_set)
283            .with_message(sc_message);
284        check_basic_service_check_eq(expected, actual);
285
286        let config = DogstatsdCodecConfiguration::default();
287        let (_, packet) = parse_dogstatsd_service_check(raw.as_bytes(), &config).expect("should not fail to parse");
288        assert_eq!(packet.local_data, Some(sc_local_data));
289        assert_eq!(packet.external_data, Some(sc_external_data));
290        assert_eq!(packet.cardinality, Some(OriginTagCardinality::None));
291    }
292
293    #[test]
294    fn service_check_semi_real_payload_kafka() {
295        let raw_payload = "_sc|kafka.can_connect|2|#env:staging,service:datadog-agent,dd.internal.entity_id:none,dd.internal.card:none,instance:kafka-127.0.0.1-9999,jmx_server:127.0.0.1|m:Unable to instantiate or initialize instance 127.0.0.1:9999. Is the target JMX Server or JVM running? Failed to retrieve RMIServer stub: javax.naming.ServiceUnavailableException [Root exception is java.rmi.ConnectException: Connection refused to host: 127.0.0.1; nested exception is: \\n\tjava.net.ConnectException: Connection refused (Connection refused)]";
296        let _ = parse_dsd_service_check(raw_payload.as_bytes()).unwrap();
297    }
298}
saluki_io/deser/codec/dogstatsd/service_check.rs

saluki_io/deser/codec/dogstatsd/
service_check.rs
