updated timestamp to nanoseconds instead of millis

Author: ruokun-niu

sources/shared/change-dispatcher/src/main.rs

@@ -151,22 +151,24 @@ async fn process_changes(
             );
 
             let mut dispatch_event = change_event.clone();
-            dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherStart_ms"] =
-                match serde_json::to_value(chrono::Utc::now().timestamp_millis()) {
+
+            // Start time, measured in nanoseconds
+            dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherStart_ns"] =
+                match serde_json::to_value(chrono::Utc::now().timestamp_nanos()) {
                     Ok(val) => val,
                     Err(_) => {
                         return Err(Box::<dyn std::error::Error>::from(
                             "Error serializing timestamp into json value",
                         ));
                     }
                 };
-            dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherEnd_ms"] =
-                match serde_json::to_value(0) {
-                    Ok(val) => val,
-                    Err(_) => {
-                        unreachable!();
-                    }
-                };
+            // dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherEnd_ms"] =
+            //     match serde_json::to_value(0) {
+            //         Ok(val) => val,
+            //         Err(_) => {
+            //             unreachable!();
+            //         }
+            //     };
 
             let subscriptions = match change_event["subscriptions"].as_array() {
                 Some(subs) => subs.clone(),
@@ -185,15 +187,7 @@ async fn process_changes(
             for query_node_id in query_nodes {
                 let app_id = format!("{}-publish-api", query_node_id);
 
-                dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherEnd_ms"] =
-                    match serde_json::to_value(chrono::Utc::now().timestamp_millis()) {
-                        Ok(val) => val,
-                        Err(_) => {
-                            return Err(Box::<dyn std::error::Error>::from(
-                                "Error serializing timestamp into json value",
-                            ));
-                        }
-                    };
+                
                 let queries: Vec<_> = subscriptions
                     .iter()
                     .filter(|x| x["queryNodeId"] == query_node_id)
@@ -213,6 +207,17 @@ async fn process_changes(
                     headers.insert("traceparent".to_string(), traceparent.clone());
                 }
                 let headers = Headers::new(headers);
+
+                // End time, measured in nanoseconds
+                dispatch_event["metadata"]["tracking"]["source"]["changeDispatcherEnd_ns"] =
+                    match serde_json::to_value(chrono::Utc::now().timestamp_nanos()) {
+                        Ok(val) => val,
+                        Err(_) => {
+                            return Err(Box::<dyn std::error::Error>::from(
+                                "Error serializing timestamp into json value",
+                            ));
+                        }
+                    };
                 match invoker
                     .invoke(
                         Payload::Json(dispatch_event.clone()),

sources/shared/change-router/src/main.rs

@@ -266,8 +266,11 @@ async fn process_changes(
     }
     if let Some(changes) = changes.as_array() {
         for change in changes {
-            let change_router_start = chrono::Utc::now().timestamp_millis();
             let change_id = Uuid::new_v4().to_string();
+            info!("Begin processing change with id: {}", change_id);
+            // Capture the start time of this change
+            let change_router_start = chrono::Utc::now().timestamp_nanos();
+            
 
             info!(
                 "Processing change - db:{}, type:{}, id:{}",
@@ -523,8 +526,8 @@ async fn process_changes(
                                 "source": {
                                     "seq": change["payload"]["source"]["lsn"],
                                     "reactivator_ms": change["ts_ms"],
-                                    "changeSvcStart_ms": change_router_start,
-                                    "changeSvcEnd_ms": chrono::Utc::now().timestamp_millis()
+                                    "changeSvcStart_ns": change_router_start,
+                                    "changeSvcEnd_ns": chrono::Utc::now().timestamp_nanos(),
                                 }
                             }
                         }