Merge pull request #102 from sgodin/datagram-logging

englishm-cloudflare · web-flow · commit 6ecfcd7cc987 · 2025-11-05T11:13:43.000-05:00
Update logging for Datagrams
diff --git a/moq-transport/src/mlog/events.rs b/moq-transport/src/mlog/events.rs
@@ -11,7 +11,6 @@
 //
 // TODO: Unimplemented data plane events (from draft-pardue-moq-qlog-moq-events):
 // - stream_type_set (when stream type becomes known)
-// - object_datagram_created/parsed
 // - object_datagram_status_created/parsed
 // - fetch_header_created/parsed
 // - fetch_object_created/parsed
@@ -61,6 +60,12 @@ pub enum EventData {
     #[serde(rename = "subgroup_object_created")]
     SubgroupObjectCreated(SubgroupObjectCreated),
 
+    #[serde(rename = "object_datagram_parsed")]
+    ObjectDatagramParsed(ObjectDatagramParsed),
+
+    #[serde(rename = "object_datagram_created")]
+    ObjectDatagramCreated(ObjectDatagramCreated),
+
     #[serde(rename = "loglevel")]
     LogLevel(LogLevelEvent),
 }
@@ -133,6 +138,28 @@ pub struct SubgroupObjectCreated {
     pub object: JsonValue,
 }
 
+/// Object Datagram parsed event (data plane)
+#[serde_with::skip_serializing_none]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ObjectDatagramParsed {
+    pub stream_id: u64,
+
+    /// Object-specific fields
+    #[serde(flatten)]
+    pub object: JsonValue,
+}
+
+/// Object Datagram created event (data plane)
+#[serde_with::skip_serializing_none]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ObjectDatagramCreated {
+    pub stream_id: u64,
+
+    /// Object-specific fields
+    #[serde(flatten)]
+    pub object: JsonValue,
+}
+
 /// LogLevel event for flexible logging (qlog loglevel schema)
 /// See: https://www.ietf.org/archive/id/draft-ietf-quic-qlog-main-schema-12.html#name-loglevel-events
 #[serde_with::skip_serializing_none]
