enable verbose by default

Signed-off-by: Bugen Zhao <[email protected]>

Author: BugenZhao

Cargo.lock

@@ -58,6 +58,7 @@ license = "Apache-2.0"
 repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
+await-tree = "0.1.1"
 aws-config = { version = "0.51", default-features = false, features = ["rt-tokio", "native-tls"] }
 aws-sdk-kinesis = { version = "0.21", default-features = false, features = ["rt-tokio", "native-tls"] }
 aws-sdk-s3 = { version = "0.21", default-features = false, features = ["rt-tokio","native-tls"] }

Cargo.toml

@@ -420,12 +420,27 @@ pub struct FileCacheConfig {
     pub unrecognized: Unrecognized<Self>,
 }
 
-#[derive(Debug, Default, Clone, ValueEnum, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Copy, ValueEnum, Serialize, Deserialize)]
 pub enum AsyncStackTraceOption {
+    /// Disabled.
     Off,
-    #[default]
+    /// Enabled with basic instruments.
     On,
-    Verbose,
+    /// Enabled with extra verbose instruments in release build.
+    /// Behaves the same as `on` in debug build due to performance concern.
+    #[default]
+    #[clap(alias = "verbose")]
+    ReleaseVerbose,
+}
+
+impl AsyncStackTraceOption {
+    pub fn is_verbose(self) -> Option<bool> {
+        match self {
+            Self::Off => None,
+            Self::On => Some(false),
+            Self::ReleaseVerbose => Some(!cfg!(debug_assertions)),
+        }
+    }
 }
 
 serde_with::with_prefix!(streaming_prefix "stream_");
@@ -723,7 +738,7 @@ mod default {
         }
 
         pub fn async_stack_trace() -> AsyncStackTraceOption {
-            AsyncStackTraceOption::On
+            AsyncStackTraceOption::default()
         }
 
         pub fn unique_user_stream_errors() -> usize {

src/common/src/config.rs

@@ -16,7 +16,7 @@ normal = ["workspace-hack"]
 [dependencies]
 anyhow = "1"
 async-trait = "0.1"
-await-tree = "0.1.1"
+await-tree = { workspace = true }
 clap = { version = "4", features = ["derive"] }
 either = "1"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }

src/compute/Cargo.toml

