telemetry more

Author: BohuTANG

src/common/telemetry/src/simple.rs

@@ -28,7 +28,8 @@ pub async fn report_node_telemetry(payload: serde_json::Value) {
     };
 
     let version = payload
-        .get("version")
+        .get("node")
+        .and_then(|node| node.get("version"))
         .and_then(|v| v.as_str())
         .unwrap_or("unknown");
 

src/query/config/src/config.rs

@@ -74,7 +74,11 @@ use serfig::collectors::from_self;
 #[serde(default)]
 pub struct TelemetryConfig {
     /// Enable/disable telemetry reporting (only works with EE license)
-    #[clap(long = "telemetry-enabled", value_name = "BOOL")]
+    #[clap(
+        long = "telemetry-enabled",
+        value_name = "BOOL",
+        default_value = "true"
+    )]
     #[serde(default = "default_telemetry_enabled")]
     pub enabled: bool,
 }

src/query/service/src/clusters/cluster.rs

@@ -530,7 +530,13 @@ impl ClusterDiscovery {
     }
 
     #[async_backtrace::framed]
-    pub async fn unregister_to_metastore(self: &Arc<Self>, signal: &mut SignalStream) {
+    pub async fn unregister_to_metastore(
+        self: &Arc<Self>,
+        signal: &mut SignalStream,
+        cfg: &InnerConfig,
+    ) {
+        self.report_telemetry_data(cfg, None).await;
+
         let mut heartbeat = self.heartbeat.lock().await;
 
         if let Err(shutdown_failure) = heartbeat.shutdown().await {
@@ -646,8 +652,7 @@ impl ClusterDiscovery {
         match self.warehouse_manager.start_node(node_info.clone()).await {
             Ok(seq_node) => {
                 self.start_heartbeat(seq_node).await?;
-                self.report_telemetry_data(&node_info, cfg, online_nodes)
-                    .await;
+                self.report_telemetry_data(cfg, Some(online_nodes)).await;
                 Ok(())
             }
             Err(cause) => Err(cause.add_message_back("(while cluster api add_node).")),
@@ -688,43 +693,138 @@ impl ClusterDiscovery {
         Ok(settings.get_enterprise_license())
     }
 
-    async fn report_telemetry_data(
-        &self,
-        _node_info: &NodeInfo,
-        cfg: &InnerConfig,
-        online_nodes: Vec<NodeInfo>,
-    ) {
-        if let Ok(key) = Self::get_license_key(&self.tenant_id).await {
-            if !key.is_empty() {
-                if let Ok(claims) = LicenseManagerSwitch::instance().parse_license(&key) {
-                    let license_type = claims.custom.r#type.as_deref().unwrap_or("").to_lowercase();
-                    if license_type != "trial" && !cfg.telemetry.enabled {
-                        return;
-                    }
-                }
+    async fn report_telemetry_data(&self, cfg: &InnerConfig, online_nodes: Option<Vec<NodeInfo>>) {
+        let start_time = std::time::Instant::now();
+
+        let license_result = Self::get_license_key(&self.tenant_id)
+            .await
+            .ok()
+            .filter(|key| !key.is_empty())
+            .and_then(|key| LicenseManagerSwitch::instance().parse_license(&key).ok());
+
+        if let Some(ref claims) = license_result {
+            let license_type = claims.custom.r#type.as_deref().unwrap_or("").to_lowercase();
+            if license_type != "trial" && !cfg.telemetry.enabled {
+                return;
             }
         }
 
+        let license_info = match license_result {
+            Some(claims) => {
+                let expires_at = claims.expires_at.map(|d| d.as_secs()).unwrap_or(0);
+                serde_json::json!({
+                    "has_license": true,
+                    "expires_at": expires_at,
+                    "license_info": claims.custom
+                })
+            }
+            None => serde_json::json!({
+                "has_license": false,
+                "license_info": null
+            }),
+        };
+
+        // Collect system information
+        let mut system = sysinfo::System::new();
+        system.refresh_memory();
+        system.refresh_cpu_all();
+
+        let event_type = if online_nodes.is_none() {
+            "node_shutdown"
+        } else {
+            "node_startup"
+        };
+
         let payload = serde_json::json!({
             "timestamp": std::time::SystemTime::now()
                 .duration_since(std::time::UNIX_EPOCH)
                 .map(|d| d.as_secs())
                 .unwrap_or(0),
-            "cluster_id": self.cluster_id,
-            "node_id": self.local_id,
-            "tenant_name": self.tenant_id,
-            "version": DATABEND_COMMIT_VERSION.as_str(),
-            "cpu_cores": num_cpus::get(),
-            "storage_info": cfg.storage.params.to_string(),
-            "nodes": online_nodes.iter().map(|node| serde_json::json!({
-                "id": node.id,
-                "secret": node.secret,
-                "cpu_nums": node.cpu_nums,
-                "version": node.version,
-                "binary_version": node.binary_version,
-                "cluster_id": node.cluster_id,
-                "warehouse_id": node.warehouse_id,
-            })).collect::<Vec<_>>(),
+            "event_type": event_type,
+            "cluster": {
+                "cluster_id": self.cluster_id,
+                "tenant_name": self.tenant_id
+            },
+            "node": {
+                "node_id": self.local_id,
+                "version": DATABEND_COMMIT_VERSION.as_str()
+            },
+            "system": {
+                "os_name": sysinfo::System::name().unwrap_or_else(|| "unknown".to_string()),
+                "os_version": sysinfo::System::os_version().unwrap_or_else(|| "unknown".to_string()),
+                "kernel_version": sysinfo::System::kernel_version().unwrap_or_else(|| "unknown".to_string()),
+                "arch": sysinfo::System::cpu_arch(),
+                "cpu_cores": system.cpus().len(),
+                "total_memory": system.total_memory(),
+                "available_memory": system.available_memory(),
+                "used_memory": system.used_memory(),
+                "uptime_seconds": sysinfo::System::uptime()
+            },
+            "cache_strategy": {
+                "enable_table_meta_cache": cfg.cache.enable_table_meta_cache,
+                "table_meta_snapshot_count": cfg.cache.table_meta_snapshot_count,
+                "table_meta_segment_bytes": cfg.cache.table_meta_segment_bytes,
+                "block_meta_count": cfg.cache.block_meta_count,
+                "segment_block_metas_count": cfg.cache.segment_block_metas_count,
+                "table_meta_statistic_count": cfg.cache.table_meta_statistic_count,
+                "segment_statistics_count": cfg.cache.segment_statistics_count,
+                "enable_table_index_bloom": cfg.cache.enable_table_index_bloom,
+                "table_bloom_index_meta_count": cfg.cache.table_bloom_index_meta_count,
+                "table_bloom_index_filter_count": cfg.cache.table_bloom_index_filter_count,
+                "table_bloom_index_filter_size": cfg.cache.table_bloom_index_filter_size,
+                "table_prune_partitions_count": cfg.cache.table_prune_partitions_count,
+                "inverted_index_meta_count": cfg.cache.inverted_index_meta_count,
+                "inverted_index_filter_size": cfg.cache.inverted_index_filter_size,
+                "vector_index_meta_count": cfg.cache.vector_index_meta_count,
+                "vector_index_filter_size": cfg.cache.vector_index_filter_size,
+                "data_cache_in_memory_bytes": cfg.cache.data_cache_in_memory_bytes,
+                "table_data_cache_population_queue_size": cfg.cache.table_data_cache_population_queue_size,
+                "table_data_deserialized_data_bytes": cfg.cache.table_data_deserialized_data_bytes,
+                "disk_cache_max_bytes": cfg.cache.disk_cache_config.max_bytes,
+                "disk_cache_sync_data": cfg.cache.disk_cache_config.sync_data
+            },
+            "memory_management": {
+                "max_server_memory_usage": cfg.query.max_server_memory_usage,
+                "max_memory_limit_enabled": cfg.query.max_memory_limit_enabled,
+                "spill_enabled": cfg.spill.global_bytes_limit > 0,
+                "reserved_disk_ratio": cfg.spill.reserved_disk_ratio
+            },
+            "query_execution": {
+                "max_running_queries": cfg.query.max_running_queries,
+                "global_statement_queue": cfg.query.global_statement_queue,
+            },
+            "storage_engine": {
+                "table_engine_memory_enabled": cfg.query.table_engine_memory_enabled,
+                "storage_type": cfg.storage.params.to_string(),
+                "storage_allow_insecure": cfg.storage.allow_insecure
+            },
+            "meta_config": {
+                "meta_embedded": cfg.meta.embedded_dir.is_empty(),
+                "meta_client_timeout": cfg.meta.client_timeout_in_second,
+                "rpc_client_timeout_secs": cfg.query.rpc_client_timeout_secs,
+            },
+            "license": license_info,
+            "cluster_nodes": {
+                "count": online_nodes.as_ref().map(|nodes| nodes.len()).unwrap_or(0),
+                "nodes": online_nodes.as_ref().map(|nodes|
+                    nodes.iter().map(|node| serde_json::json!({
+                        "id": node.id,
+                        "cpu_nums": node.cpu_nums,
+                        "version": node.version,
+                        "binary_version": node.binary_version,
+                        "cluster_id": node.cluster_id,
+                        "warehouse_id": node.warehouse_id
+                    })).collect::<Vec<_>>()
+                ).unwrap_or_default()
+            },
+            "observability": {
+                "collection_duration_ms": start_time.elapsed().as_millis() as u64,
+                "report_timestamp": std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .map(|d| d.as_secs())
+                    .unwrap_or(0),
+                "schema_version": "1.0"
+            }
         });
 
         let _ = report_node_telemetry(payload).await;

src/query/service/src/servers/server.rs

@@ -23,6 +23,7 @@ use databend_common_base::base::DummySignalStream;
 use databend_common_base::base::SignalStream;
 use databend_common_base::base::SignalType;
 use databend_common_base::runtime::drop_guard;
+use databend_common_config::GlobalConfig;
 use databend_common_exception::Result;
 use futures::stream::Abortable;
 use futures::StreamExt;
@@ -71,8 +72,9 @@ impl ShutdownHandle {
     #[async_backtrace::framed]
     pub async fn shutdown(&mut self, mut signal: SignalStream, timeout: Option<Duration>) {
         self.shutdown_services(true).await;
+        let config = GlobalConfig::instance();
         ClusterDiscovery::instance()
-            .unregister_to_metastore(&mut signal)
+            .unregister_to_metastore(&mut signal, &config)
             .await;
         self.sessions.graceful_shutdown(signal, timeout).await;
         self.shutdown_services(false).await;

tests/metactl/metactl_utils.py

@@ -85,9 +85,9 @@ def verify_kv(grpc_addr, key, expected_value=None):
         print(f"Actual value: '{actual_value}', Expected: '{expected_value}'")
 
         if expected_value is not None:
-            assert actual_value == expected_value, (
-                f"Expected '{expected_value}', got '{actual_value}'"
-            )
+            assert (
+                actual_value == expected_value
+            ), f"Expected '{expected_value}', got '{actual_value}'"
 
 
 def metactl_export_from_grpc(addr: str) -> str:

tests/metactl/subcommands/cmd_export_from_grpc.py

@@ -71,9 +71,9 @@ def test_export_from_grpc():
 
     # Compare with expected output by converting all to JSON
     print(f"Got {len(lines)} lines, expected {len(want)} lines")
-    assert len(lines) == len(want), (
-        f"Line count mismatch: got {len(lines)}, expected {len(want)}"
-    )
+    assert len(lines) == len(
+        want
+    ), f"Line count mismatch: got {len(lines)}, expected {len(want)}"
 
     def normalize_json(obj):
         """Remove dynamic fields like time_ms from JSON object for comparison"""
@@ -100,9 +100,9 @@ def normalize_json(obj):
 
         want_json = normalize_json(want_json)
 
-        assert actual_json == want_json, (
-            f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
-        )
+        assert (
+            actual_json == want_json
+        ), f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
 
     print(f"✓ All {len(lines)} JSON lines match expected output")
 

tests/metactl/subcommands/cmd_export_from_raft_dir.py

@@ -75,9 +75,9 @@ def test_export_from_raft_dir():
 
     # Compare with expected output by converting all to JSON
     print(f"Got {len(lines)} lines, expected {len(want)} lines")
-    assert len(lines) == len(want), (
-        f"Line count mismatch: got {len(lines)}, expected {len(want)}"
-    )
+    assert len(lines) == len(
+        want
+    ), f"Line count mismatch: got {len(lines)}, expected {len(want)}"
 
     def normalize_json(obj):
         """Remove dynamic fields like time_ms from JSON object for comparison"""
@@ -105,9 +105,9 @@ def normalize_json(obj):
         except json.JSONDecodeError as e:
             assert False, f"Invalid JSON in expected line {i}: {want_line}, error: {e}"
 
-        assert actual_json == want_json, (
-            f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
-        )
+        assert (
+            actual_json == want_json
+        ), f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
 
     print(f"✓ All {len(lines)} JSON lines match expected output")
 

tests/metactl/subcommands/cmd_import.py

@@ -67,9 +67,9 @@ def test_import_subcommand():
         assert False, f"Import command failed with return code {process.returncode}"
 
     # Verify raft directory was created
-    assert os.path.exists(target_raft_dir), (
-        f"Raft directory should exist: {target_raft_dir}"
-    )
+    assert os.path.exists(
+        target_raft_dir
+    ), f"Raft directory should exist: {target_raft_dir}"
     print(f"✓ Raft directory created: {target_raft_dir}")
 
     # Check for raft log files in correct location
@@ -110,9 +110,9 @@ def test_import_subcommand():
 
     # Parse first line to check version
     header_data = json.loads(lines[0])
-    assert header_data[1]["DataHeader"]["value"]["version"] == "V004", (
-        "Should import V004 data"
-    )
+    assert (
+        header_data[1]["DataHeader"]["value"]["version"] == "V004"
+    ), "Should import V004 data"
     print("✓ Imported data version verification passed")
 
     # Check for required sections

tests/metactl/subcommands/cmd_lua_grpc.py

@@ -25,7 +25,7 @@ def test_lua_grpc_client():
     grpc_addr = setup_test_environment()
 
     # Create a Lua script that uses the gRPC client
-    lua_script = f'''
+    lua_script = f"""
 local client = metactl.new_grpc_client("{grpc_addr}")
 
 -- Test upsert operation
@@ -51,7 +51,7 @@ def test_lua_grpc_client():
 else
     print("Get null result:", metactl.to_string(get_null))
 end
-'''
+"""
 
     # Run metactl lua with gRPC client script
     result = subprocess.run(
@@ -71,9 +71,9 @@ def test_lua_grpc_client():
     print("expect:", expected_output)
 
     # Check if entire output matches expected value
-    assert output == expected_output, (
-        f"Expected:\n{expected_output}\n\nActual:\n{output}"
-    )
+    assert (
+        output == expected_output
+    ), f"Expected:\n{expected_output}\n\nActual:\n{output}"
 
     print("✓ Lua gRPC client test passed")
 

tests/metactl/subcommands/cmd_lua_spawn_concurrent.py

@@ -53,9 +53,9 @@ def test_spawn_basic():
     ]
 
     for phrase in expected_phrases:
-        assert phrase in output, (
-            f"Expected phrase '{phrase}' not found in output:\n{output}"
-        )
+        assert (
+            phrase in output
+        ), f"Expected phrase '{phrase}' not found in output:\n{output}"
     print("✓ Basic spawn functionality test passed")