-
Notifications
You must be signed in to change notification settings - Fork 11.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[indexer-alt] Log TPS and CPS in all pipelines (#20441)
## Description Describe the changes or additions included in this PR. ## Test plan How did you test the new or updated feature? --- ## Release notes Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required. For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates. - [ ] Protocol: - [ ] Nodes (Validators and Full nodes): - [ ] Indexer: - [ ] JSON-RPC: - [ ] GraphQL: - [ ] CLI: - [ ] Rust SDK: - [ ] REST API:
- Loading branch information
Showing
5 changed files
with
146 additions
and
114 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
104 changes: 104 additions & 0 deletions
104
crates/sui-indexer-alt-framework/src/pipeline/logging.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| // Copyright (c) Mysten Labs, Inc. | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| use std::time::Instant; | ||
|
|
||
| use tracing::{debug, info}; | ||
|
|
||
| use crate::watermarks::PrunerWatermark; | ||
|
|
||
| use super::{CommitterWatermark, Processor}; | ||
|
|
||
| /// Tracing message for the watermark update will be logged at info level at least this many | ||
| /// checkpoints. | ||
| const LOUD_WATERMARK_UPDATE_INTERVAL: i64 = 5 * 10; | ||
|
|
||
| #[derive(Default)] | ||
| pub(crate) struct LoggerWatermark { | ||
| checkpoint: i64, | ||
| transaction: Option<i64>, | ||
| } | ||
|
|
||
| pub(crate) struct WatermarkLogger { | ||
| name: &'static str, | ||
| timer: Instant, | ||
| prev_watermark: LoggerWatermark, | ||
| } | ||
|
|
||
| impl WatermarkLogger { | ||
| pub fn new(name: &'static str, init_watermark: impl Into<LoggerWatermark>) -> Self { | ||
| Self { | ||
| name, | ||
| timer: Instant::now(), | ||
| prev_watermark: init_watermark.into(), | ||
| } | ||
| } | ||
|
|
||
| /// Log the watermark update. | ||
| /// `watermark_update_latency` is the time spent to update the watermark. | ||
| /// | ||
| /// Given the new watermark, the logger will compare with the previous watermark to compute the | ||
| /// average TPS (transactions per second) and CPS (checkpoints per second) since the last update. | ||
| /// | ||
| /// If the watermark update is less than `LOUD_WATERMARK_UPDATE_INTERVAL` checkpoints apart, | ||
| /// the log message will be at debug level. Otherwise, it will be at info level. | ||
| pub fn log<H: Processor>( | ||
| &mut self, | ||
| watermark: impl Into<LoggerWatermark>, | ||
| watermark_update_latency: f64, | ||
| ) { | ||
| let watermark: LoggerWatermark = watermark.into(); | ||
| let logger_timer_elapsed = self.timer.elapsed().as_secs_f64(); | ||
| let realtime_average_tps = match (self.prev_watermark.transaction, watermark.transaction) { | ||
| (Some(prev), Some(curr)) => Some((curr - prev) as f64 / logger_timer_elapsed), | ||
| _ => None, | ||
| }; | ||
| let realtime_average_cps = | ||
| (watermark.checkpoint - self.prev_watermark.checkpoint) as f64 / logger_timer_elapsed; | ||
|
|
||
| if watermark.checkpoint < self.prev_watermark.checkpoint + LOUD_WATERMARK_UPDATE_INTERVAL { | ||
| debug!( | ||
| logger = self.name, | ||
| pipeline = H::NAME, | ||
| checkpoint = watermark.checkpoint, | ||
| transaction = watermark.transaction, | ||
| tps = realtime_average_tps, | ||
| cps = realtime_average_cps, | ||
| elapsed_ms = format!("{:.3}", watermark_update_latency * 1000.0), | ||
| "Updated watermark", | ||
| ); | ||
| return; | ||
| } | ||
|
|
||
| info!( | ||
| logger = self.name, | ||
| pipeline = H::NAME, | ||
| checkpoint = watermark.checkpoint, | ||
| transaction = watermark.transaction, | ||
| tps = realtime_average_tps, | ||
| cps = realtime_average_cps, | ||
| elapsed_ms = format!("{:.3}", watermark_update_latency * 1000.0), | ||
| "Updated watermark", | ||
| ); | ||
| self.prev_watermark = watermark; | ||
| self.timer = Instant::now(); | ||
| } | ||
| } | ||
|
|
||
| impl From<&CommitterWatermark<'_>> for LoggerWatermark { | ||
| fn from(watermark: &CommitterWatermark) -> Self { | ||
| Self { | ||
| checkpoint: watermark.checkpoint_hi_inclusive, | ||
| transaction: Some(watermark.tx_hi), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl From<&PrunerWatermark<'_>> for LoggerWatermark { | ||
| fn from(watermark: &PrunerWatermark) -> Self { | ||
| Self { | ||
| checkpoint: watermark.pruner_hi, | ||
| transaction: None, | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters