feat(sidecar): track Data Streams Monitoring backlogs in shared memory

datadog-sidecar/src/service/blocking.rs

@@ -552,6 +552,26 @@ pub fn add_dsm_checkpoint(
     Ok(transport.with_retry(|s| s.send(&request)))
 }
 
+/// IPC fallback for DSM backlogs. Returns whether the offset was sent, as
+/// [`add_dsm_checkpoint`] does.
+pub fn add_dsm_backlog(
+    transport: &mut SidecarTransport,
+    env: String,
+    version: String,
+    service: String,
+    primary_tag: String,
+    backlog: libdd_ipc::shm_dsm_backlog::OwnedDsmBacklog,
+) -> io::Result<bool> {
+    let request = SidecarInterfaceRequest::AddDsmBacklog {
+        env,
+        version,
+        service,
+        primary_tag,
+        backlog,
+    };
+    Ok(transport.with_retry(|s| s.send(&request)))
+}
+
 /// Starts the AppSec backend in the sidecar and waits for initialization to
 /// complete before returning.
 pub fn ensure_appsec_started(
@@ -624,8 +644,8 @@ mod tests {
     use super::try_submit_ffe;
     use crate::service::SidecarAction;
     use crate::service::blocking::{
-        SidecarTransport, add_dsm_checkpoint, add_span_to_concentrator, enqueue_actions,
-        send_dogstatsd_actions,
+        SidecarTransport, add_dsm_backlog, add_dsm_checkpoint, add_span_to_concentrator,
+        enqueue_actions, send_dogstatsd_actions,
     };
     use crate::service::ffe_submission::FfeSubmissionStatus;
     use crate::service::sidecar_interface::{
@@ -946,6 +966,60 @@ mod tests {
         assert!(!send(&mut transport));
     }
 
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn dsm_backlog_fallback_reports_whether_the_offset_was_sent() {
+        use libdd_ipc::shm_dsm_backlog::BacklogTag;
+        let (conn, peer) = SeqpacketConn::socketpair().unwrap();
+        let mut transport = SidecarTransport::from(conn);
+        transport
+            .set_read_timeout(Some(Duration::from_millis(100)))
+            .unwrap();
+        let send = |transport: &mut SidecarTransport| {
+            add_dsm_backlog(
+                transport,
+                "env".into(),
+                "v1".into(),
+                "svc".into(),
+                "".into(),
+                libdd_ipc::shm_dsm_backlog::OwnedDsmBacklog {
+                    tags: vec![
+                        (BacklogTag::Topic, "orders".into()),
+                        (BacklogTag::Type, "kafka_produce".into()),
+                    ],
+                    offset: 42,
+                },
+            )
+            .unwrap()
+        };
+        assert!(send(&mut transport));
+        let mut buf = [0; 1024];
+        let (len, _) = peer.try_recv_raw(&mut buf).unwrap();
+        match libdd_ipc::codec::decode::<SidecarInterfaceRequest>(&buf[..len]).unwrap() {
+            SidecarInterfaceRequest::AddDsmBacklog {
+                env,
+                service,
+                backlog,
+                ..
+            } => {
+                assert_eq!(env, "env");
+                assert_eq!(service, "svc");
+                assert_eq!(
+                    backlog.tags,
+                    [
+                        (BacklogTag::Topic, "orders".to_owned()),
+                        (BacklogTag::Type, "kafka_produce".to_owned()),
+                    ]
+                );
+                assert_eq!(backlog.offset, 42);
+            }
+            _ => panic!("expected a DSM backlog"),
+        }
+
+        drop(peer);
+        assert!(!send(&mut transport));
+    }
+
     /// Decodes all requests sent to `peer` so far.
     fn received(peer: &SeqpacketConn) -> Vec<SidecarInterfaceRequest> {
         let mut buf = vec![0; libdd_ipc::max_message_size()];