@@ -263,7 +263,7 @@ pub async fn compute_node_serve(
     let await_tree_config = match &config.streaming.async_stack_trace {
         AsyncStackTraceOption::Off => None,
         c => await_tree::ConfigBuilder::default()
-            .verbose(matches!(c, AsyncStackTraceOption::Verbose))
+            .verbose(c.is_verbose().unwrap())
             .build()
             .ok(),
     };

src/compute/src/server.rs

@@ -1,80 +1,2 @@
 # This file is generated by ./risedev generate-example-config
 # Check detailed comments in src/common/src/config.rs
-[server]
-heartbeat_interval_ms = 1000
-connection_pool_size = 16
-metrics_level = 0
-telemetry_enabled = true
-
-[meta]
-min_sst_retention_time_sec = 604800
-collect_gc_watermark_spin_interval_sec = 5
-periodic_compaction_interval_sec = 60
-vacuum_interval_sec = 30
-hummock_version_checkpoint_interval_sec = 30
-min_delta_log_num_for_hummock_version_checkpoint = 10
-max_heartbeat_interval_secs = 300
-disable_recovery = false
-meta_leader_lease_secs = 30
-enable_compaction_deterministic = false
-enable_committed_sst_sanity_check = false
-node_num_monitor_interval_sec = 10
-backend = "Mem"
-periodic_space_reclaim_compaction_interval_sec = 3600
-periodic_ttl_reclaim_compaction_interval_sec = 1800
-periodic_split_compact_group_interval_sec = 180
-max_compactor_task_multiplier = 2
-move_table_size_limit = 2147483648
-split_group_size_limit = 21474836480
-do_not_config_object_storage_lifecycle = false
-partition_vnode_count = 64
-
-[batch.developer]
-batch_connector_message_buffer_size = 16
-batch_output_channel_size = 64
-batch_chunk_size = 1024
-
-[streaming]
-in_flight_barrier_nums = 10000
-enable_jaeger_tracing = false
-async_stack_trace = "On"
-unique_user_stream_errors = 10
-
-[streaming.developer]
-stream_enable_executor_row_count = false
-stream_connector_message_buffer_size = 16
-stream_unsafe_extreme_cache_size = 10
-stream_chunk_size = 1024
-stream_exchange_initial_permits = 8192
-stream_exchange_batched_permits = 1024
-stream_exchange_concurrent_barriers = 2
-
-[storage]
-share_buffers_sync_parallelism = 1
-share_buffer_compaction_worker_threads_number = 4
-imm_merge_threshold = 4
-write_conflict_detection_enabled = true
-disable_remote_compactor = false
-share_buffer_upload_concurrency = 8
-sstable_id_remote_fetch_number = 10
-min_sst_size_for_streaming_upload = 33554432
-max_sub_compaction = 4
-max_concurrent_compaction_task_number = 16
-max_preload_wait_time_mill = 10
-
-[storage.file_cache]
-dir = ""
-capacity_mb = 1024
-cache_file_fallocate_unit_mb = 512
-cache_meta_fallocate_unit_mb = 16
-cache_file_max_write_size_mb = 4
-
-[system]
-barrier_interval_ms = 1000
-checkpoint_frequency = 10
-sstable_size_mb = 256
-block_size_kb = 64
-bloom_false_positive = 0.001
-backup_storage_url = "memory"
-backup_storage_directory = "backup"
-telemetry_enabled = true

src/config/example.toml

@@ -21,6 +21,7 @@ arrow-array = "36"
 arrow-schema = "36"
 async-trait = "0.1"
 auto_enums = "0.8"
+await-tree = { workspace = true }
 chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
 chrono-tz = { version = "0.7", features = ["case-insensitive"] }
 ctor = "0.1"

src/expr/Cargo.toml

@@ -17,6 +17,7 @@ use std::convert::TryFrom;
 use std::sync::{Arc, LazyLock, Mutex, Weak};
 
 use arrow_schema::{Field, Schema, SchemaRef};
+use await_tree::InstrumentAwait;
 use risingwave_common::array::{ArrayImpl, ArrayRef, DataChunk};
 use risingwave_common::row::OwnedRow;
 use risingwave_common::types::{DataType, Datum};
@@ -35,6 +36,7 @@ pub struct UdfExpression {
     arg_schema: SchemaRef,
     client: Arc<ArrowFlightUdfClient>,
     identifier: String,
+    span: await_tree::Span,
 }
 
 #[cfg(not(madsim))]
@@ -80,7 +82,11 @@ impl UdfExpression {
         let input =
             arrow_array::RecordBatch::try_new_with_options(self.arg_schema.clone(), columns, &opts)
                 .expect("failed to build record batch");
-        let output = self.client.call(&self.identifier, input).await?;
+        let output = self
+            .client
+            .call(&self.identifier, input)
+            .instrument_await(self.span.clone())
+            .await?;
         if output.num_rows() != vis.len() {
             bail!(
                 "UDF returned {} rows, but expected {}",
@@ -121,6 +127,7 @@ impl<'a> TryFrom<&'a ExprNode> for UdfExpression {
             arg_schema,
             client,
             identifier: udf.identifier.clone(),
+            span: format!("expr_udf_call ({})", udf.identifier).into(),
         })
     }
 }

src/expr/src/expr/expr_udf.rs

@@ -10,7 +10,7 @@ repository = { workspace = true }
 
 [dependencies]
 async-trait = "0.1"
-await-tree = "0.1.1"
+await-tree = { workspace = true }
 aws-config = { workspace = true }
 aws-sdk-s3 = { version = "0.2.15", package = "madsim-aws-sdk-s3" }
 aws-smithy-http = { workspace = true }

src/object_store/Cargo.toml

@@ -17,7 +17,7 @@ normal = ["workspace-hack"]
 arc-swap = "1"
 async-trait = "0.1"
 auto_enums = { version = "0.8", features = ["futures03"] }
-await-tree = "0.1.1"
+await-tree = { workspace = true }
 bytes = { version = "1", features = ["serde"] }
 crossbeam = "0.8.1"
 dashmap = { version = "5", default-features = false }

src/storage/Cargo.toml

@@ -185,7 +185,11 @@ impl<S: LocalStateStore> LocalStateStore for MonitoredStateStore<S> {
                 .may_exist_duration
                 .with_label_values(&[table_id_label.as_str()])
                 .start_timer();
-            let res = self.inner.may_exist(key_range, read_options).await;
+            let res = self
+                .inner
+                .may_exist(key_range, read_options)
+                .verbose_instrument_await("store_may_exist")
+                .await;
             timer.observe_duration();
             res
         }
@@ -217,7 +221,9 @@ impl<S: LocalStateStore> LocalStateStore for MonitoredStateStore<S> {
 
     fn flush(&mut self, delete_ranges: Vec<(Bound<Bytes>, Bound<Bytes>)>) -> Self::FlushFuture<'_> {
         // TODO: collect metrics
-        self.inner.flush(delete_ranges)
+        self.inner
+            .flush(delete_ranges)
+            .verbose_instrument_await("store_flush")
     }
 
     fn epoch(&self) -> u64 {
@@ -264,7 +270,7 @@ impl<S: StateStore> StateStore for MonitoredStateStore<S> {
             let sync_result = self
                 .inner
                 .sync(epoch)
-                .instrument_await("store_await_sync")
+                .instrument_await("store_sync")
                 .await
                 .inspect_err(|e| error!("Failed in sync: {:?}", e))?;
             timer.observe_duration();

src/storage/src/monitor/monitored_store.rs

@@ -19,7 +19,7 @@ anyhow = "1"
 async-recursion = "1"
 async-stream = "0.3"
 async-trait = "0.1"
-await-tree = "0.1.1"
+await-tree = { workspace = true }
 bytes = "1"
 dyn-clone = "1"
 educe = "0.4"

src/stream/Cargo.toml

@@ -15,7 +15,7 @@ ignored = ["workspace-hack"]
 normal = ["workspace-hack"]
 
 [dependencies]
-await-tree = "0.1.1"
+await-tree = { workspace = true }
 console = "0.15"
 console-subscriber = "0.1.8"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }