pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-17 19:31:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

restructure polkadot-node-jaeger (#2642)

Browse Source
This commit is contained in:
Bernhard Schuster
2021-03-19 16:51:16 +01:00
committed by GitHub
parent 59640a38bc
commit ea6294fa79
19 changed files with 658 additions and 498 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polkadot/node/jaeger/src/lib.rs
+38 -390
View File
@@ -25,7 +25,7 @@
//! The easiest way to try Jaeger is:
//!
//! - Start a docker container with the all-in-one docker image (see below).
//! - Open your browser and navigate to <https://localhost:16686> to acces the UI.
//! - Open your browser and navigate to <https://localhost:16686> to access the UI.
//!
//! The all-in-one image can be started with:
//!
@@ -44,375 +44,27 @@
//! docker.io/jaegertracing/all-in-one:1.21
//! ```
#![forbid(unused_imports)]
mod config;
mod errors;
mod spans;
pub use self::config::{JaegerConfig, JaegerConfigBuilder};
pub use self::errors::JaegerError;
pub use self::spans::{PerLeafSpan, Span, Stage};
use self::spans::TraceIdentifier;
use sp_core::traits::SpawnNamed;
use polkadot_primitives::v1::{CandidateHash, Hash, PoV, ValidatorIndex, BlakeTwo256, HashT, Id as ParaId};
use parity_scale_codec::Encode;
use sc_network::PeerId;
use parking_lot::RwLock;
use std::{sync::Arc, result};
/// A description of an error causing the chain API request to be unservable.
#[derive(Debug, thiserror::Error)]
#[allow(missing_docs)]
pub enum JaegerError {
#[error("Already launched the collector thread")]
AlreadyLaunched,
#[error("Missing jaeger configuration")]
MissingConfiguration,
}
use std::{result, sync::Arc};
lazy_static::lazy_static! {
static ref INSTANCE: RwLock<Jaeger> = RwLock::new(Jaeger::None);
}
/// Configuration for the jaeger tracing.
#[derive(Clone)]
pub struct JaegerConfig {
node_name: String,
agent_addr: std::net::SocketAddr,
}
impl std::default::Default for JaegerConfig {
fn default() -> Self {
Self {
node_name: "unknown_".to_owned(),
agent_addr: "127.0.0.1:6831".parse().expect(r#"Static "127.0.0.1:6831" is a valid socket address string. qed"#),
}
}
}
impl JaegerConfig {
/// Use the builder pattern to construct a configuration.
pub fn builder() -> JaegerConfigBuilder {
JaegerConfigBuilder::default()
}
}
/// Jaeger configuration builder.
#[derive(Default)]
pub struct JaegerConfigBuilder {
inner: JaegerConfig
}
impl JaegerConfigBuilder {
/// Set the name for this node.
pub fn named<S>(mut self, name: S) -> Self where S: AsRef<str> {
self.inner.node_name = name.as_ref().to_owned();
self
}
/// Set the agent address to send the collected spans to.
pub fn agent<U>(mut self, addr: U) -> Self where U: Into<std::net::SocketAddr> {
self.inner.agent_addr = addr.into();
self
}
/// Construct the configuration.
pub fn build(self) -> JaegerConfig {
self.inner
}
}
/// A special "per leaf span".
///
/// Essentially this span wraps two spans:
///
/// 1. The span that is created per leaf in the overseer.
/// 2. Some child span of the per-leaf span.
///
/// This just works as auxiliary structure to easily store both.
#[derive(Debug)]
pub struct PerLeafSpan {
leaf_span: Arc<Span>,
span: Span,
}
impl PerLeafSpan {
/// Creates a new instance.
///
/// Takes the `leaf_span` that is created by the overseer per leaf and a name for a child span.
/// Both will be stored in this object, while the child span is implicitly accessible by using the
/// [`Deref`](std::ops::Deref) implementation.
pub fn new(leaf_span: Arc<Span>, name: &'static str) -> Self {
let span = leaf_span.child(name);
Self {
span,
leaf_span,
}
}
/// Returns the leaf span.
pub fn leaf_span(&self) -> &Arc<Span> {
&self.leaf_span
}
}
/// Returns a reference to the child span.
impl std::ops::Deref for PerLeafSpan {
type Target = Span;
fn deref(&self) -> &Span {
&self.span
}
}
/// A helper to annotate the stage with a numerical value
/// to ease the life of the tooling team creating viable
/// statistical metrics for which stage of the inclusion
/// pipeline drops a significant amount of candidates,
/// statistically speaking.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
#[non_exhaustive]
pub enum Stage {
CandidateSelection = 1,
CandidateBacking = 2,
StatementDistribution = 3,
PoVDistribution = 4,
AvailabilityDistribution = 5,
AvailabilityRecovery = 6,
BitfieldDistribution = 7,
ApprovalChecking = 8,
// Expand as needed, numbers should be ascending according to the stage
// through the inclusion pipeline, or according to the descriptions
// in [the path of a para chain block]
// (https://polkadot.network/the-path-of-a-parachain-block/)
// see [issue](https://github.com/paritytech/polkadot/issues/2389)
}
/// Builder pattern for children and root spans to unify
/// information annotations.
pub struct SpanBuilder {
span: Span,
}
impl SpanBuilder {
/// Attach a peer id to the span.
#[inline(always)]
pub fn with_peer_id(mut self, peer: &PeerId) -> Self {
self.span.add_string_tag("peer-id", &peer.to_base58());
self
}
/// Attach a candidate hash to the span.
#[inline(always)]
pub fn with_candidate(mut self, candidate_hash: &CandidateHash) -> Self {
self.span.add_string_tag("candidate-hash", &format!("{:?}", candidate_hash.0));
self
}
/// Attach a para-id to the span.
#[inline(always)]
pub fn with_para_id(mut self, para_id: ParaId) -> Self {
self.span.add_para_id(para_id);
self
}
/// Attach a candidate stage.
/// Should always come with a `CandidateHash`.
#[inline(always)]
pub fn with_stage(mut self, stage: Stage) -> Self {
self.span.add_stage(stage);
self
}
#[inline(always)]
pub fn with_validator_index(mut self, validator: ValidatorIndex) -> Self {
self.span.add_string_tag("validator-index", &validator.0.to_string());
self
}
#[inline(always)]
pub fn with_chunk_index(mut self, chunk_index: u32) -> Self {
self.span.add_string_tag("chunk-index", &format!("{}", chunk_index));
self
}
#[inline(always)]
pub fn with_relay_parent(mut self, relay_parent: &Hash) -> Self {
self.span.add_string_tag("relay-parent", &format!("{:?}", relay_parent));
self
}
#[inline(always)]
pub fn with_claimed_validator_index(mut self, claimed_validator_index: ValidatorIndex) -> Self {
self.span.add_string_tag(
"claimed-validator",
&claimed_validator_index.0.to_string(),
);
self
}
#[inline(always)]
pub fn with_pov(mut self, pov: &PoV) -> Self {
self.span.add_pov(pov);
self
}
/// Complete construction.
#[inline(always)]
pub fn build(self) -> Span {
self.span
}
}
/// A wrapper type for a span.
///
/// Handles running with and without jaeger.
pub enum Span {
/// Running with jaeger being enabled.
Enabled(mick_jaeger::Span),
/// Running with jaeger disabled.
Disabled,
}
impl Span {
/// Derive a child span from `self`.
pub fn child(&self, name: &'static str) -> Self {
match self {
Self::Enabled(inner) => Self::Enabled(inner.child(name)),
Self::Disabled => Self::Disabled,
}
}
pub fn child_builder(&self, name: &'static str) -> SpanBuilder {
SpanBuilder {
span: self.child(name),
}
}
/// Derive a child span from `self` but add a candidate annotation.
/// A shortcut for
///
/// ```rust,no_run
/// # use polkadot_primitives::v1::CandidateHash;
/// # use polkadot_node_jaeger::candidate_hash_span;
/// # let hash = CandidateHash::default();
/// # let span = candidate_hash_span(&hash, "foo");
/// let _span = span.child_builder("name").with_candidate(&hash).build();
/// ```
#[inline(always)]
pub fn child_with_candidate(&self, name: &'static str, candidate_hash: &CandidateHash) -> Self {
self.child_builder(name).with_candidate(candidate_hash).build()
}
/// Add candidate stage annotation.
pub fn add_stage(&mut self, stage: Stage) {
self.add_string_tag("candidate-stage", &format!("{}", stage as u8));
}
pub fn add_pov(&mut self, pov: &PoV) {
if self.is_enabled() {
// avoid computing the pov hash if jaeger is not enabled
self.add_string_tag("pov", &format!("{:?}", pov.hash()));
}
}
/// Add the para-id to the span.
pub fn add_para_id(&mut self, para_id: ParaId) {
self.add_int_tag("para-id", u32::from(para_id) as i64);
}
/// Add an additional tag to the span.
pub fn add_string_tag(&mut self, tag: &str, value: &str) {
match self {
Self::Enabled(ref mut inner) => inner.add_string_tag(tag, value),
Self::Disabled => {},
}
}
/// Add an additional int tag to the span.
pub fn add_int_tag(&mut self, tag: &str, value: i64) {
match self {
Self::Enabled(ref mut inner) => inner.add_int_tag(tag, value),
Self::Disabled => {},
}
}
/// Adds the `FollowsFrom` relationship to this span with respect to the given one.
pub fn add_follows_from(&mut self, other: &Self) {
match (self, other) {
(Self::Enabled(ref mut inner), Self::Enabled(ref other_inner)) => inner.add_follows_from(&other_inner),
_ => {},
}
}
/// Helper to check whether jaeger is enabled
/// in order to avoid computational overhead.
pub const fn is_enabled(&self) -> bool {
match self {
Span::Enabled(_) => true,
_ => false,
}
}
}
impl std::fmt::Debug for Span {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
write!(f, "<jaeger span>")
}
}
impl From<Option<mick_jaeger::Span>> for Span {
fn from(src: Option<mick_jaeger::Span>) -> Self {
if let Some(span) = src {
Self::Enabled(span)
} else {
Self::Disabled
}
}
}
impl From<mick_jaeger::Span> for Span {
fn from(src: mick_jaeger::Span) -> Self {
Self::Enabled(src)
}
}
/// Shortcut for [`hash_span`] with the hash of the `Candidate` block.
pub fn candidate_hash_span(candidate_hash: &CandidateHash, span_name: &'static str) -> Span {
let mut span: Span = INSTANCE.read_recursive()
.span(|| { candidate_hash.0 }, span_name).into();
span.add_string_tag("candidate-hash", &format!("{:?}", candidate_hash.0));
span
}
/// Shortcut for [`hash_span`] with the hash of the `PoV`.
#[inline(always)]
pub fn pov_span(pov: &PoV, span_name: &'static str) -> Span {
let mut span: Span = INSTANCE.read_recursive()
.span(|| { pov.hash() }, span_name).into();
span.add_pov(pov);
span
}
/// Creates a `Span` referring to the given hash. All spans created with [`hash_span`] with the
/// same hash (even from multiple different nodes) will be visible in the same view on Jaeger.
///
/// This span automatically has the `relay-parent` tag set.
#[inline(always)]
pub fn hash_span(hash: &Hash, span_name: &'static str) -> Span {
let mut span: Span = INSTANCE.read_recursive().span(|| { *hash }, span_name).into();
span.add_string_tag("relay-parent", &format!("{:?}", hash));
span
}
/// Creates a `Span` referring to the given descriptor, which should be unique.
#[inline(always)]
pub fn descriptor_span(descriptor: impl Encode, span_name: &'static str) -> Span {
INSTANCE.read_recursive().span(
|| { BlakeTwo256::hash_of(&descriptor) },
span_name,
).into()
}
/// Stateful convenience wrapper around [`mick_jaeger`].
pub enum Jaeger {
/// Launched and operational state.
@@ -443,9 +95,7 @@ impl Jaeger {
pub fn launch<S: SpawnNamed>(self, spawner: S) -> result::Result<(), JaegerError> {
let cfg = match self {
Self::Prep(cfg) => Ok(cfg),
Self::Launched{ .. } => {
return Err(JaegerError::AlreadyLaunched)
}
Self::Launched { .. } => return Err(JaegerError::AlreadyLaunched),
Self::None => Err(JaegerError::MissingConfiguration),
}?;
@@ -453,41 +103,39 @@ impl Jaeger {
log::info!("🐹 Collecting jaeger spans for {:?}", &jaeger_agent);
let (traces_in, mut traces_out) = mick_jaeger::init(mick_jaeger::Config {
service_name: format!("polkadot-{}", cfg.node_name),
});
let (traces_in, mut traces_out) =
mick_jaeger::init(mick_jaeger::Config { service_name: format!("polkadot-{}", cfg.node_name) });
// Spawn a background task that pulls span information and sends them on the network.
spawner.spawn("jaeger-collector", Box::pin(async move {
match async_std::net::UdpSocket::bind("0.0.0.0:0").await {
Ok(udp_socket) => loop {
let buf = traces_out.next().await;
// UDP sending errors happen only either if the API is misused or in case of missing privilege.
if let Err(e) = udp_socket.send_to(&buf, jaeger_agent).await {
log::debug!(target: "jaeger", "UDP send error: {}", e);
spawner.spawn(
"jaeger-collector",
Box::pin(async move {
match async_std::net::UdpSocket::bind("0.0.0.0:0").await {
Ok(udp_socket) => loop {
let buf = traces_out.next().await;
// UDP sending errors happen only either if the API is misused or in case of missing privilege.
if let Err(e) = udp_socket.send_to(&buf, jaeger_agent).await {
log::debug!(target: "jaeger", "UDP send error: {}", e);
}
},
Err(e) => {
log::warn!(target: "jaeger", "UDP socket open error: {}", e);
}
}
Err(e) => {
log::warn!(target: "jaeger", "UDP socket open error: {}", e);
}
}
}));
}),
);
*INSTANCE.write() = Self::Launched {
traces_in,
};
*INSTANCE.write() = Self::Launched { traces_in };
Ok(())
}
fn span<F>(&self, lazy_hash: F, span_name: &'static str) -> Option<mick_jaeger::Span>
pub(crate) fn span<F>(&self, lazy_hash: F, span_name: &'static str) -> Option<mick_jaeger::Span>
where
F: Fn() -> Hash,
F: Fn() -> TraceIdentifier,
{
if let Self::Launched { traces_in , .. } = self {
let hash = lazy_hash();
let mut buf = [0u8; 16];
buf.copy_from_slice(&hash.as_ref()[0..16]);
let trace_id = std::num::NonZeroU128::new(u128::from_be_bytes(buf))?;
if let Self::Launched { traces_in, .. } = self {
let ident = lazy_hash();
let trace_id = std::num::NonZeroU128::new(ident)?;
Some(traces_in.span(trace_id, span_name))
} else {
None