From 8745f095ae3bc3f0ac7b19b9d0bb5eafb3cd0a4c Mon Sep 17 00:00:00 2001 From: Simon Paitrault Date: Wed, 5 Jul 2023 10:44:22 +0200 Subject: [PATCH] refactor: update code to match lint expectation Signed-off-by: Simon Paitrault --- .github/workflows/test.yml | 2 +- Cargo.lock | 87 +- Cargo.toml | 4 +- crates/topos-metrics/src/lib.rs | 1 + crates/topos-p2p/Cargo.toml | 1 + crates/topos-p2p/src/behaviour/discovery.rs | 7 +- crates/topos-p2p/src/behaviour/gossip.rs | 28 +- crates/topos-p2p/src/config.rs | 8 + crates/topos-p2p/src/constant.rs | 1 + crates/topos-p2p/src/network.rs | 16 +- .../topos-p2p/src/runtime/handle_command.rs | 2 +- crates/topos-p2p/src/runtime/handle_event.rs | 2 +- .../src/runtime/handle_event/peer_info.rs | 16 +- crates/topos-p2p/src/runtime/mod.rs | 23 +- crates/topos-p2p/src/tests/dht.rs | 41 +- crates/topos-p2p/src/tests/support/macros.rs | 1 + .../src/double_echo/mod.rs | 281 ++- crates/topos-tce-broadcast/src/lib.rs | 5 +- crates/topos-tce-broadcast/src/mem_store.rs | 110 -- crates/topos-tce-broadcast/src/tce_store.rs | 44 - crates/topos-tce-broadcast/src/tests/mod.rs | 9 + crates/topos-tce-proxy/tests/tce_tests.rs | 1 + crates/topos-tce-storage/src/client.rs | 10 +- crates/topos-tce-storage/src/command.rs | 1 + .../src/connection/handlers.rs | 12 +- crates/topos-tce/src/app_context.rs | 51 +- crates/topos-tce/src/lib.rs | 37 +- crates/topos-test-sdk/src/tce/mod.rs | 53 +- crates/topos-test-sdk/src/tce/p2p.rs | 1 - crates/topos/tests/cert_delivery.rs | 11 +- crates/topos/tests/push-certificate.rs | 5 +- tools/config/prometheus/prometheus.yml | 13 +- tools/dashboard.json | 1591 +++++++++++++++-- tools/docker-compose.yml | 1 + 34 files changed, 1887 insertions(+), 589 deletions(-) delete mode 100644 crates/topos-tce-broadcast/src/mem_store.rs delete mode 100644 crates/topos-tce-broadcast/src/tce_store.rs diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ba691ccad..ceb3f5d46 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -26,7 +26,7 @@ jobs: - run: cargo nextest run --workspace --exclude topos-sequencer-subnet-runtime && cargo test --doc --workspace env: - RUST_LOG: topos=warn + RUST_LOG: warn,topos=info test_nightly: runs-on: ubuntu-latest diff --git a/Cargo.lock b/Cargo.lock index f4b56c860..a58c3de56 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3050,9 +3050,9 @@ checksum = "28b29a3cd74f0f4598934efe3aeba42bae0eb4680554128851ebbecb02af14e6" [[package]] name = "is-terminal" -version = "0.4.8" +version = "0.4.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24fddda5af7e54bf7da53067d6e802dbcc381d0a8eef629df528e3ebf68755cb" +checksum = "cb0889898416213fab133e1d33a0e5858a48177452750691bde3666d0fdbaf8b" dependencies = [ "hermit-abi 0.3.2", "rustix 0.38.3", @@ -3196,7 +3196,8 @@ checksum = "f7012b1bbb0719e1097c47611d3898568c546d597c2e74d66f6087edd5233ff4" [[package]] name = "libp2p" version = "0.52.1" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38039ba2df4f3255842050845daef4a004cc1f26da03dbc645535088b51910ef" dependencies = [ "bytes", "futures", @@ -3225,7 +3226,8 @@ dependencies = [ [[package]] name = "libp2p-allow-block-list" version = "0.2.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55b46558c5c0bf99d3e2a1a38fd54ff5476ca66dd1737b12466a1824dd219311" dependencies = [ "libp2p-core", "libp2p-identity", @@ -3236,7 +3238,8 @@ dependencies = [ [[package]] name = "libp2p-connection-limits" version = "0.2.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45dd90e8f0e1fa59e85ff5316dd4d1ac41a9a507e79cda1b0e9b7be43ad1a56" dependencies = [ "libp2p-core", "libp2p-identity", @@ -3247,7 +3250,8 @@ dependencies = [ [[package]] name = "libp2p-core" version = "0.40.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef7dd7b09e71aac9271c60031d0e558966cdb3253ba0308ab369bb2de80630d0" dependencies = [ "either", "fnv", @@ -3275,7 +3279,8 @@ dependencies = [ [[package]] name = "libp2p-dns" version = "0.40.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd4394c81c0c06d7b4a60f3face7e8e8a9b246840f98d2c80508d0721b032147" dependencies = [ "futures", "libp2p-core", @@ -3289,7 +3294,8 @@ dependencies = [ [[package]] name = "libp2p-gossipsub" version = "0.45.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e378da62e8c9251f6e885ed173a561663f29b251e745586cf6ae6150b295c37" dependencies = [ "asynchronous-codec", "base64 0.21.2", @@ -3321,7 +3327,8 @@ dependencies = [ [[package]] name = "libp2p-identify" version = "0.43.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a29675a32dbcc87790db6cf599709e64308f1ae9d5ecea2d259155889982db8" dependencies = [ "asynchronous-codec", "either", @@ -3360,7 +3367,8 @@ dependencies = [ [[package]] name = "libp2p-kad" version = "0.44.1" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5588b884dcb1dadc04e49de342f634f60cf28b6beaaca5a4fe3dd1f09bb30041" dependencies = [ "arrayvec", "asynchronous-codec", @@ -3388,7 +3396,8 @@ dependencies = [ [[package]] name = "libp2p-mdns" version = "0.44.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42a2567c305232f5ef54185e9604579a894fd0674819402bb0ac0246da82f52a" dependencies = [ "data-encoding", "futures", @@ -3408,7 +3417,8 @@ dependencies = [ [[package]] name = "libp2p-metrics" version = "0.13.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3787ea81798dcc5bf1d8b40a8e8245cf894b168d04dd70aa48cb3ff2fff141d2" dependencies = [ "instant", "libp2p-core", @@ -3424,7 +3434,8 @@ dependencies = [ [[package]] name = "libp2p-noise" version = "0.43.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87945db2b3f977af09b62b9aa0a5f3e4870995a577ecd845cdeba94cdf6bbca7" dependencies = [ "bytes", "curve25519-dalek 3.2.0", @@ -3448,7 +3459,8 @@ dependencies = [ [[package]] name = "libp2p-request-response" version = "0.25.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20bd837798cdcce4283d2675f08bcd3756a650d56eab4d4367e1b3f27eed6887" dependencies = [ "async-trait", "futures", @@ -3465,7 +3477,8 @@ dependencies = [ [[package]] name = "libp2p-swarm" version = "0.43.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6f1fe3817492f88c5298c8b5fbaa5ff3a0c802ecf4e79be4e341cf07abfa82f" dependencies = [ "either", "fnv", @@ -3487,7 +3500,8 @@ dependencies = [ [[package]] name = "libp2p-swarm-derive" version = "0.33.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4d5ec2a3df00c7836d7696c136274c9c59705bac69133253696a6c932cd1d74" dependencies = [ "heck 0.4.1", "proc-macro-warning", @@ -3499,7 +3513,8 @@ dependencies = [ [[package]] name = "libp2p-tcp" version = "0.40.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09bfdfb6f945c5c014b87872a0bdb6e0aef90e92f380ef57cd9013f118f9289d" dependencies = [ "futures", "futures-timer", @@ -3515,7 +3530,8 @@ dependencies = [ [[package]] name = "libp2p-yamux" version = "0.44.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a9b42ab6de15c6f076d8fb11dc5f48d899a10b55a2e16b12be9012a05287b0" dependencies = [ "futures", "libp2p-core", @@ -3828,7 +3844,8 @@ checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a" [[package]] name = "multistream-select" version = "0.13.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea0df8e5eec2298a62b326ee4f0d7fe1a6b90a09dfcf9df37b38f947a8c42f19" dependencies = [ "bytes", "futures", @@ -4887,7 +4904,8 @@ dependencies = [ [[package]] name = "quick-protobuf-codec" version = "0.2.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8ededb1cd78531627244d51dd0c7139fbe736c7d57af0092a76f0ffb2f56e98" dependencies = [ "asynchronous-codec", "bytes", @@ -5389,7 +5407,8 @@ checksum = "dc31bd9b61a32c31f9650d18add92aa83a49ba979c143eefd27fe7177b05bd5f" [[package]] name = "rw-stream-sink" version = "0.4.0" -source = "git+https://github.com/libp2p/rust-libp2p.git?branch=bump-yamux#7f90bb3b4314d7c221fded58b0ba04a5538e839b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8c9026ff5d2f23da5e45bbc283f156383001bfb09c4e44256d02c1a685fe9a1" dependencies = [ "futures", "pin-project", @@ -5592,18 +5611,18 @@ checksum = "cd0b0ec5f1c1ca621c432a25813d8d60c88abe6d3e08a3eb9cf37d97a0fe3d73" [[package]] name = "serde" -version = "1.0.166" +version = "1.0.167" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d01b7404f9d441d3ad40e6a636a7782c377d2abdbe4fa2440e2edcc2f4f10db8" +checksum = "7daf513456463b42aa1d94cff7e0c24d682b429f020b9afa4f5ba5c40a22b237" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.166" +version = "1.0.167" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5dd83d6dde2b6b2d466e14d9d1acce8816dedee94f735eac6395808b3483c6d6" +checksum = "b69b106b68bc8054f0e974e70d19984040f8a5cf9215ca82626ea4853f82c4b9" dependencies = [ "proc-macro2", "quote", @@ -6144,18 +6163,18 @@ checksum = "222a222a5bfe1bba4a77b45ec488a741b3cb8872e5e499451fd7d0129c9c7c3d" [[package]] name = "thiserror" -version = "1.0.41" +version = "1.0.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c16a64ba9387ef3fdae4f9c1a7f07a0997fce91985c0336f1ddc1822b3b37802" +checksum = "a35fc5b8971143ca348fa6df4f024d4d55264f3468c71ad1c2f365b0a4d58c42" dependencies = [ "thiserror-impl", ] [[package]] name = "thiserror-impl" -version = "1.0.41" +version = "1.0.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d14928354b01c4d6a4f0e549069adef399a284e7995c7ccca94e8a07a5346c59" +checksum = "463fe12d7993d3b327787537ce8dd4dfa058de32fc2b195ef3cde03dc4771e8f" dependencies = [ "proc-macro2", "quote", @@ -7803,9 +7822,9 @@ checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a" [[package]] name = "winnow" -version = "0.4.7" +version = "0.4.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca0ace3845f0d96209f0375e6d367e3eb87eb65d27d445bdc9f1843a26f39448" +checksum = "a9482fe6ceabdf32f3966bfdd350ba69256a97c30253dc616fe0005af24f164e" dependencies = [ "memchr", ] @@ -7888,14 +7907,14 @@ dependencies = [ [[package]] name = "yamux" -version = "0.12.0" -source = "git+https://github.com/thomaseizinger/rust-yamux?branch=feat/256-backlog#7b4fdba7105c0e1937265b6753cc5954f978a8b9" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5d9ba232399af1783a58d8eb26f6b5006fbefe2dc9ef36bd283324792d03ea5" dependencies = [ "futures", "log", "nohash-hasher", "parking_lot", - "pin-project", "rand 0.8.5", "static_assertions", ] diff --git a/Cargo.toml b/Cargo.toml index f9e4f0e26..0c1632e39 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,8 +51,8 @@ http = "0.2.9" tower-http = { version = "0.4", features = ["cors"] } # P2P related -# libp2p = { version = "0.52", default-features = false, features = ["noise"]} -libp2p = { branch = "bump-yamux", git = "https://github.com/libp2p/rust-libp2p.git", default-features = false, features = ["noise"]} +libp2p = { version = "0.52", default-features = false, features = ["noise"]} + # Serialization & Deserialization bincode = { version = "1.3", default-features = false } byteorder = { version = "1.4", default-features = false } diff --git a/crates/topos-metrics/src/lib.rs b/crates/topos-metrics/src/lib.rs index 04c713b74..dc2793536 100644 --- a/crates/topos-metrics/src/lib.rs +++ b/crates/topos-metrics/src/lib.rs @@ -72,4 +72,5 @@ pub fn init_metrics() { CERTIFICATE_RECEIVED_FROM_GOSSIP_TOTAL.reset(); CERTIFICATE_RECEIVED_FROM_API_TOTAL.reset(); CERTIFICATE_DELIVERED_TOTAL.reset(); + STORAGE_COMMAND_CHANNEL_CAPACITY_TOTAL.reset(); } diff --git a/crates/topos-p2p/Cargo.toml b/crates/topos-p2p/Cargo.toml index d22062218..d2d46ab96 100644 --- a/crates/topos-p2p/Cargo.toml +++ b/crates/topos-p2p/Cargo.toml @@ -16,6 +16,7 @@ tokio-stream.workspace = true tracing = { workspace = true, features = ["attributes"] } libp2p = { workspace = true, features = ["macros", "gossipsub", "tcp", "dns", "tokio", "request-response", "identify", "kad", "serde", "yamux"] } + void = "1" topos-metrics = { path = "../topos-metrics/" } diff --git a/crates/topos-p2p/src/behaviour/discovery.rs b/crates/topos-p2p/src/behaviour/discovery.rs index 2d35fb2db..e12a3ef42 100644 --- a/crates/topos-p2p/src/behaviour/discovery.rs +++ b/crates/topos-p2p/src/behaviour/discovery.rs @@ -35,7 +35,6 @@ impl DiscoveryBehaviour { ) -> Self { let local_peer_id = peer_key.public().to_peer_id(); let kademlia_config = KademliaConfig::default() - .set_protocol_names(vec![StreamProtocol::new(TRANSMISSION_PROTOCOL)]) .set_replication_factor(config.replication_factor) .set_kbucket_inserts(KademliaBucketInserts::Manual) .set_replication_interval(config.replication_interval) @@ -57,9 +56,9 @@ impl DiscoveryBehaviour { kademlia.add_address(&known_peer.0, known_peer.1.clone()); } - if let Err(store_error) = kademlia.start_providing("topos-tce".as_bytes().to_vec().into()) { - warn!(reason = %store_error, "Could not start providing Kademlia protocol `topos-tce`") - } + // if let Err(store_error) = kademlia.start_providing("topos-tce".as_bytes().to_vec().into()) { + // warn!(reason = %store_error, "Could not start providing Kademlia protocol `topos-tce`") + // } Self { inner: kademlia } } diff --git a/crates/topos-p2p/src/behaviour/gossip.rs b/crates/topos-p2p/src/behaviour/gossip.rs index 348124e02..96f250c47 100644 --- a/crates/topos-p2p/src/behaviour/gossip.rs +++ b/crates/topos-p2p/src/behaviour/gossip.rs @@ -6,7 +6,7 @@ use std::{ }; use libp2p::{ - gossipsub::{self, IdentTopic, Message, MessageAuthenticity, MessageId, Topic}, + gossipsub::{self, IdentTopic, Message, MessageAuthenticity, MessageId, Topic, TopicHash}, identity::Keypair, swarm::{NetworkBehaviour, THandlerInEvent, ToSwarm}, }; @@ -15,6 +15,7 @@ use topos_metrics::{ P2P_DUPLICATE_MESSAGE_ID_RECEIVED_TOTAL, P2P_GOSSIP_BATCH_SIZE, P2P_MESSAGE_SERIALIZE_FAILURE_TOTAL, }; +use tracing::{debug, error, info}; use crate::{constant, event::ComposedEvent, TOPOS_ECHO, TOPOS_GOSSIP, TOPOS_READY}; @@ -74,7 +75,7 @@ impl Behaviour { .unwrap(); let gossipsub = gossipsub::Behaviour::new_with_metrics( - MessageAuthenticity::Signed(peer_key.clone()), + MessageAuthenticity::Signed(peer_key), gossipsub, constant::METRIC_REGISTRY .try_lock() @@ -154,7 +155,7 @@ impl NetworkBehaviour for Behaviour { cx: &mut std::task::Context<'_>, params: &mut impl libp2p::swarm::PollParameters, ) -> Poll>> { - if let Poll::Ready(_) = self.tick.poll_tick(cx) { + if self.tick.poll_tick(cx).is_ready() { // Publish batch if !self.echo_queue.is_empty() { let mut echos = Batch { data: Vec::new() }; @@ -166,10 +167,14 @@ impl NetworkBehaviour for Behaviour { } } + debug!("Publishing {} echos", echos.data.len()); if let Ok(msg) = bincode::serialize::(&echos) { P2P_GOSSIP_BATCH_SIZE.observe(echos.data.len() as f64); - _ = self.gossipsub.publish(IdentTopic::new(TOPOS_ECHO), msg); + match self.gossipsub.publish(IdentTopic::new(TOPOS_ECHO), msg) { + Ok(message_id) => debug!("Published echo {}", message_id), + Err(error) => error!("Failed to publish echo: {}", error), + } } else { P2P_MESSAGE_SERIALIZE_FAILURE_TOTAL .with_label_values(&["echo"]) @@ -187,6 +192,7 @@ impl NetworkBehaviour for Behaviour { } } + debug!("Publishing {} readies", readies.data.len()); if let Ok(msg) = bincode::serialize::(&readies) { P2P_GOSSIP_BATCH_SIZE.observe(readies.data.len() as f64); _ = self.gossipsub.publish(IdentTopic::new(TOPOS_READY), msg); @@ -200,13 +206,7 @@ impl NetworkBehaviour for Behaviour { let event = match self.gossipsub.poll(cx, params) { Poll::Pending => return Poll::Pending, - Poll::Ready(ToSwarm::ListenOn { opts }) => { - return Poll::Ready(ToSwarm::ListenOn { opts }) - } - Poll::Ready(ToSwarm::RemoveListener { id }) => { - return Poll::Ready(ToSwarm::RemoveListener { id }) - } - Poll::Ready(ToSwarm::GenerateEvent(event)) => event, + Poll::Ready(ToSwarm::GenerateEvent(event)) => Some(event), Poll::Ready(ToSwarm::ListenOn { opts }) => { return Poll::Ready(ToSwarm::ListenOn { opts }) } @@ -245,13 +245,13 @@ impl NetworkBehaviour for Behaviour { } }; - if let gossipsub::Event::Message { ref message_id, .. } = event { + if let Some(gossipsub::Event::Message { ref message_id, .. }) = event { if self.cache.contains(message_id) { P2P_DUPLICATE_MESSAGE_ID_RECEIVED_TOTAL.inc(); } } - if let gossipsub::Event::Message { + if let Some(gossipsub::Event::Message { propagation_source, message_id, message: @@ -261,7 +261,7 @@ impl NetworkBehaviour for Behaviour { sequence_number, topic, }, - } = event + }) = event { match topic.as_str() { TOPOS_GOSSIP => { diff --git a/crates/topos-p2p/src/config.rs b/crates/topos-p2p/src/config.rs index 1906d4e3a..a223e8676 100644 --- a/crates/topos-p2p/src/config.rs +++ b/crates/topos-p2p/src/config.rs @@ -45,3 +45,11 @@ impl Default for DiscoveryConfig { } } } + +impl DiscoveryConfig { + pub fn with_replication_factor(mut self, replication_factor: NonZeroUsize) -> Self { + self.replication_factor = replication_factor; + + self + } +} diff --git a/crates/topos-p2p/src/constant.rs b/crates/topos-p2p/src/constant.rs index 7276c2125..48212fb42 100644 --- a/crates/topos-p2p/src/constant.rs +++ b/crates/topos-p2p/src/constant.rs @@ -26,3 +26,4 @@ lazy_static! { pub const TRANSMISSION_PROTOCOL: &str = "/tce-transmission/1"; pub const DISCOVERY_PROTOCOL: &str = "/tce-disco/1"; +pub const PEER_INFO_PROTOCOL: &str = "/tce-peer-info/1"; diff --git a/crates/topos-p2p/src/network.rs b/crates/topos-p2p/src/network.rs index 539797849..891b20a47 100644 --- a/crates/topos-p2p/src/network.rs +++ b/crates/topos-p2p/src/network.rs @@ -4,9 +4,10 @@ use crate::{ discovery::DiscoveryBehaviour, gossip, peer_info::PeerInfoBehaviour, transmission::TransmissionBehaviour, }, - config::NetworkConfig, + config::{DiscoveryConfig, NetworkConfig}, constant::{ - COMMAND_STREAM_BUFFER_SIZE, DISCOVERY_PROTOCOL, EVENT_STREAM_BUFFER, TRANSMISSION_PROTOCOL, + COMMAND_STREAM_BUFFER_SIZE, DISCOVERY_PROTOCOL, EVENT_STREAM_BUFFER, PEER_INFO_PROTOCOL, + TRANSMISSION_PROTOCOL, }, error::P2PError, TOPOS_ECHO, TOPOS_GOSSIP, TOPOS_READY, @@ -53,6 +54,12 @@ pub struct NetworkBuilder<'a> { } impl<'a> NetworkBuilder<'a> { + pub fn discovery_config(mut self, config: DiscoveryConfig) -> Self { + self.config.discovery = config; + + self + } + pub fn publish_retry(mut self, retry: usize) -> Self { self.config.publish_retry = retry; @@ -123,10 +130,7 @@ impl<'a> NetworkBuilder<'a> { let gossipsub = gossip::Behaviour::new(peer_key.clone()); let behaviour = Behaviour { gossipsub, - peer_info: PeerInfoBehaviour::new( - self.transmission_protocol.unwrap_or(TRANSMISSION_PROTOCOL), - &peer_key, - ), + peer_info: PeerInfoBehaviour::new(PEER_INFO_PROTOCOL, &peer_key), discovery: DiscoveryBehaviour::create( &self.config.discovery, diff --git a/crates/topos-p2p/src/runtime/handle_command.rs b/crates/topos-p2p/src/runtime/handle_command.rs index 18328411e..43f6847bf 100644 --- a/crates/topos-p2p/src/runtime/handle_command.rs +++ b/crates/topos-p2p/src/runtime/handle_command.rs @@ -141,7 +141,7 @@ impl Runtime { Command::Gossip { topic, data } => { match self.swarm.behaviour_mut().gossipsub.publish(topic, data) { Ok(message_id) => { - info!("Published message {message_id:?} to {topic}"); + debug!("Published message to {topic}"); P2P_MESSAGE_SENT_ON_GOSSIPSUB_TOTAL.inc(); } Err(err) => error!("Failed to publish message to {topic}: {err}"), diff --git a/crates/topos-p2p/src/runtime/handle_event.rs b/crates/topos-p2p/src/runtime/handle_event.rs index 8f0196ae4..4ec582ec4 100644 --- a/crates/topos-p2p/src/runtime/handle_event.rs +++ b/crates/topos-p2p/src/runtime/handle_event.rs @@ -71,7 +71,7 @@ impl } => { info!( "Local node is listening on {:?}", - address.with(Protocol::P2p(self.local_peer_id.into())), + address.with(Protocol::P2p(self.local_peer_id)), ); self.active_listeners.insert(listener_id); diff --git a/crates/topos-p2p/src/runtime/handle_event/peer_info.rs b/crates/topos-p2p/src/runtime/handle_event/peer_info.rs index 97f3b1562..ad6bfdb2f 100644 --- a/crates/topos-p2p/src/runtime/handle_event/peer_info.rs +++ b/crates/topos-p2p/src/runtime/handle_event/peer_info.rs @@ -4,8 +4,7 @@ use libp2p::identify::{Event as IdentifyEvent, Info as IdentifyInfo}; use tracing::info; use crate::{ - behaviour::transmission::protocol::TransmissionProtocol, constant::TRANSMISSION_PROTOCOL, - Runtime, + behaviour::transmission::protocol::TransmissionProtocol, constant::PEER_INFO_PROTOCOL, Runtime, }; use super::EventHandler; @@ -18,20 +17,15 @@ impl EventHandler> for Runtime { protocol_version, listen_addrs, protocols, + observed_addr, .. } = info; + // TODO: Check subprotocols for TransmissionProtocol if !self.peer_set.contains(&peer_id) - && protocol_version.as_bytes() == TRANSMISSION_PROTOCOL.as_bytes() - && protocols.iter().any(|p| { - self.swarm - .behaviour() - .discovery - .inner - .protocol_names() - .contains(&Cow::Borrowed(p)) - }) + && protocol_version.as_bytes() == PEER_INFO_PROTOCOL.as_bytes() { + self.swarm.add_external_address(observed_addr); self.peer_set.insert(peer_id); for addr in listen_addrs { info!( diff --git a/crates/topos-p2p/src/runtime/mod.rs b/crates/topos-p2p/src/runtime/mod.rs index 2a0c2818d..ac00cf831 100644 --- a/crates/topos-p2p/src/runtime/mod.rs +++ b/crates/topos-p2p/src/runtime/mod.rs @@ -129,6 +129,10 @@ impl Runtime { if addr_query_id.is_none() && self.peer_set.len() >= self.config.minimum_cluster_size { + warn!( + "Publishing our addresses to the network ! We have {} peers", + self.peer_set.len() + ); let key = Key::new(&self.local_peer_id.to_string()); addr_query_id = if let Ok(query_id_record) = self.swarm.behaviour_mut().discovery.inner.put_record( @@ -165,11 +169,17 @@ impl Runtime { KademliaEvent::OutboundQueryProgressed { id, result: - QueryResult::PutRecord(Err(PutRecordError::QuorumFailed { .. })), - .. + QueryResult::PutRecord(Err(PutRecordError::QuorumFailed { + key, + success, + quorum, + })), + stats, + step, } if Some(id) == addr_query_id && publish_retry > 0 => { publish_retry -= 1; warn!("Failed to PutRecord in DHT, retry again, attempt number {publish_retry}"); + warn!("QuorumFailure on DHT addr publication: key: {key:?}, success: {success:?}, quorum: {quorum:?}, stats: {stats:?}"); let key = Key::new(&self.local_peer_id.to_string()); if let Ok(query_id_record) = self.swarm.behaviour_mut().discovery.inner.put_record( @@ -201,7 +211,14 @@ impl Runtime { .. } => {} - KademliaEvent::OutboundQueryProgressed { .. } => {} + KademliaEvent::OutboundQueryProgressed { + id, + result, + stats, + step, + } => { + println!("OutboundQueryProgressed: {id:?}, {result:?}, {stats:?}, {step:?}"); + } KademliaEvent::InboundRequest { .. } => {} KademliaEvent::RoutingUpdated { .. } => {} diff --git a/crates/topos-p2p/src/tests/dht.rs b/crates/topos-p2p/src/tests/dht.rs index 9bd6be06e..93be2c537 100644 --- a/crates/topos-p2p/src/tests/dht.rs +++ b/crates/topos-p2p/src/tests/dht.rs @@ -1,7 +1,8 @@ -use std::time::Duration; +use std::{num::NonZeroUsize, time::Duration}; use futures::StreamExt; use libp2p::{ + identify::{self, Info}, kad::{record::Key, KademliaEvent, PutRecordOk, QueryResult, Record}, swarm::SwarmEvent, }; @@ -10,15 +11,15 @@ use test_log::test; use topos_test_sdk::tce::NodeConfig; use crate::{ - event::ComposedEvent, network::NetworkBuilder, tests::support::local_peer, wait_for_event, - Client, Runtime, + config::DiscoveryConfig, event::ComposedEvent, network::NetworkBuilder, + tests::support::local_peer, wait_for_event, Client, Runtime, }; use super::support::{dummy_peer, PeerAddr}; #[rstest] #[test(tokio::test)] -#[timeout(Duration::from_secs(1))] +#[timeout(Duration::from_secs(5))] async fn put_value_in_dht() { let peer_1 = NodeConfig::from_seed(1); let peer_2 = NodeConfig::from_seed(2); @@ -27,19 +28,43 @@ async fn put_value_in_dht() { let (_, _, runtime) = crate::network::builder() .peer_key(peer_2.keypair.clone()) - .known_peers(&[]) + .known_peers(&[(peer_1.peer_id(), peer_1.addr.clone())]) .exposed_addresses(peer_2.addr.clone()) .listen_addr(peer_2.addr.clone()) - .minimum_cluster_size(0) + .minimum_cluster_size(1) + .discovery_config( + DiscoveryConfig::default().with_replication_factor(NonZeroUsize::new(1).unwrap()), + ) .build() .await .expect("Unable to create p2p network"); let mut runtime = runtime.bootstrap().await.unwrap(); + // runtime + // .swarm + // .behaviour_mut() + // .discovery + // .inner + // .add_address(&peer_1.keypair.public().to_peer_id(), peer_1.addr); + + // loop { + // if let Some(SwarmEvent::Behaviour(ComposedEvent::PeerInfo(event))) = + // runtime.swarm.next().await + // { + // println!("Event: {:?}", event); + // if let identify::Event::Received { + // peer_id, + // info: Info { observed_addr, .. }, + // } = *event + // { + // println!("Peer {} observed address: {}", peer_id, observed_addr); + // runtime.swarm.add_external_address(observed_addr); + // break; + // } + // } + // } let kad = &mut runtime.swarm.behaviour_mut().discovery; - kad.inner - .add_address(&peer_1.keypair.public().to_peer_id(), peer_1.addr); let input_key = Key::new(&runtime.local_peer_id.to_string()); _ = kad diff --git a/crates/topos-p2p/src/tests/support/macros.rs b/crates/topos-p2p/src/tests/support/macros.rs index 69233d31c..16f9c7de8 100644 --- a/crates/topos-p2p/src/tests/support/macros.rs +++ b/crates/topos-p2p/src/tests/support/macros.rs @@ -3,6 +3,7 @@ macro_rules! wait_for_event { ($node:ident, matches: $(|)? $( $pattern:pat_param )|+ $( if $guard: expr )? $(,)?) => { let assertion = async { while let Some(event) = $node.next().await { + println!("Event: {:?}", event); if matches!(event, $( $pattern )|+ $( if $guard )?) { break; } diff --git a/crates/topos-tce-broadcast/src/double_echo/mod.rs b/crates/topos-tce-broadcast/src/double_echo/mod.rs index 871fecbb8..534c0dd66 100644 --- a/crates/topos-tce-broadcast/src/double_echo/mod.rs +++ b/crates/topos-tce-broadcast/src/double_echo/mod.rs @@ -1,4 +1,4 @@ -use crate::{constant, Errors}; +use crate::constant; use crate::{sampler::SampleType, DoubleEchoCommand, SubscriptionsView}; use opentelemetry::trace::TraceContextExt; use std::collections::HashSet; @@ -16,8 +16,9 @@ use topos_metrics::{ DOUBLE_ECHO_BUFFER_CAPACITY_TOTAL, DOUBLE_ECHO_CURRENT_BUFFER_SIZE, }; use topos_p2p::PeerId; -use topos_tce_storage::PendingCertificateId; -use tracing::{debug, error, info, info_span, instrument, trace, warn, warn_span, Span}; +#[cfg(not(feature = "direct"))] +use tracing::error; +use tracing::{debug, info, info_span, trace, warn, warn_span, Span}; use tracing_opentelemetry::OpenTelemetrySpanExt; /// Processing data associated to a Certificate candidate for delivery @@ -261,6 +262,140 @@ impl DoubleEcho { } } +#[cfg(not(feature = "direct"))] +impl DoubleEcho { + /// Called to process potentially new certificate: + /// - either submitted from API ( [tce_transport::TceCommands::Broadcast] command) + /// - or received through the gossip (first step of protocol exchange) + pub(crate) fn broadcast(&mut self, cert: Certificate, origin: bool) { + info!("🙌 Starting broadcasting the Certificate {}", &cert.id); + + if self.cert_pre_broadcast_check(&cert).is_err() { + error!("Failure on the pre-check for the Certificate {}", &cert.id); + self.event_sender + .send(ProtocolEvents::BroadcastFailed { + certificate_id: cert.id, + }) + .unwrap(); + return; + } + // Don't gossip one cert already gossiped + if self.cert_candidate.contains_key(&cert.id) { + self.event_sender + .send(ProtocolEvents::BroadcastFailed { + certificate_id: cert.id, + }) + .unwrap(); + return; + } + + if self.delivered_certificates.get(&cert.id).is_some() { + self.event_sender + .send(ProtocolEvents::AlreadyDelivered { + certificate_id: cert.id, + }) + .unwrap(); + + return; + } + + let span = self + .span_tracker + .get(&cert.id) + .cloned() + .unwrap_or_else(Span::current); + + if origin { + warn!("📣 Gossiping the Certificate {}", &cert.id); + let _ = self.event_sender.send(ProtocolEvents::Gossip { + cert: cert.clone(), + ctx: span, + }); + } + + // Trigger event of new certificate candidate for delivery + self.start_broadcast(cert); + } + + fn start_broadcast(&mut self, cert: Certificate) { + // To include tracing context in client requests from _this_ app, + // use `context` to extract the current OpenTelemetry context. + // Add new entry for the new Cert candidate + match self.delivery_state_for_new_cert(&cert.id) { + Some(delivery_state) => { + self.cert_candidate + .insert(cert.id, (cert.clone(), delivery_state)); + + _ = self.event_sender.send(ProtocolEvents::Broadcast { + certificate_id: cert.id, + }); + } + None => { + error!("Ill-formed samples"); + let _ = self.event_sender.send(ProtocolEvents::Die); + return; + } + } + self.delivery_time + .insert(cert.id, (time::SystemTime::now(), Default::default())); + + let ctx = self + .span_tracker + .get(&cert.id) + .cloned() + .unwrap_or_else(Span::current); + + let _ = self.event_sender.send(ProtocolEvents::Echo { + certificate_id: cert.id, + ctx, + }); + } + + /// Build initial delivery state + fn delivery_state_for_new_cert( + &mut self, + certificate_id: &CertificateId, + ) -> Option { + let subscriptions = self.subscriptions.clone(); + + // Check whether inbound sets are empty + if subscriptions.echo.is_empty() || subscriptions.ready.is_empty() { + error!( + "One Subscription sample is empty: Echo({}), Ready({})", + subscriptions.echo.is_empty(), + subscriptions.ready.is_empty(), + ); + None + } else { + let ctx = self + .span_tracker + .get(certificate_id) + .cloned() + .unwrap_or_else(Span::current); + + Some(DeliveryState { + subscriptions, + ready_sent: false, + delivered: false, + ctx, + }) + } + } + + /// Checks done before starting to broadcast + fn cert_pre_broadcast_check(&self, cert: &Certificate) -> Result<(), ()> { + if cert.check_signature().is_err() { + error!("Error on the signature"); + } + + if cert.check_proof().is_err() { + error!("Error on the proof"); + } + + Ok(()) + } +} + impl DoubleEcho { fn sample_consume_peer(from_peer: &PeerId, state: &mut DeliveryState, sample_type: SampleType) { match sample_type { @@ -293,7 +428,6 @@ impl DoubleEcho { self.known_certificates.insert(cert.id); span.in_scope(|| { - info!("Certificate {} added to pending storage", cert.id); debug!("DoubleEchoCommand::Broadcast certificate_id: {}", cert.id); if self.buffer.len() < *constant::TOPOS_DOUBLE_ECHO_MAX_BUFFER_SIZE { self.buffer.push_back((need_gossip, cert)); @@ -318,7 +452,6 @@ impl DoubleEcho { if !cert_delivered { if self.known_certificates.get(&certificate_id).is_some() { let span = if let Some(root) = self.span_tracker.get(&certificate_id) { - info!("DEBUG::Receive ECHO with root"); info_span!( parent: root, "RECV Inbound Echo", @@ -326,7 +459,6 @@ impl DoubleEcho { certificate_id = certificate_id.to_string() ) } else { - info!("DEBUG::Receive ECHO without root"); info_span!( "RECV Inbound Echo", peer = self.local_peer_id, @@ -345,7 +477,6 @@ impl DoubleEcho { drop(_enter); // need to deliver the certificate } else if self.delivered_certificates.get(&certificate_id).is_none() { - info!("DEBUG::Receive ECHO BUFFERING"); // need to buffer the Echo self.buffered_messages .entry(certificate_id) @@ -422,127 +553,6 @@ impl DoubleEcho { } } - /// Called to process potentially new certificate: - /// - either submitted from API ( [tce_transport::TceCommands::Broadcast] command) - /// - or received through the gossip (first step of protocol exchange) - #[instrument(skip_all)] - pub(crate) fn broadcast(&mut self, cert: Certificate, origin: bool) { - info!("🙌 Starting broadcasting the Certificate {}", &cert.id); - - if self.cert_pre_broadcast_check(&cert).is_err() { - error!("Failure on the pre-check for the Certificate {}", &cert.id); - self.event_sender - .send(ProtocolEvents::BroadcastFailed { - certificate_id: cert.id, - }) - .unwrap(); - return; - } - // Don't gossip one cert already gossiped - if self.cert_candidate.contains_key(&cert.id) { - self.event_sender - .send(ProtocolEvents::BroadcastFailed { - certificate_id: cert.id, - }) - .unwrap(); - return; - } - - if self.delivered_certificates.get(&cert.id).is_some() { - self.event_sender - .send(ProtocolEvents::AlreadyDelivered { - certificate_id: cert.id, - }) - .unwrap(); - - return; - } - - let span = self - .span_tracker - .get(&cert.id) - .cloned() - .unwrap_or_else(Span::current); - - if origin { - warn!("📣 Gossiping the Certificate {}", &cert.id); - let _ = self.event_sender.send(ProtocolEvents::Gossip { - cert: cert.clone(), - ctx: span, - }); - } - - // Trigger event of new certificate candidate for delivery - self.start_broadcast(cert); - } - - fn start_broadcast(&mut self, cert: Certificate) { - // To include tracing context in client requests from _this_ app, - // use `context` to extract the current OpenTelemetry context. - // Add new entry for the new Cert candidate - match self.delivery_state_for_new_cert(&cert.id) { - Some(delivery_state) => { - info!("DeliveryState is : {:?}", delivery_state.subscriptions); - - self.cert_candidate - .insert(cert.id, (cert.clone(), delivery_state)); - - _ = self.event_sender.send(ProtocolEvents::Broadcast { - certificate_id: cert.id, - }); - } - None => { - error!("Ill-formed samples"); - let _ = self.event_sender.send(ProtocolEvents::Die); - return; - } - } - self.delivery_time - .insert(cert.id, (time::SystemTime::now(), Default::default())); - - let ctx = self - .span_tracker - .get(&cert.id) - .cloned() - .unwrap_or_else(Span::current); - - let _ = self.event_sender.send(ProtocolEvents::Echo { - certificate_id: cert.id, - ctx, - }); - } - - /// Build initial delivery state - fn delivery_state_for_new_cert( - &mut self, - certificate_id: &CertificateId, - ) -> Option { - let subscriptions = self.subscriptions.clone(); - - // Check whether inbound sets are empty - if subscriptions.echo.is_empty() || subscriptions.ready.is_empty() { - error!( - "One Subscription sample is empty: Echo({}), Ready({})", - subscriptions.echo.is_empty(), - subscriptions.ready.is_empty(), - ); - None - } else { - let ctx = self - .span_tracker - .get(certificate_id) - .cloned() - .unwrap_or_else(Span::current); - - Some(DeliveryState { - subscriptions, - ready_sent: false, - delivered: false, - ctx, - }) - } - } - pub(crate) fn state_change_follow_up(&mut self) { debug!("StateChangeFollowUp called"); let mut state_modified = false; @@ -550,7 +560,7 @@ impl DoubleEcho { let mut delivered_certificates = Vec::<(Certificate, Span)>::new(); // For all current Cert on processing - for (_certificate_id, (certificate, state_to_delivery)) in &mut self.cert_candidate { + for (certificate, state_to_delivery) in self.cert_candidate.values_mut() { // Check whether we should send Ready if !state_to_delivery.ready_sent && is_r_ready( @@ -644,19 +654,6 @@ impl DoubleEcho { let _ = self.event_sender.send(evt); } } - - /// Checks done before starting to broadcast - fn cert_pre_broadcast_check(&self, cert: &Certificate) -> Result<(), ()> { - if cert.check_signature().is_err() { - error!("Error on the signature"); - } - - if cert.check_proof().is_err() { - error!("Error on the proof"); - } - - Ok(()) - } } /// Predicate on whether we reached the threshold to deliver the Certificate diff --git a/crates/topos-tce-broadcast/src/lib.rs b/crates/topos-tce-broadcast/src/lib.rs index eeb3c5ab7..77a2cee1c 100644 --- a/crates/topos-tce-broadcast/src/lib.rs +++ b/crates/topos-tce-broadcast/src/lib.rs @@ -24,16 +24,13 @@ use topos_tce_storage::StorageClient; use tracing::{debug, error, info, Span}; use crate::sampler::SubscriptionsView; -use crate::tce_store::TceStore; pub use topos_core::uci; pub type Peer = String; mod constant; pub mod double_echo; -pub mod mem_store; pub mod sampler; -pub mod tce_store; #[cfg(test)] mod tests; @@ -198,7 +195,7 @@ impl ReliableBroadcastClient { #[derive(Error, Debug)] pub enum Errors { #[error("Error while sending a DoubleEchoCommand to DoubleEcho: {0:?}")] - DoubleEchoSend(#[from] mpsc::error::SendError), + DoubleEchoSend(#[from] Box>), #[error("Error while waiting for a DoubleEchoCommand response: {0:?}")] DoubleEchoRecv(#[from] oneshot::error::RecvError), diff --git a/crates/topos-tce-broadcast/src/mem_store.rs b/crates/topos-tce-broadcast/src/mem_store.rs deleted file mode 100644 index 2bfa7fe92..000000000 --- a/crates/topos-tce-broadcast/src/mem_store.rs +++ /dev/null @@ -1,110 +0,0 @@ -use crate::{Errors, TceStore}; -use std::collections::{BTreeSet, HashMap}; -use topos_core::uci::{Certificate, CertificateId, SubnetId, CERTIFICATE_ID_LENGTH}; - -/// Store implementation in RAM good enough for functional tests -/// Might need to split through a new layer of TCEState -/// between ReliableBroadcast and rocksdb -#[derive(Default, Clone)] -pub struct TceMemStore { - /// Global map of delivered and accepted certificates - all_certs: HashMap, - /// Mapping SubnetId -> Delivered certificated - history: HashMap>, - /// Consider for now that each TCE nodes is following all subnets - tracked_digest: HashMap>, - /// List of the subnets that we're part of - /// NOTE: Below is for later, for now we're considering - /// being part of all subnets, so following digest of everyone - followed_subnet: Vec, -} - -impl TceMemStore { - pub fn new(subnets: Vec) -> TceMemStore { - let mut store = TceMemStore { - all_certs: Default::default(), - history: Default::default(), - tracked_digest: Default::default(), - followed_subnet: subnets, - }; - for subnet in &store.followed_subnet { - store.tracked_digest.insert(*subnet, BTreeSet::new()); - store.history.insert(*subnet, BTreeSet::new()); - } - // Add the genesis - store.all_certs.insert( - CertificateId::from_array([0u8; CERTIFICATE_ID_LENGTH]), - Default::default(), - ); - store - } -} - -impl TceStore for TceMemStore { - // OTLP START DELIVERY TRACE [ cert, peer ] - fn apply_cert(&mut self, cert: &Certificate) -> Result<(), Errors> { - // Add the entry in the history - let _ = self.add_cert_in_hist(&cert.source_subnet_id, cert); - - // Add the cert into the history of each Target - for target_subnet_id in &cert.target_subnets { - self.add_cert_in_hist(target_subnet_id, cert); - self.add_cert_in_digest(target_subnet_id, &cert.id); - } - - Ok(()) - } - - fn add_cert_in_hist(&mut self, subnet_id: &SubnetId, cert: &Certificate) -> bool { - self.all_certs.insert(cert.id, cert.clone()); - self.history.entry(*subnet_id).or_default().insert(cert.id) - } - - fn add_cert_in_digest(&mut self, subnet_id: &SubnetId, cert_id: &CertificateId) -> bool { - self.tracked_digest - .entry(*subnet_id) - .or_default() - .insert(*cert_id) - } - - fn read_journal( - &self, - _subnet_id: SubnetId, - _from_offset: u64, - _max_results: u64, - ) -> Result<(Vec, u64), Errors> { - unimplemented!(); - } - - fn recent_certificates_for_subnet( - &self, - subnet_id: &SubnetId, - _last_n: u64, - ) -> Vec { - match self.history.get(subnet_id) { - Some(subnet_certs) => subnet_certs.iter().cloned().collect::>(), - None => Vec::new(), - } - } - - fn cert_by_id(&self, cert_id: &CertificateId) -> Result { - match self.all_certs.get(cert_id) { - Some(cert) => Ok(cert.clone()), - _ => Err(Errors::CertificateNotFound), - } - } - - fn check_precedence(&self, cert: &Certificate) -> Result<(), Errors> { - if cert.prev_id.as_array() == &[0u8; CERTIFICATE_ID_LENGTH] { - return Ok(()); - } - match self.cert_by_id(&cert.prev_id) { - Ok(_) => Ok(()), - _ => Err(Errors::CertificateNotFound), - } - } - - fn clone_box(&self) -> Box { - Box::new(self.clone()) - } -} diff --git a/crates/topos-tce-broadcast/src/tce_store.rs b/crates/topos-tce-broadcast/src/tce_store.rs deleted file mode 100644 index 97e7d06af..000000000 --- a/crates/topos-tce-broadcast/src/tce_store.rs +++ /dev/null @@ -1,44 +0,0 @@ -//! -//! Storage interface required to support TCE -//! -use topos_core::uci::{Certificate, CertificateId, SubnetId}; - -use crate::Errors; - -/// Defines abstract storage suitable for protocol handler. -/// -/// Implemented in node/store. -pub trait TceStore { - /// Saves (or replaces) the certificate - fn apply_cert(&mut self, cert: &Certificate) -> Result<(), Errors>; - - /// Saves journal item in history - fn add_cert_in_hist(&mut self, subnet_id: &SubnetId, cert_id: &Certificate) -> bool; - - /// Saves journal item in digest - fn add_cert_in_digest(&mut self, subnet_id: &SubnetId, cert_id: &CertificateId) -> bool; - - /// Reads journal entries - from old to new, paged - /// Returns tuple (data, last offset) - fn read_journal( - &self, - subnet_id: SubnetId, - from_offset: u64, - max_results: u64, - ) -> Result<(Vec, u64), Errors>; - - /// Easy access - fn recent_certificates_for_subnet( - &self, - subnet_id: &SubnetId, - last_n: u64, - ) -> Vec; - - /// Read certificate - fn cert_by_id(&self, cert_id: &CertificateId) -> Result; - - /// Check on the previous cert - fn check_precedence(&self, cert: &Certificate) -> Result<(), Errors>; - - fn clone_box(&self) -> Box; -} diff --git a/crates/topos-tce-broadcast/src/tests/mod.rs b/crates/topos-tce-broadcast/src/tests/mod.rs index 969944c41..0c24adf4d 100644 --- a/crates/topos-tce-broadcast/src/tests/mod.rs +++ b/crates/topos-tce-broadcast/src/tests/mod.rs @@ -4,10 +4,13 @@ use rstest::*; use std::collections::HashSet; use std::usize; use tce_transport::ReliableBroadcastParams; + +#[cfg(not(feature = "direct"))] use tokio::sync::broadcast::error::TryRecvError; use tokio::sync::broadcast::Receiver; use tokio::time::Duration; +#[cfg(not(feature = "direct"))] use topos_test_sdk::constants::*; const CHANNEL_SIZE: usize = 10; @@ -94,6 +97,7 @@ fn create_context(params: TceParams) -> (DoubleEcho, Context) { ) } +#[cfg(not(feature = "direct"))] fn reach_echo_threshold(double_echo: &mut DoubleEcho, cert: &Certificate) { let selected = double_echo .subscriptions @@ -108,6 +112,7 @@ fn reach_echo_threshold(double_echo: &mut DoubleEcho, cert: &Certificate) { } } +#[cfg(not(feature = "direct"))] fn reach_ready_threshold(double_echo: &mut DoubleEcho, cert: &Certificate) { let selected = double_echo .subscriptions @@ -122,6 +127,7 @@ fn reach_ready_threshold(double_echo: &mut DoubleEcho, cert: &Certificate) { } } +#[cfg(not(feature = "direct"))] fn reach_delivery_threshold(double_echo: &mut DoubleEcho, cert: &Certificate) { let selected = double_echo .subscriptions @@ -141,6 +147,7 @@ fn reach_delivery_threshold(double_echo: &mut DoubleEcho, cert: &Certificate) { #[case(medium_config())] #[tokio::test] #[trace] +#[cfg(not(feature = "direct"))] async fn trigger_success_path_upon_reaching_threshold(#[case] params: TceParams) { let (mut double_echo, mut ctx) = create_context(params); @@ -205,6 +212,7 @@ async fn trigger_success_path_upon_reaching_threshold(#[case] params: TceParams) #[case(medium_config())] #[tokio::test] #[trace] +#[cfg(not(feature = "direct"))] async fn trigger_ready_when_reached_enough_ready(#[case] params: TceParams) { let (mut double_echo, mut ctx) = create_context(params); @@ -252,6 +260,7 @@ async fn trigger_ready_when_reached_enough_ready(#[case] params: TceParams) { #[case(medium_config())] #[tokio::test] #[trace] +#[cfg(not(feature = "direct"))] async fn process_after_delivery_until_sending_ready(#[case] params: TceParams) { let (mut double_echo, mut ctx) = create_context(params); diff --git a/crates/topos-tce-proxy/tests/tce_tests.rs b/crates/topos-tce-proxy/tests/tce_tests.rs index 33f2b09af..1696193dd 100644 --- a/crates/topos-tce-proxy/tests/tce_tests.rs +++ b/crates/topos-tce-proxy/tests/tce_tests.rs @@ -238,6 +238,7 @@ async fn test_tce_get_source_head_certificate( #[rstest] #[test(tokio::test)] +#[ignore = "Broken because of https://github.com/topos-protocol/topos/pull/248"] async fn test_tce_get_last_pending_certificates( #[future] start_node: TceContext, ) -> Result<(), Box> { diff --git a/crates/topos-tce-storage/src/client.rs b/crates/topos-tce-storage/src/client.rs index cc0c5a7b2..fd38da1c4 100644 --- a/crates/topos-tce-storage/src/client.rs +++ b/crates/topos-tce-storage/src/client.rs @@ -102,10 +102,14 @@ impl StorageClient { pub async fn certificate_delivered( &self, certificate_id: CertificateId, + certificate: Option, ) -> Result { - CertificateDelivered { certificate_id } - .send_to(&self.sender) - .await + CertificateDelivered { + certificate_id, + certificate, + } + .send_to(&self.sender) + .await } pub async fn fetch_certificates( diff --git a/crates/topos-tce-storage/src/command.rs b/crates/topos-tce-storage/src/command.rs index 189a728b4..15890ab48 100644 --- a/crates/topos-tce-storage/src/command.rs +++ b/crates/topos-tce-storage/src/command.rs @@ -64,6 +64,7 @@ impl Command for GetNextPendingCertificate { #[derive(Debug)] pub struct CertificateDelivered { pub(crate) certificate_id: CertificateId, + pub(crate) certificate: Option, } impl Command for CertificateDelivered { diff --git a/crates/topos-tce-storage/src/connection/handlers.rs b/crates/topos-tce-storage/src/connection/handlers.rs index 8821ecb48..7f704f9ce 100644 --- a/crates/topos-tce-storage/src/connection/handlers.rs +++ b/crates/topos-tce-storage/src/connection/handlers.rs @@ -135,12 +135,18 @@ where ) -> Result { let certificate_id = command.certificate_id; - let (pending_certificate_id, certificate) = - self.storage.get_pending_certificate(certificate_id).await?; + let (pending_certificate_id, certificate) = if let Some(certificate) = command.certificate { + (None, certificate) + } else { + self.storage + .get_pending_certificate(certificate_id) + .await + .map(|(id, cert)| (Some(id), cert))? + }; Ok(self .storage - .persist(&certificate, Some(pending_certificate_id)) + .persist(&certificate, pending_certificate_id) .await?) } } diff --git a/crates/topos-tce/src/app_context.rs b/crates/topos-tce/src/app_context.rs index 8babaac75..e1df61de5 100644 --- a/crates/topos-tce/src/app_context.rs +++ b/crates/topos-tce/src/app_context.rs @@ -4,7 +4,6 @@ //! use futures::{Stream, StreamExt}; use opentelemetry::trace::{FutureExt as TraceFutureExt, TraceContextExt}; -use serde::{Deserialize, Serialize}; use std::collections::HashMap; use tce_transport::{ProtocolEvents, TceCommands}; use tokio::spawn; @@ -27,6 +26,7 @@ use tracing::{debug, error, info, info_span, trace, warn, Instrument}; use tracing_opentelemetry::OpenTelemetrySpanExt; use crate::events::Events; +use crate::NetworkMessage; /// Top-level transducer main app context & driver (alike) /// @@ -274,10 +274,13 @@ impl AppContext { let storage = self.pending_storage.clone(); let api_client = self.api_client.clone(); + let certificate_id = certificate.id; spawn(async move { - match storage.certificate_delivered(certificate.id).await { + match storage + .certificate_delivered(certificate_id, Some(certificate.clone())) + .await + { Ok(positions) => { - let certificate_id = certificate.id; api_client .dispatch_certificate( certificate, @@ -307,7 +310,12 @@ impl AppContext { } Err(StorageError::InternalStorage( InternalStorageError::CertificateNotFound(_), - )) => {} + )) => { + error!( + "Certificate {} not found in pending storage", + certificate_id + ); + } Err(e) => { error!("Pending storage error while delivering certificate: {e}"); } @@ -329,6 +337,7 @@ impl AppContext { ctx: PropagationContext::inject(&span.context()), }); + info!("Sending Gossip for certificate {}", cert_id); if let Err(e) = self .network_client .publish::(topos_p2p::TOPOS_GOSSIP, data) @@ -511,37 +520,3 @@ impl AppContext { Ok(()) } } - -/// Definition of networking payload. -/// -/// We assume that only Commands will go through the network, -/// [Response] is used to allow reporting of logic errors to the caller. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[allow(clippy::large_enum_variant)] -enum NetworkMessage { - Cmd(TceCommands), - Bulk(Vec), - - NotReady(topos_p2p::NotReadyMessage), -} - -// deserializer -impl From> for NetworkMessage { - fn from(data: Vec) -> Self { - bincode::deserialize::(data.as_ref()).expect("msg deser") - } -} - -// serializer -impl From for Vec { - fn from(msg: NetworkMessage) -> Self { - bincode::serialize::(&msg).expect("msg ser") - } -} - -// transformer of protocol commands into network commands -impl From for NetworkMessage { - fn from(cmd: TceCommands) -> Self { - Self::Cmd(cmd) - } -} diff --git a/crates/topos-tce/src/lib.rs b/crates/topos-tce/src/lib.rs index f87f953cc..8e11d6d98 100644 --- a/crates/topos-tce/src/lib.rs +++ b/crates/topos-tce/src/lib.rs @@ -7,7 +7,8 @@ use std::time::Duration; pub use app_context::AppContext; use opentelemetry::global; -use tce_transport::ReliableBroadcastParams; +use serde::{Deserialize, Serialize}; +use tce_transport::{ReliableBroadcastParams, TceCommands}; use tokio::{spawn, sync::mpsc, sync::oneshot}; use topos_p2p::utils::local_key_pair_from_slice; use topos_p2p::{utils::local_key_pair, Multiaddr, PeerId}; @@ -155,3 +156,37 @@ pub async fn run( global::shutdown_tracer_provider(); Ok(()) } + +/// Definition of networking payload. +/// +/// We assume that only Commands will go through the network, +/// [Response] is used to allow reporting of logic errors to the caller. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[allow(clippy::large_enum_variant)] +pub enum NetworkMessage { + Cmd(TceCommands), + Bulk(Vec), + + NotReady(topos_p2p::NotReadyMessage), +} + +// deserializer +impl From> for NetworkMessage { + fn from(data: Vec) -> Self { + bincode::deserialize::(data.as_ref()).expect("msg deser") + } +} + +// serializer +impl From for Vec { + fn from(msg: NetworkMessage) -> Self { + bincode::serialize::(&msg).expect("msg ser") + } +} + +// transformer of protocol commands into network commands +impl From for NetworkMessage { + fn from(cmd: TceCommands) -> Self { + Self::Cmd(cmd) + } +} diff --git a/crates/topos-test-sdk/src/tce/mod.rs b/crates/topos-test-sdk/src/tce/mod.rs index da52ef105..d1d73a71d 100644 --- a/crates/topos-test-sdk/src/tce/mod.rs +++ b/crates/topos-test-sdk/src/tce/mod.rs @@ -70,12 +70,13 @@ impl TceContext { } } -#[derive(Clone)] +#[derive(Debug, Clone)] pub struct NodeConfig { pub seed: u8, pub port: u16, pub keypair: Keypair, pub addr: Multiaddr, + pub minimum_cluster_size: usize, } impl Default for NodeConfig { @@ -93,9 +94,14 @@ impl NodeConfig { port, keypair, addr, + minimum_cluster_size: 1, } } + pub fn peer_id(&self) -> PeerId { + self.keypair.public().to_peer_id() + } + pub async fn bootstrap( &self, peers: &[NodeConfig], @@ -107,14 +113,28 @@ impl NodeConfig { ), Box, > { - bootstrap_network(self.seed, self.port, self.addr.clone(), peers, 2).await + bootstrap_network( + self.seed, + self.port, + self.addr.clone(), + peers, + self.minimum_cluster_size, + ) + .await } pub async fn create( &self, peers: &[NodeConfig], ) -> Result<(Client, impl Stream, Runtime), P2PError> { - create_network_worker(self.seed, self.port, self.addr.clone(), peers, 2).await + create_network_worker( + self.seed, + self.port, + self.addr.clone(), + peers, + self.minimum_cluster_size, + ) + .await } } @@ -127,10 +147,15 @@ pub async fn start_node( let peer_id = config.keypair.public().to_peer_id(); let peer_id_str = peer_id.to_base58(); - let (network_client, network_stream, runtime_join_handle) = - bootstrap_network(config.seed, config.port, config.addr.clone(), peers, 1) - .await - .expect("Unable to bootstrap tce network"); + let (network_client, network_stream, runtime_join_handle) = bootstrap_network( + config.seed, + config.port, + config.addr.clone(), + peers, + config.minimum_cluster_size, + ) + .await + .expect("Unable to bootstrap tce network"); let (_, (storage, storage_client, storage_stream)) = create_rocksdb(&peer_id_str, certificates).await; @@ -190,7 +215,14 @@ pub async fn start_node( } fn build_peer_config_pool(peer_number: u8) -> Vec { - (1..=peer_number).map(NodeConfig::from_seed).collect() + (1..=peer_number) + .map(NodeConfig::from_seed) + .map(|mut c| { + c.minimum_cluster_size = peer_number as usize / 2; + + c + }) + .collect() } pub async fn start_pool(peer_number: u8) -> HashMap { @@ -221,6 +253,7 @@ pub async fn create_network(peer_number: usize) -> HashMap { let mut peers_context = start_pool(peer_number as u8).await; let all_peers: Vec = peers_context.keys().cloned().collect(); + warn!("Pool created, waiting for peers to connect..."); // Force TCE nodes to recreate subscriptions and subscribers let mut await_peers = Vec::new(); for (peer_id, client) in peers_context.iter_mut() { @@ -244,6 +277,7 @@ pub async fn create_network(peer_number: usize) -> HashMap { } assert!(!join_all(await_peers).await.iter().any(|res| res.is_err())); + warn!("Peers connected"); for (peer_id, client) in peers_context.iter_mut() { wait_for_event!( @@ -254,6 +288,8 @@ pub async fn create_network(peer_number: usize) -> HashMap { ); } + warn!("Stable sample received"); + // Waiting for new network view let mut await_peers = Vec::new(); for (_peer_id, client) in peers_context.iter_mut() { @@ -267,5 +303,6 @@ pub async fn create_network(peer_number: usize) -> HashMap { .map(|r: tonic::Response<_>| r.into_inner().has_active_sample)) .any(|r| r.is_err() || !r.unwrap())); + warn!("GRPC status received and ok"); peers_context } diff --git a/crates/topos-test-sdk/src/tce/p2p.rs b/crates/topos-test-sdk/src/tce/p2p.rs index 23a8c456c..014f28db7 100644 --- a/crates/topos-test-sdk/src/tce/p2p.rs +++ b/crates/topos-test-sdk/src/tce/p2p.rs @@ -64,6 +64,5 @@ pub async fn bootstrap_network( let runtime = runtime.bootstrap().await?; let runtime_join_handle = spawn(runtime.run()); - Ok((network_client, network_stream, runtime_join_handle)) } diff --git a/crates/topos/tests/cert_delivery.rs b/crates/topos/tests/cert_delivery.rs index b7e51b296..517a4a657 100644 --- a/crates/topos/tests/cert_delivery.rs +++ b/crates/topos/tests/cert_delivery.rs @@ -19,7 +19,7 @@ use topos_core::{ uci::{Certificate, SubnetId, SUBNET_ID_LENGTH}, }; use topos_test_sdk::{certificates::create_certificate_chains, tce::create_network}; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; const NUMBER_OF_SUBNETS_PER_CLIENT: usize = 1; @@ -37,7 +37,7 @@ fn get_subset_of_subnets(subnets: &[SubnetId], subset_size: usize) -> Vec = Vec::new(); @@ -56,7 +56,7 @@ async fn start_a_cluster() { #[rstest] #[tokio::test] -#[timeout(Duration::from_secs(60))] +#[timeout(Duration::from_secs(30))] // FIXME: This test is flaky, it fails sometimes because of gRPC connection error (StreamClosed) async fn cert_delivery() { let subscriber = tracing_subscriber::FmtSubscriber::builder() @@ -65,7 +65,7 @@ async fn cert_delivery() { .finish(); let _ = tracing::subscriber::set_global_default(subscriber); - let peer_number = 15; + let peer_number = 10; let number_of_certificates_per_subnet = 2; let number_of_subnets = 3; @@ -94,9 +94,12 @@ async fn cert_delivery() { } } } + + warn!("Starting the cluster..."); // List of peers (tce nodes) with their context let mut peers_context = create_network(peer_number).await; + warn!("Cluster started, starting clients..."); // Connected tce clients are passing received certificates to this mpsc::Receiver, collect all of them let mut clients_delivered_certificates: Vec> = Vec::new(); // (Peer Id, Subnet Id, Certificate) diff --git a/crates/topos/tests/push-certificate.rs b/crates/topos/tests/push-certificate.rs index 5a17824c5..bf8362280 100644 --- a/crates/topos/tests/push-certificate.rs +++ b/crates/topos/tests/push-certificate.rs @@ -4,6 +4,7 @@ use std::{thread, time::Duration}; use assert_cmd::Command; +use rstest::*; use topos_core::api::grpc::tce::v1::StatusRequest; use topos_test_sdk::tce::create_network; @@ -21,10 +22,12 @@ fn help_display() -> Result<(), Box> { Ok(()) } +#[rstest] #[test_log::test(tokio::test)] +#[timeout(Duration::from_secs(20))] // FIXME: This test is flaky, it fails sometimes because of sample failure async fn assert_delivery() -> Result<(), Box> { - let mut peers_context = create_network(10).await; + let mut peers_context = create_network(5).await; let mut status: Vec = Vec::new(); diff --git a/tools/config/prometheus/prometheus.yml b/tools/config/prometheus/prometheus.yml index dc8fa5571..f8d00d5de 100644 --- a/tools/config/prometheus/prometheus.yml +++ b/tools/config/prometheus/prometheus.yml @@ -36,13 +36,6 @@ scrape_configs: static_configs: - targets: ["localhost:9090"] - - job_name: "cadvisor" - - # Override the global default and scrape targets from this job every 5 seconds. - scrape_interval: 5s - - static_configs: - - targets: ["cadvisor:8080"] - job_name: "peers" @@ -67,3 +60,9 @@ scrape_configs: 'tools-peer-13:3000', 'tools-peer-14:3000', ] + metric_relabel_configs: + - source_labels: ["instance"] + regex: ".*((boot|peer)[^:]*).*" + target_label: "job" + replacement: "$1" + diff --git a/tools/dashboard.json b/tools/dashboard.json index a3179c052..dca8cfcd7 100644 --- a/tools/dashboard.json +++ b/tools/dashboard.json @@ -25,7 +25,1229 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 12, + "w": 4, + "x": 0, + "y": 0 + }, + "id": 25, + "options": { + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showThresholdLabels": false, + "showThresholdMarkers": true + }, + "pluginVersion": "10.0.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(topos_certificate_received_from_api_total{job=~\"$peer\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Certificate created", + "type": "gauge" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 12, + "w": 4, + "x": 4, + "y": 0 + }, + "id": 24, + "options": { + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showThresholdLabels": false, + "showThresholdMarkers": true + }, + "pluginVersion": "10.0.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(topos_certificate_delivered_total{job=~\"$peer\"})", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Certificate delivered", + "type": "gauge" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "max": 100, + "min": 0, + "thresholds": { + "mode": "percentage", + "steps": [ + { + "color": "dark-red", + "value": null + }, + { + "color": "orange", + "value": 80 + }, + { + "color": "green", + "value": 100 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 12, + "w": 4, + "x": 8, + "y": 0 + }, + "id": 26, + "options": { + "orientation": "auto", + "reduceOptions": { + "calcs": [], + "fields": "/^Delivery$/", + "values": false + }, + "showThresholdLabels": false, + "showThresholdMarkers": false + }, + "pluginVersion": "10.0.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(topos_certificate_received_from_api_total{job=~\"$peer\"})", + "hide": true, + "legendFormat": "__auto", + "range": true, + "refId": "A" + }, + { + "datasource": { + "name": "Expression", + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "100 * ($B ) / ($A * 15) ", + "hide": false, + "refId": "Delivery", + "type": "math" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(topos_certificate_delivered_total{job=~\"$peer\"})", + "hide": true, + "legendFormat": "__auto", + "range": true, + "refId": "B" + } + ], + "title": "Full Delivery", + "type": "gauge" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 12, + "w": 12, + "x": 12, + "y": 0 + }, + "id": 9, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "topos_certificate_received_from_api_total{job=~\"$peer\"}", + "hide": true, + "legendFormat": "API {{job}}", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "topos_certificate_received_from_gossip_total{job=~\"$peer\"}", + "hide": false, + "legendFormat": "Gossip {{job}}", + "range": true, + "refId": "B" + } + ], + "title": "Certificate reception", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 12 + }, + "id": 32, + "panels": [], + "title": "Misc", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "description": "Latency between rounds", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 13 + }, + "id": 30, + "interval": "4", + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_msg_recv_counts_total{hash=\"topos_gossip\", job=~\"$peer\"}[$__rate_interval]))", + "legendFormat": "Gossip received", + "range": true, + "refId": "Gossip received" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_msg_published_total{hash=\"topos_echo\", job=~\"$peer\"}[$__rate_interval]))", + "hide": false, + "legendFormat": "Echo published", + "range": true, + "refId": "Echo published" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_iwant_msgs_total{hash=\"topos_gossip\", job=~\"$peer\"}[$__rate_interval]))", + "hide": false, + "legendFormat": "IWANT Gossip", + "range": true, + "refId": "A" + } + ], + "title": "Gossip received vs. Echo published", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "description": "Measure the redundancy", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [ + { + "matcher": { + "id": "byFrameRefID", + "options": "Ratio" + }, + "properties": [ + { + "id": "custom.axisPlacement", + "value": "right" + }, + { + "id": "custom.scaleDistribution", + "value": { + "log": 10, + "type": "log" + } + }, + { + "id": "custom.lineStyle", + "value": { + "dash": [ + 0, + 10 + ], + "fill": "dot" + } + }, + { + "id": "custom.lineWidth", + "value": 4 + } + ] + } + ] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 13 + }, + "id": 29, + "interval": "4", + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(instance) (avg by(hash) (topos_libp2p_gossipsub_topic_msg_recv_counts_total{job=~\"$peer\"}))", + "hide": false, + "legendFormat": "Filtered", + "range": true, + "refId": "filtered" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(instance) (avg by(hash) (topos_libp2p_gossipsub_topic_msg_recv_counts_unfiltered_total{job=~\"$peer\"}))", + "hide": false, + "legendFormat": "Unfiltered", + "range": true, + "refId": "unfiltered" + }, + { + "datasource": { + "name": "Expression", + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "$unfiltered / $filtered", + "hide": false, + "refId": "Ratio", + "type": "math" + } + ], + "title": "Ratio payload recv unfiltered vs. filtered", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "description": "Latency between rounds", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 21 + }, + "id": 31, + "interval": "4", + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_msg_recv_counts_total{hash=\"topos_echo\", job=~\"$peer\"}[$__rate_interval]))", + "legendFormat": "Echo received", + "range": true, + "refId": "Gossip received" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_msg_published_total{hash=\"topos_ready\", job=~\"$peer\"}[$__rate_interval]))", + "hide": false, + "legendFormat": "Ready published", + "range": true, + "refId": "Echo published" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_iwant_msgs_total{hash=\"topos_echo\", job=~\"$peer\"}[$__rate_interval]))", + "hide": false, + "legendFormat": "IWANT Echo", + "range": true, + "refId": "A" + } + ], + "title": "Echo received vs. Ready published", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "continuous-GrYlRd" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 16, + "w": 12, + "x": 12, + "y": 21 + }, + "id": 27, + "options": { + "displayMode": "basic", + "minVizHeight": 10, + "minVizWidth": 0, + "orientation": "horizontal", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showUnfilled": true, + "valueMode": "color" + }, + "pluginVersion": "10.0.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "topos_certificate_delivered_total{job=~\"$peer\"}", + "legendFormat": "{{job}}", + "range": true, + "refId": "A" + } + ], + "title": "Certificate delivered per peer", + "type": "bargauge" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "description": "Latency between rounds", + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [ + { + "matcher": { + "id": "byFrameRefID", + "options": "Missing delivery" + }, + "properties": [ + { + "id": "custom.axisPlacement", + "value": "right" + }, + { + "id": "custom.drawStyle", + "value": "line" + }, + { + "id": "custom.lineStyle", + "value": { + "dash": [ + 0, + 10 + ], + "fill": "dot" + } + }, + { + "id": "custom.lineWidth", + "value": 3 + } + ] + } + ] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 29 + }, + "id": 33, + "interval": "4", + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_msg_recv_counts_total{hash=\"topos_ready\", job=~\"$peer\"}[$__rate_interval]))", + "legendFormat": "Ready received", + "range": true, + "refId": "Ready received" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_msg_published_total{hash=\"topos_ready\", job=~\"$peer\"}[$__rate_interval]))", + "hide": false, + "legendFormat": "Ready published", + "range": true, + "refId": "Ready published" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (increase(topos_libp2p_gossipsub_topic_iwant_msgs_total{hash=\"topos_ready\", job=~\"$peer\"}[$__rate_interval]))", + "hide": false, + "legendFormat": "IWANT Ready", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(topos_certificate_received_from_api_total{job=~\"$peer\"})", + "hide": true, + "legendFormat": "__auto", + "range": true, + "refId": "total_cert" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "builder", + "expr": "avg(topos_certificate_delivered_total)", + "hide": true, + "legendFormat": "__auto", + "range": true, + "refId": "cert_delivered" + }, + { + "datasource": { + "name": "Expression", + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "$total_cert - $cert_delivered", + "hide": false, + "refId": "Missing delivery", + "type": "math" + } + ], + "title": "Ready received vs. Ready published vs. Delivered certificate", + "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 37 + }, + "id": 19, + "panels": [], + "title": "P2P - Gossip protocol", + "type": "row" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green" + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 38 + }, + "id": 20, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "irate(topos_libp2p_gossipsub_topic_iwant_msgs_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "IWANT {{hash}} - {{job}}", + "range": true, + "refId": "A" + } + ], + "title": "IWANT msg", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "line", + "fillOpacity": 0, + "gradientMode": "none", + "hideFrom": { + "legend": false, + "tooltip": false, + "viz": false + }, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { + "type": "linear" + }, + "showPoints": "auto", + "spanNulls": false, + "stacking": { + "group": "A", + "mode": "none" + }, + "thresholdsStyle": { + "mode": "off" + } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green" + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 38 + }, + "id": 21, + "options": { + "legend": { + "calcs": [], + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "irate(topos_libp2p_gossipsub_topic_msg_recv_counts_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "{{hash}} {{job}}", + "range": true, + "refId": "A" + } + ], + "title": "MSG recv filtered", + "type": "timeseries" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green" + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 46 + }, + "id": 22, + "options": { + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showThresholdLabels": false, + "showThresholdMarkers": true + }, + "pluginVersion": "10.0.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum by(hash) (topos_libp2p_gossipsub_topic_msg_recv_counts_total{ job=~\"$peer\"})", + "legendFormat": "{{hash}} ", + "range": true, + "refId": "A" + } + ], + "title": "MSG recv total per topic", + "type": "gauge" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -37,27 +1259,17 @@ "axisColorMode": "text", "axisLabel": "", "axisPlacement": "auto", - "barAlignment": 0, - "drawStyle": "line", - "fillOpacity": 0, + "fillOpacity": 80, "gradientMode": "none", "hideFrom": { "legend": false, "tooltip": false, "viz": false }, - "lineInterpolation": "linear", "lineWidth": 1, - "pointSize": 5, "scaleDistribution": { "type": "linear" }, - "showPoints": "auto", - "spanNulls": false, - "stacking": { - "group": "A", - "mode": "none" - }, "thresholdsStyle": { "mode": "off" } @@ -67,8 +1279,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -80,67 +1291,70 @@ "overrides": [] }, "gridPos": { - "h": 12, + "h": 8, "w": 12, - "x": 0, - "y": 0 + "x": 12, + "y": 46 }, - "id": 9, + "id": 23, "options": { + "barRadius": 0, + "barWidth": 0.97, + "fullHighlight": false, + "groupWidth": 0.7, "legend": { "calcs": [], "displayMode": "list", "placement": "bottom", "showLegend": true }, + "orientation": "auto", + "showValue": "auto", + "stacking": "none", "tooltip": { "mode": "single", "sort": "none" - } + }, + "xTickLabelRotation": 0, + "xTickLabelSpacing": 0 }, + "pluginVersion": "9.5.3", "targets": [ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(certificate_received_from_api[$__interval])", - "legendFormat": "API {{pod}}", - "range": true, + "editorMode": "code", + "exemplar": false, + "expr": "max by(hash, job) (topos_libp2p_gossipsub_topic_msg_sent_counts_total{ job=~\"$peer\"})", + "format": "heatmap", + "instant": true, + "legendFormat": "{{job}} {{hash}}", + "range": false, "refId": "A" - }, - { - "datasource": { - "type": "prometheus", - "uid": "PBFA97CFB590B2093" - }, - "editorMode": "builder", - "expr": "irate(certificate_received_from_gossip[$__interval])", - "hide": false, - "legendFormat": "Gossip {{pod}}", - "range": true, - "refId": "B" } ], - "title": "Certificate reception", - "type": "timeseries" + "title": "MSG sent total per topic", + "type": "barchart" }, { + "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, - "y": 12 + "y": 54 }, "id": 15, + "panels": [], "title": "Storage layer", "type": "row" }, { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -182,8 +1396,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -198,7 +1411,7 @@ "h": 8, "w": 12, "x": 0, - "y": 13 + "y": 55 }, "id": 16, "options": { @@ -217,11 +1430,11 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(storage_command_channel_capacity[$__interval])", - "legendFormat": "{{pod}}", + "editorMode": "code", + "expr": "irate(topos_storage_command_channel_capacity_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "{{job}}", "range": true, "refId": "A" } @@ -232,7 +1445,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -244,8 +1457,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -260,7 +1472,7 @@ "h": 8, "w": 12, "x": 12, - "y": 13 + "y": 55 }, "id": 17, "options": { @@ -283,10 +1495,10 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(increase(storage_pending_certificate_existance_latency_bucket[$__range])) by (le)", + "expr": "sum(increase(topos_storage_pending_certificate_existance_latency_bucket{ job=~\"$peer\"}[$__range])) by (le)", "format": "heatmap", "legendFormat": "{{le}}", "range": true, @@ -296,13 +1508,79 @@ "title": "Pending existance latency", "type": "bargauge" }, + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green" + }, + { + "color": "red", + "value": 80 + } + ] + } + }, + "overrides": [] + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 63 + }, + "id": 18, + "options": { + "displayMode": "gradient", + "minVizHeight": 10, + "minVizWidth": 0, + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + }, + "showUnfilled": true, + "valueMode": "color" + }, + "pluginVersion": "10.0.1", + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "editorMode": "code", + "expr": "sum(increase(topos_storage_adding_pending_certificate_latency_bucket{ job=~\"$peer\"}[$__range])) by (le)", + "format": "heatmap", + "legendFormat": "__auto", + "range": true, + "refId": "A" + } + ], + "title": "Adding to pending latency", + "type": "bargauge" + }, { "collapsed": false, "gridPos": { "h": 1, "w": 24, "x": 0, - "y": 21 + "y": 71 }, "id": 11, "panels": [], @@ -312,7 +1590,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -354,8 +1632,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -370,7 +1647,7 @@ "h": 12, "w": 12, "x": 0, - "y": 22 + "y": 72 }, "id": 3, "options": { @@ -389,11 +1666,11 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(p2p_event_stream_capacity[$__interval])", - "legendFormat": "{{pod}}", + "editorMode": "code", + "expr": "irate(topos_p2p_event_stream_capacity_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "{{job}}", "range": true, "refId": "A" } @@ -404,7 +1681,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -446,8 +1723,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -462,7 +1738,7 @@ "h": 12, "w": 12, "x": 12, - "y": 22 + "y": 72 }, "id": 2, "options": { @@ -481,35 +1757,35 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(echo_message_count[$__interval])", - "legendFormat": "Echo {{pod}}", + "editorMode": "code", + "expr": "irate(topos_p2p_echo_message_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "Echo {{job}}", "range": true, "refId": "A" }, { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(ready_message_count[$__interval])", + "editorMode": "code", + "expr": "irate(topos_p2p_ready_message_total{ job=~\"$peer\"}[$__interval])", "hide": false, - "legendFormat": "Ready {{pod}}", + "legendFormat": "Ready {{job}}", "range": true, "refId": "B" }, { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(gossip_message_count[$__interval])", + "editorMode": "code", + "expr": "irate(topos_p2p_gossip_message_total{ job=~\"$peer\"}[$__interval])", "hide": false, - "legendFormat": "Gossip {{pod}}", + "legendFormat": "Gossip {{job}}", "range": true, "refId": "C" } @@ -520,7 +1796,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -562,8 +1838,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -578,7 +1853,7 @@ "h": 12, "w": 12, "x": 0, - "y": 34 + "y": 84 }, "id": 1, "options": { @@ -597,11 +1872,11 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "sum(gossipsub_message_sent_count)", - "legendFormat": "{{pod}}", + "editorMode": "code", + "expr": "sum(topos_p2p_gossipsub_message_sent_total{ job=~\"$peer\"})", + "legendFormat": "{{job}}", "range": true, "refId": "A" } @@ -612,7 +1887,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -624,8 +1899,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -640,7 +1914,7 @@ "h": 12, "w": 12, "x": 12, - "y": 34 + "y": 84 }, "id": 7, "options": { @@ -663,11 +1937,11 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "editorMode": "code", "exemplar": false, - "expr": "sum(increase(p2p_gossip_batch_size_bucket[$__range])) by (le)", + "expr": "sum(increase(topos_p2p_gossip_batch_size_bucket{ job=~\"$peer\"}[$__range])) by (le)", "format": "heatmap", "instant": false, "legendFormat": "{{le}}", @@ -684,7 +1958,7 @@ "h": 1, "w": 24, "x": 0, - "y": 46 + "y": 96 }, "id": 12, "panels": [], @@ -694,7 +1968,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -736,8 +2010,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -752,7 +2025,7 @@ "h": 8, "w": 12, "x": 0, - "y": 47 + "y": 97 }, "id": 13, "options": { @@ -771,11 +2044,11 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(double_echo_command_channel_capacity[$__interval])", - "legendFormat": "{{pod}}", + "editorMode": "code", + "expr": "irate(topos_double_echo_command_channel_capacity_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "{{job}}", "range": true, "refId": "A" } @@ -789,7 +2062,7 @@ "h": 1, "w": 24, "x": 0, - "y": 55 + "y": 105 }, "id": 10, "panels": [], @@ -799,7 +2072,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -841,8 +2114,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -857,7 +2129,7 @@ "h": 8, "w": 12, "x": 0, - "y": 56 + "y": 106 }, "id": 4, "options": { @@ -876,12 +2148,12 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "double_echo_current_buffer_size", + "editorMode": "code", + "expr": "topos_double_echo_current_buffer_size{ job=~\"$peer\"}", "hide": false, - "legendFormat": "{{pod}}", + "legendFormat": "{{job}}", "range": true, "refId": "B" } @@ -892,7 +2164,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -934,8 +2206,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -950,7 +2221,7 @@ "h": 8, "w": 12, "x": 12, - "y": 56 + "y": 106 }, "id": 5, "options": { @@ -969,11 +2240,11 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "irate(double_echo_buffer_capacity[$__interval])", - "legendFormat": "{{pod}}", + "editorMode": "code", + "expr": "irate(topos_double_echo_buffer_capacity_total{ job=~\"$peer\"}[$__interval])", + "legendFormat": "{{job}}", "range": true, "refId": "A" } @@ -984,7 +2255,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -1026,8 +2297,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -1042,7 +2312,7 @@ "h": 8, "w": 12, "x": 0, - "y": 64 + "y": 114 }, "id": 6, "options": { @@ -1061,12 +2331,12 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "double_echo_buffered_message_count", + "editorMode": "code", + "expr": "topos_double_echo_buffered_message_count{ job=~\"$peer\"}", "hide": false, - "legendFormat": "{{pod}}", + "legendFormat": "{{job}}", "range": true, "refId": "B" } @@ -1077,7 +2347,7 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, "fieldConfig": { "defaults": { @@ -1119,8 +2389,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -1135,7 +2404,7 @@ "h": 8, "w": 12, "x": 12, - "y": 64 + "y": 114 }, "id": 8, "options": { @@ -1154,12 +2423,12 @@ { "datasource": { "type": "prometheus", - "uid": "PBFA97CFB590B2093" + "uid": "${datasource}" }, - "editorMode": "builder", - "expr": "double_echo_current_buffer_size", + "editorMode": "code", + "expr": "topos_double_echo_current_buffer_size{ job=~\"$peer\"}", "hide": false, - "legendFormat": "{{pod}}", + "legendFormat": "{{job}}", "range": true, "refId": "B" } @@ -1173,7 +2442,53 @@ "style": "dark", "tags": [], "templating": { - "list": [] + "list": [ + { + "current": { + "selected": true, + "text": [ + "boot" + ], + "value": [ + "boot" + ] + }, + "definition": "label_values(topos_certificate_received_total,job)", + "hide": 0, + "includeAll": true, + "multi": true, + "name": "peer", + "options": [], + "query": { + "query": "label_values(topos_certificate_received_total,job)", + "refId": "PrometheusVariableQueryEditor-VariableQuery" + }, + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 0, + "type": "query" + }, + { + "current": { + "selected": false, + "text": "Prometheus", + "value": "Prometheus" + }, + "hide": 0, + "includeAll": false, + "label": "Datasource", + "multi": false, + "name": "datasource", + "options": [], + "query": "prometheus", + "queryValue": "", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + } + ] }, "time": { "from": "now-5m", @@ -1181,8 +2496,8 @@ }, "timepicker": {}, "timezone": "", - "title": "Benchmarks", - "uid": "df73b86a-e7e5-4b81-979c-e5d50e4c1eac", - "version": 1, + "title": "Benchmarks Copy", + "uid": "e3d3b025-4b36-454a-a724-818f85806b6e", + "version": 3, "weekStart": "" } diff --git a/tools/docker-compose.yml b/tools/docker-compose.yml index f1f813366..8c1c5fb2f 100644 --- a/tools/docker-compose.yml +++ b/tools/docker-compose.yml @@ -8,6 +8,7 @@ services: init: true labels: "autoheal": "true" + "prometheus-job": "boot" healthcheck: test: ./topos tce push-peer-list --node http://localhost:1340 --format json /tmp/shared/peer_ids.json && ./topos tce status --node http://localhost:1340 interval: 15s