feat(metrics): monitor s3 sdk retry (#9790)

grafana/risingwave-dev-dashboard.dashboard.py

@@ -555,6 +555,21 @@ def section_object_storage(outer_panels):
  )
  ],
  ),
+ panels.timeseries_ops(
+ "Operation Retry Rate",
+ "",
+ [
+ panels.target(
+ f"sum(irate({metric('aws_sdk_retry_counts')}[$__rate_interval])) by (instance, job, type)",
+ "{{type}} - {{job}} @ {{instance}}",
+ ),
+
+ panels.target(
+ f"sum(irate({metric('s3_read_request_retry_count')}[$__rate_interval])) by (instance, job, type)",
+ "{{type}} - {{job}} @ {{instance}}",
+ )
+ ],
+ ),
  panels.timeseries_dollar(
  "Estimated S3 Cost (Realtime)",
  "There are two types of operations: 1. GET, SELECT, and DELETE, they cost 0.0004 USD per 1000 "

src/cmd/Cargo.toml

@@ -21,6 +21,7 @@ normal = ["workspace-hack", "workspace-config", "task_stats_alloc"]
 [dependencies]
 anyhow = "1"
 clap = { version = "4", features = ["derive"] }
+prometheus = { version = "0.13" }
 risingwave_common = { path = "../common" }
 risingwave_compactor = { path = "../storage/compactor" }
 risingwave_compute = { path = "../compute" }

src/cmd/src/lib.rs

@@ -31,37 +31,39 @@ macro_rules! main {
  #[cfg_attr(coverage, no_coverage)]
  fn main() {
  let opts = clap::Parser::parse();
- $crate::$component(opts);
+ let registry = prometheus::Registry::new();
+ $crate::$component(opts, registry);
  }
  };
 }
 
 // Entry point functions.
 
-pub fn compute(opts: ComputeNodeOpts) {
+pub fn compute(opts: ComputeNodeOpts, registry: prometheus::Registry) {
  risingwave_rt::init_risingwave_logger(
  risingwave_rt::LoggerSettings::new().enable_tokio_console(false),
+ registry.clone(),
  );
- risingwave_rt::main_okk(risingwave_compute::start(opts));
+ risingwave_rt::main_okk(risingwave_compute::start(opts, registry));
 }
 
-pub fn meta(opts: MetaNodeOpts) {
- risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new());
+pub fn meta(opts: MetaNodeOpts, registry: prometheus::Registry) {
+ risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new(), registry);
  risingwave_rt::main_okk(risingwave_meta::start(opts));
 }
 
-pub fn frontend(opts: FrontendOpts) {
- risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new());
+pub fn frontend(opts: FrontendOpts, registry: prometheus::Registry) {
+ risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new(), registry);
  risingwave_rt::main_okk(risingwave_frontend::start(opts));
 }
 
-pub fn compactor(opts: CompactorOpts) {
- risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new());
+pub fn compactor(opts: CompactorOpts, registry: prometheus::Registry) {
+ risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new(), registry);
  risingwave_rt::main_okk(risingwave_compactor::start(opts));
 }
 
