feat: batching telemetry event request avoid too many requests

proto/telemetry.proto

@@ -4,11 +4,10 @@
 
 option go_package = "risingwavelabs.com/risingwave/proto/telemetry";
 
 enum MetaBackend {
   META_BACKEND_UNSPECIFIED = 0;
   META_BACKEND_MEMORY = 1;
-  reserved 2;
-  reserved "META_BACKEND_ETCD";
+  META_BACKEND_ETCD = 2;
   META_BACKEND_RDB = 3;
 }
 
@@ -167,3 +166,7 @@
   // mark the event is a test message
   bool is_test = 11;
 }
+
+message BatchEventMessage {
+  repeated EventMessage events = 1;
+}

src/common/src/telemetry/report.rs

@@ -16,11 +16,12 @@ use std::sync::Arc;
 
 use risingwave_telemetry_event::get_telemetry_risingwave_cloud_uuid;
 pub use risingwave_telemetry_event::{
-    current_timestamp, post_telemetry_report_pb, TELEMETRY_REPORT_URL, TELEMETRY_TRACKING_ID,
+    current_timestamp, do_telemetry_event_report, post_telemetry_report_pb,
+    TELEMETRY_EVENT_REPORT_INTERVAL, TELEMETRY_REPORT_URL, TELEMETRY_TRACKING_ID,
 };
 use tokio::sync::oneshot::Sender;
 use tokio::task::JoinHandle;
-use tokio::time::{interval, Duration};
+use tokio::time::{interval as tokio_interval_fn, Duration};
 use uuid::Uuid;
 
 use super::{Result, TELEMETRY_REPORT_INTERVAL};
@@ -60,9 +61,13 @@ where
 
         let begin_time = std::time::Instant::now();
         let session_id = Uuid::new_v4().to_string();
-        let mut interval = interval(Duration::from_secs(TELEMETRY_REPORT_INTERVAL));
+        let mut interval = tokio_interval_fn(Duration::from_secs(TELEMETRY_REPORT_INTERVAL));
         interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
+        let mut event_interval =
+            tokio_interval_fn(Duration::from_secs(TELEMETRY_EVENT_REPORT_INTERVAL));
+        event_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
         // fetch telemetry tracking_id from the meta node only at the beginning
         // There is only one case tracking_id updated at the runtime ---- metastore data has been
         // cleaned. There is no way that metastore has been cleaned but nodes are still running
@@ -94,6 +99,10 @@ where
         loop {
             tokio::select! {
                 _ = interval.tick() => {},
+                _ = event_interval.tick() => {
+                    do_telemetry_event_report().await;
+                    continue;
+                },
                 _ = &mut shutdown_rx => {
                     tracing::info!("Telemetry exit");
                     return;

src/common/telemetry_event/src/lib.rs

@@ -17,13 +17,14 @@
 mod util;
 
 use std::env;
-use std::sync::OnceLock;
+use std::sync::{LazyLock, OnceLock};
 
 use prost::Message;
 use risingwave_pb::telemetry::{
-    EventMessage as PbEventMessage, PbTelemetryDatabaseObject,
+    EventMessage as PbEventMessage, PbBatchEventMessage, PbTelemetryDatabaseObject,
     TelemetryEventStage as PbTelemetryEventStage,
 };
+use tokio::sync::Mutex;
 pub use util::*;
 
 pub type TelemetryResult<T> = core::result::Result<T, TelemetryError>;
@@ -42,6 +43,27 @@ pub fn get_telemetry_risingwave_cloud_uuid() -> Option<String> {
     env::var(TELEMETRY_RISINGWAVE_CLOUD_UUID).ok()
 }
 
+static TELEMETRY_EVENT_REPORT_STASH: LazyLock<Mutex<Vec<PbEventMessage>>> =
+    LazyLock::new(|| Mutex::new(Vec::new()));
+
+pub async fn do_telemetry_event_report() {
+    const TELEMETRY_EVENT_REPORT_TYPE: &str = "event";
+    let url = (TELEMETRY_REPORT_URL.to_owned() + "/" + TELEMETRY_EVENT_REPORT_TYPE).to_owned();
+    let mut batch_message = PbBatchEventMessage { events: Vec::new() };
+
+    let mut stash_guard = TELEMETRY_EVENT_REPORT_STASH.lock().await;
+    for event in stash_guard.drain(..) {
+        batch_message.events.push(event);
+    }
+    drop(stash_guard);
+
+    post_telemetry_report_pb(&url, batch_message.encode_to_vec())
+        .await
+        .unwrap_or_else(|e| tracing::debug!("{}", e));
+}
+
+pub const TELEMETRY_EVENT_REPORT_INTERVAL: u64 = 10; // 10 seconds
+
 pub fn report_event_common(
     event_stage: PbTelemetryEventStage,
     event_name: &str,
@@ -95,15 +117,8 @@ pub fn request_to_telemetry_event(
         node,
         is_test,
     };
-    let report_bytes = event.encode_to_vec();
-
-    tokio::spawn(async move {
-        const TELEMETRY_EVENT_REPORT_TYPE: &str = "event";
-        let url = (TELEMETRY_REPORT_URL.to_owned() + "/" + TELEMETRY_EVENT_REPORT_TYPE).to_owned();
-        post_telemetry_report_pb(&url, report_bytes)
-            .await
-            .unwrap_or_else(|e| tracing::info!("{}", e))
-    });
+
+    TELEMETRY_EVENT_REPORT_STASH.blocking_lock().push(event);
 }
 
 #[cfg(test)]