feat: telemetry (#7384)

Makefile.toml

@@ -22,6 +22,8 @@ env_scripts = [
   '''
 #!@duckscript
 
+set_env ENABLE_TELEMETRY "false"
+
 is_sanitizer_enabled = get_env ENABLE_SANITIZER
 is_all_in_one_enabled = get_env ENABLE_ALL_IN_ONE
 is_hdfs_backend = get_env ENABLE_HDFS
@@ -164,7 +166,7 @@ script = '''
 #!/usr/bin/env bash
 set -e
 
-if [[ -z "$1" ]]; then 
+if [[ -z "$1" ]]; then
   echo "Please pass a parameter to this script, defining which logs you want to follow"
   echo "Available logs are..."
   ls ${PREFIX_LOG}
@@ -176,7 +178,7 @@ if [[ ! -f ${PREFIX_LOG}/$1 ]]; then
   echo "Available logs are..."
   ls ${PREFIX_LOG}
   exit 1
-fi 
+fi
 
 tail -f -n 5 ${PREFIX_LOG}/$1
 '''

ci/scripts/common.env.sh

@@ -3,6 +3,8 @@ export PROTOC_NO_VENDOR=true
 export CARGO_HOME=/risingwave/.cargo
 export RISINGWAVE_CI=true
 export RUST_BACKTRACE=1
+export ENABLE_TELEMETRY=false
+
 if [ -n "${BUILDKITE_COMMIT:-}" ]; then
   export GIT_SHA=$BUILDKITE_COMMIT
 fi

integration_tests/scripts/run_demos.py

@@ -25,8 +25,8 @@ def run_demo(demo: str, format: str):
     demo_dir = os.path.join(project_dir, demo)
     print("Running demo: {}".format(demo))
 
-    subprocess.run(["docker", "compose", "up", "-d"],
-                   cwd=demo_dir, check=True)
+    subprocess.run(["docker", "compose", "up", "-d", "-e",
+    "ENABLE_TELEMETRY=false"], cwd=demo_dir, check=True)
     sleep(40)
 
     sql_files = ['create_source.sql', 'create_mv.sql', 'query.sql']
@@ -50,8 +50,7 @@ def run_iceberg_demo():
     demo_dir = os.path.join(project_dir, demo)
     print("Running demo: iceberg-sink")
 
-    subprocess.run(["docker", "compose", "up", "-d"],
-                   cwd=demo_dir, check=True)
+    subprocess.run(["docker", "compose", "up", "-d", "-e", "ENABLE_TELEMETRY=false"], cwd=demo_dir, check=True)
     sleep(40)
 
     subprocess.run(["docker", "compose", "exec", "spark", "bash", "/spark-script/run-sql-file.sh", "create-table"],

proto/meta.proto

@@ -13,6 +13,17 @@ import "user.proto";
 option java_package = "com.risingwave.proto";
 option optimize_for = SPEED;
 
+message GetTelemetryInfoRequest {}
+
+message TelemetryInfoResponse {
+  optional string tracking_id = 1;
+}
+
+service TelemetryInfoService {
+  // Request telemetry info from meta node
+  rpc GetTelemetryInfo(GetTelemetryInfoRequest) returns (TelemetryInfoResponse);
+}
+
 message HeartbeatRequest {
   message ExtraInfo {
     oneof info {
@@ -344,6 +355,7 @@ message SystemParams {
   optional string data_directory = 7;
   optional string backup_storage_url = 8;
   optional string backup_storage_directory = 9;
+  optional bool telemetry_enabled = 10;
 }
 
 message GetSystemParamsRequest {}

src/common/Cargo.toml

@@ -56,6 +56,7 @@ prometheus = { version = "0.13" }
 prost = "0.11"
 rand = "0.8"
 regex = "1"
+reqwest = { version = "0.11", features = ["json"] }
 risingwave_pb = { path = "../prost" }
 rust_decimal = { version = "1", features = ["db-tokio-postgres"] }
 ryu = "1.0"
@@ -78,6 +79,7 @@ toml = "0.5"
 tonic = { version = "0.2", package = "madsim-tonic" }
 tracing = "0.1"
 url = "2"
+uuid = "1.2.2"
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../workspace-hack" }

src/common/src/config.rs

@@ -222,6 +222,9 @@ pub struct ServerConfig {
     /// >0 = open metrics
     pub metrics_level: u32,
 
+    #[serde(default = "default::server::telemetry_enabled")]
+    pub telemetry_enabled: bool,
+
     #[serde(flatten)]
     pub unrecognized: HashMap<String, Value>,
 }
@@ -498,6 +501,9 @@ pub struct SystemConfig {
     /// Remote directory for storing snapshots.
     #[serde(default = "default::system::backup_storage_directory")]
     pub backup_storage_directory: String,
+
+    #[serde(default = "default::system::telemetry_enabled")]
+    pub telemetry_enabled: bool,
 }
 
 impl Default for SystemConfig {
@@ -518,6 +524,7 @@ impl SystemConfig {
             data_directory: Some(self.data_directory),
             backup_storage_url: Some(self.backup_storage_url),
             backup_storage_directory: Some(self.backup_storage_directory),
+            telemetry_enabled: Some(self.telemetry_enabled),
         }
     }
 }
@@ -588,6 +595,10 @@ mod default {
         pub fn metrics_level() -> u32 {
             0
         }
+
+        pub fn telemetry_enabled() -> bool {
+            true
+        }
     }
 
     pub mod storage {
@@ -776,5 +787,9 @@ mod default {
         pub fn backup_storage_directory() -> String {
             system_param::default::backup_storage_directory()
         }
+
+        pub fn telemetry_enabled() -> bool {
+            system_param::default::telemetry_enabled()
+        }
     }
 }

src/common/src/lib.rs

@@ -54,6 +54,8 @@ pub mod monitor;
 pub mod row;
 pub mod session_config;
 pub mod system_param;
+pub mod telemetry;
+
 #[cfg(test)]
 pub mod test_utils;
 pub mod types;

src/common/src/system_param/mod.rs

@@ -42,6 +42,7 @@ macro_rules! for_all_undeprecated_params {
             { data_directory, String, "hummock_001".to_string() },
             { backup_storage_url, String, "memory".to_string() },
             { backup_storage_directory, String, "backup".to_string() },
+            { telemetry_enabled, bool, true},
             $({ $field, $type, $default },)*
         }
     };
@@ -291,6 +292,10 @@ impl ValidateOnSet for OverrideValidateOnSet {
         // TODO
         Ok(())
     }
+
+    fn telemetry_enabled(_: &bool) -> Result<()> {
+        Ok(())
+    }
 }
 
 for_all_undeprecated_params!(impl_default_from_other_params);
@@ -315,6 +320,7 @@ mod tests {
             (DATA_DIRECTORY_KEY, "a"),
             (BACKUP_STORAGE_URL_KEY, "a"),
             (BACKUP_STORAGE_DIRECTORY_KEY, "a"),
+            (TELEMETRY_ENABLED_KEY, "false"),
         ];
 
         // To kv - missing field.

src/common/src/system_param/reader.rs

@@ -76,6 +76,10 @@ impl SystemParamsReader {
         self.prost.backup_storage_directory.as_ref().unwrap()
     }
 
+    pub fn telemetry_enabled(&self) -> bool {
+        self.prost.telemetry_enabled.unwrap()
+    }
+
     pub fn to_kv(&self) -> Vec<(String, String)> {
         system_params_to_kv(&self.prost).unwrap()
     }