@@ -492,10 +519,10 @@ pub fn go_away_created(time: f64, stream_id: u64, msg: &message::GoAway) -> Even
 /// Helper to convert SubgroupHeader to JSON
 fn subgroup_header_to_json(header: &data::SubgroupHeader) -> JsonValue {
     let mut json = json!({
+        "header_type": format!("{:?}", header.header_type),
         "track_alias": header.track_alias,
         "group_id": header.group_id,
         "publisher_priority": header.publisher_priority,
-        "header_type": format!("{:?}", header.header_type),
     });
 
     if let Some(subgroup_id) = header.subgroup_id {
@@ -540,7 +567,7 @@ fn subgroup_object_to_json(
         "group_id": group_id,
         "subgroup_id": subgroup_id,
         "object_id": object_id,
-        "extension_headers_length": 0,
+        // TODO send object_playload itself
         "object_payload_length": object.payload_length,
     });
 
@@ -596,12 +623,12 @@ fn subgroup_object_ext_to_json(
     object_id: u64,
     object: &data::SubgroupObjectExt,
 ) -> JsonValue {
-    // TODO encode extension headers
     let mut object_data = json!({
         "group_id": group_id,
         "subgroup_id": subgroup_id,
         "object_id": object_id,
-        "extension_headers_length": object.extension_headers.0.len(),
+        "extension_headers": key_value_pairs_to_vec(&object.extension_headers),
+        // TODO send object_playload itself
         "object_payload_length": object.payload_length,
     });
 
@@ -650,6 +677,53 @@ pub fn subgroup_object_ext_created(
     }
 }
 
+/// Helper to convert Datagram to JSON
+fn object_datagram_to_json(datagram: &data::Datagram) -> JsonValue {
+    let mut json = json!({
+        "datagram_type": format!("{:?}", datagram.datagram_type),
+        "track_alias": datagram.track_alias,
+        "group_id": datagram.group_id,
+        "object_id": datagram.object_id.unwrap_or(0),
+        "publisher_priority": datagram.publisher_priority,
+        // TODO send object_playload
+        "payload_length": datagram.payload.as_ref().map_or(0, |p| p.len()),
+    });
+
+    if let Some(extension_headers) = &datagram.extension_headers {
+        json["extension_headers"] = json!(key_value_pairs_to_vec(extension_headers));
+    }
+
+    if let Some(status) = datagram.status {
+        json["object_status"] = json!(format!("{:?}", status));
+    }
+
+    json
+}
+
+/// Create a object_datagram_parsed event
+pub fn object_datagram_parsed(time: f64, stream_id: u64, datagram: &data::Datagram) -> Event {
+    Event {
+        time,
+        name: "moqt:object_datagram_parsed".to_string(),
+        data: EventData::ObjectDatagramParsed(ObjectDatagramParsed {
+            stream_id,
+            object: object_datagram_to_json(datagram),
+        }),
+    }
+}
+
+/// Create a object_datagram_created event
+pub fn object_datagram_created(time: f64, stream_id: u64, datagram: &data::Datagram) -> Event {
+    Event {
+        time,
+        name: "moqt:object_datagram_created".to_string(),
+        data: EventData::ObjectDatagramCreated(ObjectDatagramCreated {
+            stream_id,
+            object: object_datagram_to_json(datagram),
+        }),
+    }
+}
+
 // LogLevel events (generic logging)
 
 /// Log levels for qlog loglevel events
diff --git a/moq-transport/src/mlog/mod.rs b/moq-transport/src/mlog/mod.rs
@@ -8,7 +8,8 @@ pub use writer::MlogWriter;
 
 pub mod events;
 pub use events::{
-    client_setup_parsed, loglevel_event, server_setup_created, subgroup_header_created,
-    subgroup_header_parsed, subgroup_object_created, subgroup_object_ext_created,
-    subgroup_object_ext_parsed, subgroup_object_parsed, Event, EventData, LogLevel,
+    client_setup_parsed, loglevel_event, object_datagram_created, object_datagram_parsed,
+    server_setup_created, subgroup_header_created, subgroup_header_parsed, subgroup_object_created,
+    subgroup_object_ext_created, subgroup_object_ext_parsed, subgroup_object_parsed, Event,
+    EventData, LogLevel,
 };
diff --git a/moq-transport/src/session/subscribed.rs b/moq-transport/src/session/subscribed.rs
@@ -382,15 +382,29 @@ impl Subscribed {
             encoded_datagram.encode(&mut buffer)?;
 
             log::debug!(
-                "[PUBLISHER] serve_datagrams: sending datagram #{} - group_id={}, object_id={}, priority={}, payload_len={}, total_encoded_len={}",
+                "[PUBLISHER] serve_datagrams: sending datagram #{} - track_alias={}, group_id={}, object_id={}, priority={}, payload_len={}, total_encoded_len={}",
                 datagram_count + 1,
+                encoded_datagram.track_alias,
                 encoded_datagram.group_id,
                 encoded_datagram.object_id.unwrap(),
                 encoded_datagram.publisher_priority,
                 payload_len,
                 buffer.len()
             );
 
+            // Create mlog event for datagram created
+            if let Some(ref mlog) = self.mlog {
+                if let Ok(mut mlog_guard) = mlog.lock() {
+                    let time = mlog_guard.elapsed_ms();
+                    let stream_id = 0; // TODO: Placeholder, need actual QUIC stream ID
+                    let _ = mlog_guard.add_event(mlog::object_datagram_created(
+                        time,
+                        stream_id,
+                        &encoded_datagram,
+                    ));
+                }
+            }
+
             self.publisher.send_datagram(buffer.into()).await?;
 
             self.state
diff --git a/moq-transport/src/session/subscriber.rs b/moq-transport/src/session/subscriber.rs
@@ -550,14 +550,39 @@ impl Subscriber {
         let mut cursor = io::Cursor::new(datagram);
         let datagram = data::Datagram::decode(&mut cursor)?;
 
+        if let Some(ref mlog) = self.mlog {
+            if let Ok(mut mlog_guard) = mlog.lock() {
+                let time = mlog_guard.elapsed_ms();
+                let stream_id = 0; // TODO: Placeholder, need actual QUIC stream ID
+                let _ =
+                    mlog_guard.add_event(mlog::object_datagram_parsed(time, stream_id, &datagram));
+            }
+        }
+
         // Look up the subscribe id for this track alias
         if let Some(subscribe_id) = self.get_subscribe_id_by_alias(datagram.track_alias) {
             // Look up the subscribe by id
             if let Some(subscribe) = self.subscribes.lock().unwrap().get_mut(&subscribe_id) {
+                log::trace!(
+                    "[SUBSCRIBER] recv_datagram: track_alias={}, group_id={}, object_id={}, publisher_priority={}, status={}, payload_length={}",
+                    datagram.track_alias,
+                    datagram.group_id,
+                    datagram.object_id.unwrap_or(0),
+                    datagram.publisher_priority,
+                    datagram.status.as_ref().map_or("None".to_string(), |s| format!("{:?}", s)),
+                    datagram.payload.as_ref().map_or(0, |p| p.len()));
                 subscribe.datagram(datagram)?;
             }
+        } else {
+            log::warn!(
+                "[SUBSCRIBER] recv_datagram: discarded due to unknown track_alias: track_alias={}, group_id={}, object_id={}, publisher_priority={}, status={}, payload_length={}",
+                datagram.track_alias,
+                datagram.group_id,
+                datagram.object_id.unwrap_or(0),
+                datagram.publisher_priority,
+                datagram.status.as_ref().map_or("None".to_string(), |s| format!("{:?}", s)),
+                datagram.payload.as_ref().map_or(0, |p| p.len()));
         }
-        // TODO do we want to return an error if we can't find the subscribe?
 
         Ok(())
     }
