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 container_id: 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_container_id = 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                // Container ID: client-provided container ID for the container that this service check originated from.
68                CONTAINER_ID_PREFIX => {
69                    let (_, container_id) =
70                        all_consuming(preceded(tag(CONTAINER_ID_PREFIX), container_id)).parse(chunk)?;
71                    maybe_container_id = Some(container_id);
72                }
73                // External Data: client-provided data used for resolving the entity ID that this service check originated from.
74                EXTERNAL_DATA_PREFIX => {
75                    let (_, external_data) =
76                        all_consuming(preceded(tag(EXTERNAL_DATA_PREFIX), external_data)).parse(chunk)?;
77                    maybe_external_data = Some(external_data);
78                }
79                // Tags: additional tags to be added to the service check.
80                _ if chunk.starts_with(TAGS_PREFIX) => {
81                    let (_, tags) = all_consuming(preceded(tag(TAGS_PREFIX), tags(config))).parse(chunk)?;
82                    maybe_tags = Some(tags);
83                }
84                // Message: A message describing the current state of the service check.
85                SERVICE_CHECK_MESSAGE_PREFIX => {
86                    let (_, message) = all_consuming(preceded(tag(SERVICE_CHECK_MESSAGE_PREFIX), utf8)).parse(chunk)?;
87                    maybe_message = Some(message);
88                }
89                // Cardinality: client-provided cardinality for the service check.
90                _ if chunk.starts_with(CARDINALITY_PREFIX) => {
91                    let (_, cardinality) = cardinality(chunk)?;
92                    maybe_cardinality = cardinality;
93                }
94                _ => {
95                    // We don't know what this is, so we just skip it.
96                    //
97                    // TODO: Should we throw an error, warn, or be silently permissive?
98                }
99            }
100            remaining = tail;
101        }
102        remaining
103    } else {
104        remaining
105    };
106
107    let tags = maybe_tags.unwrap_or_else(RawTags::empty);
108
109    let service_check_packet = ServiceCheckPacket {
110        name: name.into(),
111        status: check_status,
112        tags,
113        timestamp: maybe_timestamp,
114        hostname: maybe_hostname,
115        message: maybe_message,
116        container_id: maybe_container_id,
117        external_data: maybe_external_data,
118        cardinality: maybe_cardinality,
119    };
120    Ok((remaining, service_check_packet))
121}
122
123#[cfg(test)]
124mod tests {
125    use nom::IResult;
126    use saluki_context::{
127        origin::OriginTagCardinality,
128        tags::{SharedTagSet, TagSet},
129    };
130    use saluki_core::data_model::event::service_check::{CheckStatus, ServiceCheck};
131    use stringtheory::MetaString;
132
133    use super::{parse_dogstatsd_service_check, DogstatsdCodecConfiguration};
134
135    type NomResult<'input, T> = Result<T, nom::Err<nom::error::Error<&'input [u8]>>>;
136
137    fn parse_dsd_service_check(input: &[u8]) -> NomResult<'_, ServiceCheck> {
138        let default_config = DogstatsdCodecConfiguration::default();
139        parse_dsd_service_check_with_conf(input, &default_config)
140    }
141
142    fn parse_dsd_service_check_with_conf<'input>(
143        input: &'input [u8], config: &DogstatsdCodecConfiguration,
144    ) -> NomResult<'input, ServiceCheck> {
145        let (remaining, service_check) = parse_dsd_service_check_direct(input, config)?;
146        assert!(remaining.is_empty());
147
148        Ok(service_check)
149    }
150
151    fn parse_dsd_service_check_direct<'input>(
152        input: &'input [u8], config: &DogstatsdCodecConfiguration,
153    ) -> IResult<&'input [u8], ServiceCheck> {
154        let (remaining, packet) = parse_dogstatsd_service_check(input, config)?;
155        assert!(remaining.is_empty());
156
157        let mut service_check_tags = TagSet::default();
158        for tag in packet.tags.into_iter() {
159            service_check_tags.insert_tag(tag);
160        }
161
162        let service_check = ServiceCheck::new(packet.name, packet.status)
163            .with_timestamp(packet.timestamp)
164            .with_hostname(packet.hostname.map(|s| s.into()))
165            .with_tags(service_check_tags)
166            .with_message(packet.message.map(|s| s.into()));
167
168        Ok((remaining, service_check))
169    }
170
171    #[track_caller]
172    fn check_basic_service_check_eq(expected: ServiceCheck, actual: ServiceCheck) {
173        assert_eq!(expected.name(), actual.name());
174        assert_eq!(expected.status(), actual.status());
175        assert_eq!(expected.timestamp(), actual.timestamp());
176        assert_eq!(expected.hostname(), actual.hostname());
177        assert_eq!(expected.tags(), actual.tags());
178        assert_eq!(expected.message(), actual.message());
179        assert_eq!(expected.origin_tags(), actual.origin_tags());
180    }
181
182    #[test]
183    fn basic_service_checks() {
184        let name = "testsvc";
185        let sc_status = CheckStatus::Warning;
186        let raw = format!("_sc|{}|{}", name, sc_status.as_u8());
187        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
188        let expected = ServiceCheck::new(name, sc_status);
189        check_basic_service_check_eq(expected, actual);
190    }
191
192    #[test]
193    fn service_check_timestamp() {
194        let name = "testsvc";
195        let sc_status = CheckStatus::Warning;
196        let sc_timestamp = 1234567890;
197        let raw = format!("_sc|{}|{}|d:{}", name, sc_status.as_u8(), sc_timestamp);
198        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
199        let expected = ServiceCheck::new(name, sc_status).with_timestamp(sc_timestamp);
200        check_basic_service_check_eq(expected, actual);
201    }
202
203    #[test]
204    fn service_check_tags() {
205        let name = "testsvc";
206        let sc_status = CheckStatus::Warning;
207        let tags = ["tag1", "tag2"];
208        let raw = format!("_sc|{}|{}|#{}", name, sc_status.as_u8(), tags.join(","));
209        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
210        let shared_tag_set = SharedTagSet::from(TagSet::from_iter(tags.iter().map(|&s| s.into())));
211        let expected = ServiceCheck::new(name, sc_status).with_tags(shared_tag_set);
212        check_basic_service_check_eq(expected, actual);
213    }
214
215    #[test]
216    fn service_check_message() {
217        let name = "testsvc";
218        let sc_status = CheckStatus::Ok;
219        let sc_message = MetaString::from("service running properly");
220        let raw = format!("_sc|{}|{}|m:{}", name, sc_status.as_u8(), sc_message);
221        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
222        let expected = ServiceCheck::new(name, sc_status).with_message(sc_message);
223        check_basic_service_check_eq(expected, actual);
224    }
225
226    #[test]
227    fn service_check_fields_after_message() {
228        let name = "testsvc";
229        let sc_status = CheckStatus::Ok;
230        let sc_message = MetaString::from("service running properly");
231        let sc_container_id = "ci-1234567890";
232        let sc_external_data = "it-false,cn-redis,pu-810fe89d-da47-410b-8979-9154a40f8183";
233        let raw = format!(
234            "_sc|{}|{}|#tag1,tag2|m:{}|c:{}|e:{}",
235            name,
236            sc_status.as_u8(),
237            sc_message,
238            sc_container_id,
239            sc_external_data,
240        );
241        let result = parse_dsd_service_check(raw.as_bytes()).unwrap();
242        let tags = ["tag1", "tag2"];
243        let shared_tag_set = SharedTagSet::from(TagSet::from_iter(tags.iter().map(|&s| s.into())));
244        let expected = ServiceCheck::new(name, sc_status)
245            .with_message(sc_message)
246            .with_tags(shared_tag_set);
247        check_basic_service_check_eq(expected, result);
248
249        let config = DogstatsdCodecConfiguration::default();
250        let (_, packet) = parse_dogstatsd_service_check(raw.as_bytes(), &config).expect("should not fail to parse");
251        assert_eq!(packet.container_id, Some(sc_container_id));
252        assert_eq!(packet.external_data, Some(sc_external_data));
253    }
254
255    #[test]
256    fn service_check_multiple_extensions() {
257        let name = "testsvc";
258        let sc_status = CheckStatus::Unknown;
259        let sc_timestamp = 1234567890;
260        let sc_hostname = MetaString::from("myhost");
261        let sc_container_id = "abcdef123456";
262        let sc_external_data = "it-false,cn-redis,pu-810fe89d-da47-410b-8979-9154a40f8183";
263        let tags = ["tag1", "tag2"];
264        let sc_message = MetaString::from("service status unknown");
265        let sc_cardinality = "none";
266        let raw = format!(
267            "_sc|{}|{}|d:{}|h:{}|c:{}|e:{}|card:{}|#{}|m:{}",
268            name,
269            sc_status.as_u8(),
270            sc_timestamp,
271            sc_hostname,
272            sc_container_id,
273            sc_external_data,
274            sc_cardinality,
275            tags.join(","),
276            sc_message
277        );
278        let shared_tag_set = SharedTagSet::from(TagSet::from_iter(tags.iter().map(|&s| s.into())));
279        let actual = parse_dsd_service_check(raw.as_bytes()).unwrap();
280        let expected = ServiceCheck::new(name, sc_status)
281            .with_timestamp(sc_timestamp)
282            .with_hostname(sc_hostname)
283            .with_tags(shared_tag_set)
284            .with_message(sc_message);
285        check_basic_service_check_eq(expected, actual);
286
287        let config = DogstatsdCodecConfiguration::default();
288        let (_, packet) = parse_dogstatsd_service_check(raw.as_bytes(), &config).expect("should not fail to parse");
289        assert_eq!(packet.container_id, Some(sc_container_id));
290        assert_eq!(packet.external_data, Some(sc_external_data));
291        assert_eq!(packet.cardinality, Some(OriginTagCardinality::None));
292    }
293
294    #[test]
295    fn service_check_semi_real_payload_kafka() {
296        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)]";
297        let _ = parse_dsd_service_check(raw_payload.as_bytes()).unwrap();
298    }
299}
saluki_io/deser/codec/dogstatsd/service_check.rs

saluki_io/deser/codec/dogstatsd/
service_check.rs
