pezkuwichain/pezkuwi-telemetry
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-telemetry.git synced 2026-06-12 20:21:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move a bunch of things around and flatten common crate

Browse Source
This commit is contained in:
James Wilson
2021-06-30 17:27:10 +01:00
parent 4308359feb
commit f7ab3292c2
21 changed files with 217 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/common/src/util/dense_map.rs → backend/common/src/dense_map.rs
+2 -2
View File
@@ -4,7 +4,7 @@ pub struct DenseMap<Id, T> {
/// All items
items: Vec<Option<T>>,
/// Our ID type
_id_ty: std::marker::PhantomData<Id>
_id_type: std::marker::PhantomData<Id>
}
impl<Id, T> DenseMap<Id, T>
@@ -16,7 +16,7 @@ where
DenseMap {
retired: Vec::new(),
items: Vec::new(),
_id_ty: std::marker::PhantomData
_id_type: std::marker::PhantomData
}
}
backend/common/src/internal_messages.rs
+2 -2
View File
@@ -1,7 +1,7 @@
use std::net::IpAddr;
use crate::node::Payload;
use crate::types::{NodeDetails, BlockHash};
use crate::node_message::Payload;
use crate::node_types::{NodeDetails, BlockHash};
use crate::id_type;
use serde::{Deserialize, Serialize};
backend/common/src/json/hash.rs
-213
View File
@@ -1,213 +0,0 @@
use std::fmt::{self, Debug, Display};
use std::str::FromStr;
use serde::ser::{Serialize, Serializer};
use serde::de::{self, Deserialize, Deserializer, Unexpected, Visitor, SeqAccess};
const HASH_BYTES: usize = 32;
/// Newtype wrapper for 32-byte hash values, implementing readable `Debug` and `serde::Deserialize`.
/// This can deserialize from a JSON string or array.
#[derive(Hash, PartialEq, Eq, Clone, Copy)]
pub struct Hash([u8; HASH_BYTES]);
impl From<Hash> for crate::types::BlockHash {
fn from(hash: Hash) -> Self {
hash.0.into()
}
}
impl From<crate::types::BlockHash> for Hash {
fn from(hash: crate::types::BlockHash) -> Self {
Hash(hash.0)
}
}
struct HashVisitor;
impl<'de> Visitor<'de> for HashVisitor {
type Value = Hash;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("byte array of length 32, or hexidecimal string of 32 bytes beginning with 0x")
}
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
value
.parse()
.map_err(|_| de::Error::invalid_value(Unexpected::Str(value), &self))
}
fn visit_bytes<E>(self, value: &[u8]) -> Result<Self::Value, E>
where
E: de::Error,
{
if value.len() == HASH_BYTES {
let mut hash = [0; HASH_BYTES];
hash.copy_from_slice(value);
return Ok(Hash(hash));
}
Hash::from_ascii(value)
.map_err(|_| de::Error::invalid_value(Unexpected::Bytes(value), &self))
}
fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
where
A: SeqAccess<'de>,
{
let mut hash = [0u8; HASH_BYTES];
for (i, byte) in hash.iter_mut().enumerate() {
match seq.next_element()? {
Some(b) => *byte = b,
None => return Err(de::Error::invalid_length(i, &"an array of 32 bytes"))
}
}
if seq.next_element::<u8>()?.is_some() {
return Err(de::Error::invalid_length(33, &"an array of 32 bytes"));
}
Ok(Hash(hash))
}
}
impl Hash {
pub fn from_ascii(value: &[u8]) -> Result<Self, HashParseError> {
if !value.starts_with(b"0x") {
return Err(HashParseError::InvalidPrefix);
}
let mut hash = [0; HASH_BYTES];
hex::decode_to_slice(&value[2..], &mut hash).map_err(HashParseError::HexError)?;
Ok(Hash(hash))
}
}
impl FromStr for Hash {
type Err = HashParseError;
fn from_str(value: &str) -> Result<Self, Self::Err> {
Hash::from_ascii(value.as_bytes())
}
}
impl<'de> Deserialize<'de> for Hash {
fn deserialize<D>(deserializer: D) -> Result<Hash, D::Error>
where
D: Deserializer<'de>,
{
deserializer.deserialize_bytes(HashVisitor)
}
}
impl Serialize for Hash {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_bytes(&self.0)
}
}
impl Display for Hash {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str("0x")?;
let mut ascii = [0; HASH_BYTES * 2];
hex::encode_to_slice(self.0, &mut ascii)
.expect("Encoding 32 bytes into 64 bytes of ascii; qed");
f.write_str(std::str::from_utf8(&ascii).expect("ASCII hex encoded bytes canot fail; qed"))
}
}
impl Debug for Hash {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
Display::fmt(self, f)
}
}
#[derive(thiserror::Error, Debug)]
pub enum HashParseError {
#[error("Error parsing string into hex: {0}")]
HexError(hex::FromHexError),
#[error("Invalid hex prefix: expected '0x'")]
InvalidPrefix,
}
#[cfg(test)]
mod tests {
use super::Hash;
use bincode::Options;
const DUMMY: Hash = {
let mut hash = [0; 32];
hash[0] = 0xDE;
hash[1] = 0xAD;
hash[2] = 0xBE;
hash[3] = 0xEF;
Hash(hash)
};
#[test]
fn deserialize_json_hash_str() {
let json = r#""0xdeadBEEF00000000000000000000000000000000000000000000000000000000""#;
let hash: Hash = serde_json::from_str(json).unwrap();
assert_eq!(hash, DUMMY);
}
#[test]
fn deserialize_json_array() {
let json = r#"[222,173,190,239,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]"#;
let hash: Hash = serde_json::from_str(json).unwrap();
assert_eq!(hash, DUMMY);
}
#[test]
fn deserialize_json_array_too_short() {
let json = r#"[222,173,190,239,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]"#;
let res = serde_json::from_str::<Hash>(json);
assert!(res.is_err());
}
#[test]
fn deserialize_json_array_too_long() {
let json = r#"[222,173,190,239,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]"#;
let res = serde_json::from_str::<Hash>(json);
assert!(res.is_err());
}
#[test]
fn bincode() {
let bytes = bincode::options().serialize(&DUMMY).unwrap();
let mut expected = [0; 33];
expected[0] = 32; // length
expected[1..].copy_from_slice(&DUMMY.0);
assert_eq!(bytes, &expected);
let deserialized: Hash = bincode::options().deserialize(&bytes).unwrap();
assert_eq!(DUMMY, deserialized);
}
}
backend/common/src/json/mod.rs
-7
View File
@@ -1,7 +0,0 @@
//! This module contains the types we need to deserialize JSON messages from nodes
mod hash;
mod node_message;
pub use node_message::*;
pub use hash::Hash;
backend/common/src/json/node_message.rs
-223
View File
@@ -1,223 +0,0 @@
//! The structs and enums defined in this module are largely identical to those
//! we'll use elsewhere internally, but are kept separate so that the JSON structure
//! is defined (almost) from just this file, and we don't have to worry about breaking
//! compatibility with the input data when we make changes to our internal data
//! structures (for example, to support bincode better).
use super::hash::Hash;
use serde::{Deserialize};
/// This struct represents a telemetry message sent from a node as
/// a JSON payload. Since JSON is self describing, we can use attributes
/// like serde(untagged) and serde(flatten) without issue.
///
/// Internally, we want to minimise the amount of data sent from shards to
/// the core node. For that reason, we use a non-self-describing serialization
/// format like bincode, which doesn't support things like `[serde(flatten)]` (which
/// internally wants to serialize to a map of unknown length) or `[serde(tag/untagged)]`
/// (which relies on the data to know which variant to deserialize to.)
///
/// So, this can be converted fairly cheaply into an enum we'll use internally
/// which is compatible with formats like bincode.
#[derive(Deserialize, Debug)]
#[serde(untagged)]
pub enum NodeMessage {
V1 {
#[serde(flatten)]
payload: Payload,
},
V2 {
id: NodeMessageId,
payload: Payload,
},
}
#[derive(Deserialize, Debug)]
#[serde(tag = "msg")]
pub enum Payload {
#[serde(rename = "system.connected")]
SystemConnected(SystemConnected),
#[serde(rename = "system.interval")]
SystemInterval(SystemInterval),
#[serde(rename = "block.import")]
BlockImport(Block),
#[serde(rename = "notify.finalized")]
NotifyFinalized(Finalized),
#[serde(rename = "txpool.import")]
TxPoolImport,
#[serde(rename = "afg.finalized")]
AfgFinalized(AfgFinalized),
#[serde(rename = "afg.received_precommit")]
AfgReceivedPrecommit(AfgReceived),
#[serde(rename = "afg.received_prevote")]
AfgReceivedPrevote(AfgReceived),
#[serde(rename = "afg.received_commit")]
AfgReceivedCommit(AfgReceived),
#[serde(rename = "afg.authority_set")]
AfgAuthoritySet(AfgAuthoritySet),
#[serde(rename = "afg.finalized_blocks_up_to")]
AfgFinalizedBlocksUpTo,
#[serde(rename = "aura.pre_sealed_block")]
AuraPreSealedBlock,
#[serde(rename = "prepared_block_for_proposing")]
PreparedBlockForProposing,
}
#[derive(Deserialize, Debug)]
pub struct SystemConnected {
pub genesis_hash: Hash,
#[serde(flatten)]
pub node: NodeDetails,
}
#[derive(Deserialize, Debug)]
pub struct SystemInterval {
pub peers: Option<u64>,
pub txcount: Option<u64>,
pub bandwidth_upload: Option<f64>,
pub bandwidth_download: Option<f64>,
pub finalized_height: Option<BlockNumber>,
pub finalized_hash: Option<Hash>,
#[serde(flatten)]
pub block: Option<Block>,
pub used_state_cache_size: Option<f32>,
}
#[derive(Deserialize, Debug)]
pub struct Finalized {
#[serde(rename = "best")]
pub hash: Hash,
pub height: Box<str>,
}
#[derive(Deserialize, Debug)]
pub struct AfgAuthoritySet {
pub authority_id: Box<str>,
pub authorities: Box<str>,
pub authority_set_id: Box<str>,
}
#[derive(Deserialize, Debug, Clone)]
pub struct AfgFinalized {
pub finalized_hash: Hash,
pub finalized_number: Box<str>,
}
#[derive(Deserialize, Debug, Clone)]
pub struct AfgReceived {
pub target_hash: Hash,
pub target_number: Box<str>,
pub voter: Option<Box<str>>,
}
#[derive(Deserialize, Debug, Clone, Copy)]
pub struct Block {
#[serde(rename = "best")]
pub hash: Hash,
pub height: BlockNumber,
}
#[derive(Deserialize, Debug, Clone)]
pub struct NodeDetails {
pub chain: Box<str>,
pub name: Box<str>,
pub implementation: Box<str>,
pub version: Box<str>,
pub validator: Option<Box<str>>,
pub network_id: Option<Box<str>>,
pub startup_time: Option<Box<str>>,
}
type NodeMessageId = u64;
type BlockNumber = u64;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn message_v1() {
let json = r#"{
"msg":"notify.finalized",
"level":"INFO",
"ts":"2021-01-13T12:38:25.410794650+01:00",
"best":"0x031c3521ca2f9c673812d692fc330b9a18e18a2781e3f9976992f861fd3ea0cb",
"height":"50"
}"#;
assert!(
matches!(
serde_json::from_str::<NodeMessage>(json).unwrap(),
NodeMessage::V1 { .. },
),
"message did not match variant V1",
);
}
#[test]
fn message_v2() {
let json = r#"{
"id":1,
"ts":"2021-01-13T12:22:20.053527101+01:00",
"payload":{
"best":"0xcc41708573f2acaded9dd75e07dac2d4163d136ca35b3061c558d7a35a09dd8d",
"height":"209",
"msg":"notify.finalized"
}
}"#;
assert!(
matches!(
serde_json::from_str::<NodeMessage>(json).unwrap(),
NodeMessage::V2 { .. },
),
"message did not match variant V2",
);
}
#[test]
fn message_v2_received_precommit() {
let json = r#"{
"id":1,
"ts":"2021-01-13T12:22:20.053527101+01:00",
"payload":{
"target_hash":"0xcc41708573f2acaded9dd75e07dac2d4163d136ca35b3061c558d7a35a09dd8d",
"target_number":"209",
"voter":"foo",
"msg":"afg.received_precommit"
}
}"#;
assert!(
matches!(
serde_json::from_str::<NodeMessage>(json).unwrap(),
NodeMessage::V2 {
payload: Payload::AfgReceivedPrecommit(..),
..
},
),
"message did not match the expected output",
);
}
#[test]
fn message_v2_tx_pool_import() {
// We should happily ignore any fields we don't care about.
let json = r#"{
"id":1,
"ts":"2021-01-13T12:22:20.053527101+01:00",
"payload":{
"foo":"Something",
"bar":123,
"wibble":"wobble",
"msg":"txpool.import"
}
}"#;
assert!(
matches!(
serde_json::from_str::<NodeMessage>(json).unwrap(),
NodeMessage::V2 {
payload: Payload::TxPoolImport,
..
},
),
"message did not match the expected output",
);
}
}
backend/common/src/lib.rs
+19 -8
View File
@@ -1,9 +1,20 @@
pub mod node;
pub mod node_message;
pub mod internal_messages;
pub mod types;
pub mod util;
pub mod json;
pub mod log_level;
pub mod assign_id;
pub mod most_seen;
pub mod id_type;
pub mod node_types;
pub mod id_type;
pub mod time;
mod log_level;
mod assign_id;
mod most_seen;
mod dense_map;
mod mean_list;
mod num_stats;
// Export a bunch of common bits at the top level for ease of import:
pub use assign_id::AssignId;
pub use dense_map::DenseMap;
pub use mean_list::MeanList;
pub use num_stats::NumStats;
pub use most_seen::MostSeen;
pub use log_level::LogLevel;
backend/common/src/util/mean_list.rs → backend/common/src/mean_list.rs
View File
backend/common/src/node.rs → backend/common/src/node_message.rs
+1 -123
View File
@@ -1,5 +1,4 @@
use crate::types::{Block, BlockHash, BlockNumber, NodeDetails};
use crate::json;
use crate::node_types::{Block, BlockHash, BlockNumber, NodeDetails};
use serde::{Deserialize, Serialize};
pub type NodeMessageId = u64;
@@ -39,19 +38,6 @@ impl From<NodeMessage> for Payload {
}
}
impl From<json::NodeMessage> for NodeMessage {
fn from(msg: json::NodeMessage) -> Self {
match msg {
json::NodeMessage::V1 { payload } => {
NodeMessage::V1 { payload: payload.into() }
},
json::NodeMessage::V2 { id, payload } => {
NodeMessage::V2 { id, payload: payload.into() }
},
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum Payload {
SystemConnected(SystemConnected),
@@ -69,67 +55,12 @@ pub enum Payload {
PreparedBlockForProposing,
}
impl From<json::Payload> for Payload {
fn from(msg: json::Payload) -> Self {
match msg {
json::Payload::SystemConnected(m) => {
Payload::SystemConnected(m.into())
},
json::Payload::SystemInterval(m) => {
Payload::SystemInterval(m.into())
},
json::Payload::BlockImport(m) => {
Payload::BlockImport(m.into())
},
json::Payload::NotifyFinalized(m) => {
Payload::NotifyFinalized(m.into())
},
json::Payload::TxPoolImport => {
Payload::TxPoolImport
},
json::Payload::AfgFinalized(m) => {
Payload::AfgFinalized(m.into())
},
json::Payload::AfgReceivedPrecommit(m) => {
Payload::AfgReceivedPrecommit(m.into())
},
json::Payload::AfgReceivedPrevote(m) => {
Payload::AfgReceivedPrevote(m.into())
},
json::Payload::AfgReceivedCommit(m) => {
Payload::AfgReceivedCommit(m.into())
},
json::Payload::AfgAuthoritySet(m) => {
Payload::AfgAuthoritySet(m.into())
},
json::Payload::AfgFinalizedBlocksUpTo => {
Payload::AfgFinalizedBlocksUpTo
},
json::Payload::AuraPreSealedBlock => {
Payload::AuraPreSealedBlock
},
json::Payload::PreparedBlockForProposing => {
Payload::PreparedBlockForProposing
},
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SystemConnected {
pub genesis_hash: BlockHash,
pub node: NodeDetails,
}
impl From<json::SystemConnected> for SystemConnected {
fn from(msg: json::SystemConnected) -> Self {
SystemConnected {
genesis_hash: msg.genesis_hash.into(),
node: msg.node.into()
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SystemInterval {
pub peers: Option<u64>,
@@ -142,51 +73,18 @@ pub struct SystemInterval {
pub used_state_cache_size: Option<f32>,
}
impl From<json::SystemInterval> for SystemInterval {
fn from(msg: json::SystemInterval) -> Self {
SystemInterval {
peers: msg.peers,
txcount: msg.txcount,
bandwidth_upload: msg.bandwidth_upload,
bandwidth_download: msg.bandwidth_download,
finalized_height: msg.finalized_height,
finalized_hash: msg.finalized_hash.map(|h| h.into()),
block: msg.block.map(|b| b.into()),
used_state_cache_size: msg.used_state_cache_size,
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Finalized {
pub hash: BlockHash,
pub height: Box<str>,
}
impl From<json::Finalized> for Finalized {
fn from(msg: json::Finalized) -> Self {
Finalized {
hash: msg.hash.into(),
height: msg.height,
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct AfgFinalized {
pub finalized_hash: BlockHash,
pub finalized_number: Box<str>,
}
impl From<json::AfgFinalized> for AfgFinalized {
fn from(msg: json::AfgFinalized) -> Self {
AfgFinalized {
finalized_hash: msg.finalized_hash.into(),
finalized_number: msg.finalized_number,
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct AfgReceived {
pub target_hash: BlockHash,
@@ -194,16 +92,6 @@ pub struct AfgReceived {
pub voter: Option<Box<str>>,
}
impl From<json::AfgReceived> for AfgReceived {
fn from(msg: json::AfgReceived) -> Self {
AfgReceived {
target_hash: msg.target_hash.into(),
target_number: msg.target_number,
voter: msg.voter,
}
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct AfgAuthoritySet {
pub authority_id: Box<str>,
@@ -211,16 +99,6 @@ pub struct AfgAuthoritySet {
pub authority_set_id: Box<str>,
}
impl From<json::AfgAuthoritySet> for AfgAuthoritySet {
fn from(msg: json::AfgAuthoritySet) -> Self {
AfgAuthoritySet {
authority_id: msg.authority_id,
authorities: msg.authorities,
authority_set_id: msg.authority_set_id,
}
}
}
impl Payload {
pub fn best_block(&self) -> Option<&Block> {
match self {
backend/common/src/types.rs → backend/common/src/node_types.rs
+2 -26
View File
@@ -1,8 +1,7 @@
use serde::ser::{SerializeTuple, Serializer};
use serde::{Deserialize, Serialize};
use crate::util::{now, MeanList};
use crate::json;
use crate::{time, MeanList};
pub type BlockNumber = u64;
pub type Timestamp = u64;
@@ -19,20 +18,6 @@ pub struct NodeDetails {
pub startup_time: Option<Box<str>>,
}
impl From<json::NodeDetails> for NodeDetails {
fn from(details: json::NodeDetails) -> Self {
NodeDetails {
chain: details.chain,
name: details.name,
implementation: details.implementation,
version: details.version,
validator: details.validator,
network_id: details.network_id,
startup_time: details.startup_time,
}
}
}
#[derive(Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct NodeStats {
pub peers: u64,
@@ -50,15 +35,6 @@ pub struct Block {
pub height: BlockNumber,
}
impl From<json::Block> for Block {
fn from(block: json::Block) -> Self {
Block {
hash: block.hash.into(),
height: block.height
}
}
}
impl Block {
pub fn zero() -> Self {
Block {
@@ -80,7 +56,7 @@ impl Default for BlockDetails {
fn default() -> Self {
BlockDetails {
block: Block::zero(),
block_timestamp: now(),
block_timestamp: time::now(),
block_time: 0,
propagation_time: None,
}
backend/common/src/util/num_stats.rs → backend/common/src/num_stats.rs
View File
backend/common/src/time.rs
+9
View File
@@ -0,0 +1,9 @@
/// Returns current unix time in ms (compatible with JS Date.now())
pub fn now() -> u64 {
use std::time::SystemTime;
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.expect("System time must be configured to be post Unix Epoch start; qed")
.as_millis() as u64
}
backend/common/src/util.rs
-27
View File
@@ -1,27 +0,0 @@
mod dense_map;
mod mean_list;
mod num_stats;
pub use dense_map::DenseMap;
pub use mean_list::MeanList;
pub use num_stats::NumStats;
pub fn fnv<D: AsRef<[u8]>>(data: D) -> u64 {
use fnv::FnvHasher;
use std::hash::Hasher;
let mut hasher = FnvHasher::default();
hasher.write(data.as_ref());
hasher.finish()
}
/// Returns current unix time in ms (compatible with JS Date.now())
pub fn now() -> u64 {
use std::time::SystemTime;
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.expect("System time must be configured to be post Unix Epoch start; qed")
.as_millis() as u64
}