-pub fn ctl(opts: CtlOpts) {
- risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new());
+pub fn ctl(opts: CtlOpts, registry: prometheus::Registry) {
+ risingwave_rt::init_risingwave_logger(risingwave_rt::LoggerSettings::new(), registry);
 
  // Note: Use a simple current thread runtime for ctl.
  // When there's a heavy workload, multiple thread runtime seems to respond slowly. May need

src/cmd_all/Cargo.toml

@@ -22,6 +22,8 @@ ignored = ["workspace-hack", "workspace-config", "task_stats_alloc"]
 anyhow = "1"
 clap = { version = "4", features = ["cargo", "derive"] }
 console = "0.15"
+const-str = "0.5"
+prometheus = { version = "0.13" }
 risingwave_cmd = { path = "../cmd" }
 risingwave_common = { path = "../common" }
 risingwave_compactor = { path = "../storage/compactor" }

src/cmd_all/src/bin/risingwave.rs

@@ -57,14 +57,14 @@ impl Component {
  fn parse_opts<T: FromArgMatches>(matches: &ArgMatches) -> T {
  T::from_arg_matches(matches).map_err(|e| e.exit()).unwrap()
  }
-
+ let registry = prometheus::Registry::new();
  match self {
- Self::Compute => compute(parse_opts(matches)),
- Self::Meta => meta(parse_opts(matches)),
- Self::Frontend => frontend(parse_opts(matches)),
- Self::Compactor => compactor(parse_opts(matches)),
- Self::Ctl => ctl(parse_opts(matches)),
- Self::Playground => playground(parse_opts(matches)),
+ Self::Compute => compute(parse_opts(matches), registry),
+ Self::Meta => meta(parse_opts(matches), registry),
+ Self::Frontend => frontend(parse_opts(matches), registry),
+ Self::Compactor => compactor(parse_opts(matches), registry),
+ Self::Ctl => ctl(parse_opts(matches), registry),
+ Self::Playground => playground(parse_opts(matches), registry),
  }
  }
 
@@ -136,10 +136,10 @@ fn main() -> Result<()> {
  Ok(())
 }
 
-fn playground(opts: PlaygroundOpts) {
+fn playground(opts: PlaygroundOpts, registry: prometheus::Registry) {
  let settings = risingwave_rt::LoggerSettings::new()
  .enable_tokio_console(false)
  .with_target("risingwave_storage", Level::WARN);
- risingwave_rt::init_risingwave_logger(settings);
+ risingwave_rt::init_risingwave_logger(settings, registry);
  risingwave_rt::main_okk(risingwave_cmd_all::playground(opts)).unwrap();
 }

src/cmd_all/src/playground.rs

@@ -215,8 +215,9 @@ pub async fn playground(opts: PlaygroundOpts) -> Result<()> {
  opts.insert(0, "compute-node".into());
  tracing::info!("starting compute-node thread with cli args: {:?}", opts);
  let opts = risingwave_compute::ComputeNodeOpts::parse_from(opts);
- let _compute_handle =
- tokio::spawn(async move { risingwave_compute::start(opts).await });
+ let _compute_handle = tokio::spawn(async move {
+ risingwave_compute::start(opts, prometheus::Registry::new()).await
+ });
  }
  RisingWaveService::Frontend(mut opts) => {
  opts.insert(0, "frontend-node".into());

src/common/Cargo.toml

@@ -89,6 +89,7 @@ tokio = { version = "0.2", package = "madsim-tokio", features = [
 toml = "0.7"
 tonic = { version = "0.2", package = "madsim-tonic" }
 tracing = "0.1"
+tracing-subscriber = "0.3.16"
 twox-hash = "1"
 url = "2"
 uuid = "1.2.2"

src/common/src/metrics.rs

@@ -13,8 +13,12 @@
 // limitations under the License.
 
 use hytra::TrAdder;
-use prometheus::core::{Atomic, GenericGauge};
-
+use prometheus::core::{Atomic, AtomicU64, GenericCounter, GenericGauge};
+use prometheus::{register_int_counter_with_registry, Registry};
+use tracing::Subscriber;
+use tracing_subscriber::layer::Context;
+use tracing_subscriber::registry::LookupSpan;
+use tracing_subscriber::Layer;
 pub struct TrAdderAtomic(TrAdder<i64>);
 
 impl Atomic for TrAdderAtomic {
@@ -44,3 +48,37 @@ impl Atomic for TrAdderAtomic {
 }
 
 pub type TrAdderGauge = GenericGauge<TrAdderAtomic>;
+
+/// [`MetricsLayer`] is a struct used for monitoring the frequency of certain specific logs and
+/// counting them using Prometheus metrics. Currently, it is used to monitor the frequency of retry
+/// occurrences of aws sdk.
+pub struct MetricsLayer {
+ pub aws_sdk_retry_counts: GenericCounter<AtomicU64>,
+}
+
+impl<S> Layer<S> for MetricsLayer
+where
+ S: Subscriber + for<'a> LookupSpan<'a>,
+{
+ fn on_event(&self, _event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
+ // Currently one retry will only generate one debug log,
+ // so we can monitor the number of retry only through the metadata target.
+ // Refer to <https://docs.rs/aws-smithy-client/0.55.3/src/aws_smithy_client/retry.rs.html>
+ self.aws_sdk_retry_counts.inc();
+ }
+}
+
+impl MetricsLayer {
+ pub fn new(registry: Registry) -> Self {
+ let aws_sdk_retry_counts = register_int_counter_with_registry!(
+ "aws_sdk_retry_counts",
+ "Total number of aws sdk retry happens",
+ registry
+ )
+ .unwrap();
+
+ Self {
+ aws_sdk_retry_counts,
+ }
+ }
+}

src/compute/src/lib.rs

@@ -178,7 +178,10 @@ use std::pin::Pin;
 use crate::server::compute_node_serve;
 
 /// Start compute node
-pub fn start(opts: ComputeNodeOpts) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+pub fn start(
+ opts: ComputeNodeOpts,
+ registry: prometheus::Registry,
+) -> Pin<Box<dyn Future<Output = ()> + Send>> {
  // WARNING: don't change the function signature. Making it `async fn` will cause
  // slow compile in release mode.
  Box::pin(async move {
@@ -200,7 +203,7 @@ pub fn start(opts: ComputeNodeOpts) -> Pin<Box<dyn Future<Output = ()> + Send>>
  tracing::info!("advertise addr is {}", advertise_addr);
 
  let (join_handle_vec, _shutdown_send) =
- compute_node_serve(listen_addr, advertise_addr, opts).await;
+ compute_node_serve(listen_addr, advertise_addr, opts, registry).await;
 
  for join_handle in join_handle_vec {
  join_handle.await.unwrap();

src/compute/src/server.rs

@@ -81,6 +81,7 @@ pub async fn compute_node_serve(
  listen_addr: SocketAddr,
  advertise_addr: HostAddr,
  opts: ComputeNodeOpts,
+ registry: prometheus::Registry,
 ) -> (Vec<JoinHandle<()>>, Sender<()>) {
  // Load the configuration.
  let config = load_config(&opts.config_path, Some(opts.override_config.clone()));
@@ -161,7 +162,7 @@ pub async fn compute_node_serve(
 
  let mut sub_tasks: Vec<(JoinHandle<()>, Sender<()>)> = vec![];
  // Initialize the metrics subsystem.
- let registry = prometheus::Registry::new();
+
  monitor_process(&registry).unwrap();
  let source_metrics = Arc::new(SourceMetrics::new(registry.clone()));
  let hummock_metrics = Arc::new(HummockMetrics::new(registry.clone()));

src/object_store/src/object/object_metrics.rs

@@ -25,6 +25,7 @@ pub struct ObjectStoreMetrics {
  pub operation_latency: HistogramVec,
  pub operation_size: HistogramVec,
  pub failure_count: GenericCounterVec<AtomicU64>,
+ pub request_retry_count: GenericCounterVec<AtomicU64>,
 }
 
 impl ObjectStoreMetrics {
@@ -81,12 +82,21 @@ impl ObjectStoreMetrics {
  )
  .unwrap();
 
+ let request_retry_count = register_int_counter_vec_with_registry!(
+ "s3_read_request_retry_count",
+ "The number of retry times of object store request",
+ &["type"],
+ registry
+ )
+ .unwrap();
+
  Self {
  write_bytes,
  read_bytes,
  operation_latency,
  operation_size,
  failure_count,
+ request_retry_count,
  }
  }